import asyncio import json import os import time import ezdxf import requests import yaml from flask import Flask, request, jsonify, send_from_directory, send_file from tqdm import tqdm from urllib3.exceptions import NewConnectionError import core from drawer import ShaftDrawer, TunDrawer, WindowDrawer, GateDrawer, FanMainDrawer, FanSystemDrawer, WindFlowDrawer from drawer.AirDuctDrawer import AirDuctDrawer from drawer.AirDuctExampleDrawer import AirDuctExampleDrawer from drawer.CADJson import CADJson, layer_divide from drawer.FanLocalDrawer import FanLocalDrawer from drawer.NodeGapDrawer import NodeGapDrawer from drawer.PlgTunDrawer import PlgTunDrawer from drawer.SealedDrawer import SealedDrawer from drawer.TunTextDrawer import TunTextDrawer from drawer.WindBridgeDrawer import WindBridgeDrawer from drawer.WindFlow3DDrawer import WindFlow3DDrawer app = Flask(__name__) cur_dir = os.getcwd() # url = 'http://192.168.183.216:8008/python/tunCAD' # url = 'data/1733121175.json' # url = 'http://localhost:8008/python/tunCAD' # url_2d = "http://192.168.183.216:8008/python/tunCAD" # url_3d = "http://192.168.183.216:8008/tunLt" url_2d = "http://localhost:8008/python/tunCAD" url_3d = "http://localhost:8008/tunLt" # url = 'http://localhost:8008/python/tunCAD' # url = 'data/cad.json' @app.route('/sysvent/download//', methods=['GET']) def download_dxf_file(model_id, type): # 指定文件的完整路径 try: path = None if type == '2d': with open('data/global-2d.yaml', 'r', encoding='utf-8') as file: config = yaml.safe_load(file) path = draw_system_vent(f'{url_2d}?modelid={model_id}', model_id,config) elif type == '3d': with open('data/global-3d.yaml', 'r', encoding='utf-8') as file: config = yaml.safe_load(file) path = draw_system_vent_3d(f'{url_3d}?modelid={model_id}', model_id,config) # 使用send_file发送文件，as_attachment=True表示以附件形式发送 # return send_file(path, as_attachment=True) return jsonify({"path": path}), 200 except NewConnectionError as e: # 捕获连接错误，并返回自定义的错误信息 error_message = "连接失败: 无法连接到服务器。错误信息: " + str(e) print(error_message) return jsonify({"error": error_message}), 500 except requests.exceptions.HTTPError as http_err: # 捕获其他HTTP错误 error_message = f"HTTP错误: {http_err}" print(error_message) return jsonify({"error": error_message}), 500 # except Exception as req_err: # # # 捕获所有requests库抛出的异常 # error_message = f"算法执行异常: {req_err}" # print(error_message) # return jsonify({"error": error_message}), 500 # def calculate_route_middle(wind_flow_unit): # path_start = wind_flow_unit[0]['x'], wind_flow_unit[0]['z'] # path_end = wind_flow_unit[-1]['x'], wind_flow_unit[-1]['z'] # path_middle = core.find_point_on_line(path_start, path_end, 1 / 2) # route = core.calculate_angle_with_x_axis(path_start, path_end) # return path_middle, route def msp_pre_process(path, model_id,config,model_type): cad_json = CADJson(path,config) default_template_file = "template/last/default_3d.dxf" if model_type== "3d": default_template_file = "template/last/default_3d.dxf" else: default_template_file = "template/last/default_2d.dxf" if os.path.exists(default_template_file): doc = ezdxf.readfile(default_template_file) else: doc = ezdxf.new('R2013') doc.styles.add("msyh", font="data/msyh.ttc") for layer in cad_json.json['layerMap']: doc.layers.new(name=f'图层{layer[0]}') msp = doc.modelspace() return msp,doc,cad_json def draw_system_vent_3d(path, model_id,config): msp,doc, cad_json = msp_pre_process(path, model_id,config,"3d") for gaps in tqdm(cad_json.gaps_list, desc=f' 【节点空隙绘制中】'): ngpd = NodeGapDrawer(msp, gaps, 1) ngpd.draw_gap() text_list = [] for tun in tqdm(cad_json.get_tun_plg_dict().values(), desc=' 【3d巷道绘制中】'): text = tun["name"] + " " + tun['fq'] plg_tun_agg = PlgTunDrawer(msp, tun['vec_b']['agg'], tun['vec_r']['agg'], tun['vec_l']['agg'], text, tun['layer_id'], tun['color'], tun['from']['agg'], tun['to']['agg'], tun['width']) plg_tun_agg.draw_plg_tun() # plg_tun_agg.draw_middle_line() if text not in text_list: text_list.append(text) plg_tun_agg.draw_tun_text() # plg_tun_div = PlgTunDrawer(msp, tun['vec_b']['div'], tun['vec_r']['div'], tun['vec_l']['div'], tun["name"], # tun['layer_id'], tun['color'], tun['from']['div'], tun['to']['div'], tun['width']) # plg_tun_div.draw_plg_tun() # plg_tun_div.draw_tun_text() for air_flow in tqdm(cad_json.air_flow_list_3d, desc=f' 【风流方向绘制中】'): agg_wind_flow = WindFlow3DDrawer(msp, air_flow['layer'], 3, air_flow['center']['agg'], air_flow['route']['agg'], air_flow['type']) # div_wind_flow = WindFlowDrawer(msp, air_flow['layer'], 3, air_flow['center']['div'], air_flow['route']['div'], # air_flow['type']) agg_wind_flow.draw_wind_flow() # div_wind_flow.draw_wind_flow() for window in tqdm(cad_json.window_list_3d, desc=f' 【风窗绘制中】'): agg_w = WindowDrawer(msp, window["width"], window["layer"], window["center"]["agg"], window["route"]['agg']) # div_w = WindowDrawer(msp, window["width"], window["layer"], window["center"]["div"], window["route"]['div']) agg_w.draw_window() # div_w.draw_window() for gate in tqdm(cad_json.gates_list_3d, desc=f' 【风门绘制中】'): agg_g = GateDrawer(msp, gate["layer"], gate["width"], gate['center']['agg'], gate["route"]['agg']) agg_g.draw_gate() for fan in tqdm(cad_json.fans_list_3d, desc=f' 【风扇绘制中】'): if 'fanmain' in str(fan['type']): agg_f = FanMainDrawer(msp, fan["layer"], fan["width"], fan['center']['agg'], fan["route"]['agg']) agg_f.draw_fan_main() if 'fanlocal' in str(fan['type']): fsd = FanSystemDrawer(msp,fan["layer"], fan["width"], fan['center']['agg'], fan["route"]['agg']) fsd.draw_fan_system() path = f'save/{str(time.time())}.dxf' doc.saveas(path) print(f'保存文件{os.path.abspath(path)}') return os.path.abspath(path) def draw_system_vent(path, model_id,config): msp,doc, cad_json = msp_pre_process(path, model_id,config,"2d") text_list = [] text_dict = {} for tun in tqdm(cad_json.tun_dict.values(), desc=' 【巷道绘制中】'): if tun['type'] == '1': shaft_center = tun['from'][0], tun['from'][1] sd = ShaftDrawer(msp, tun['layer_id'], tun['width'], shaft_center, 0) sd.draw_shaft_drawer() else: text = tun["name"] +" "+ tun['fq'] td_agg = TunDrawer(msp, tun['layer_id'], tun["vec12"]["agg"], tun["vec34"]["agg"], tun["from"]['agg'], tun["to"]['agg'], text, tun["width"], tun["head_vec_from"]['agg'], tun["head_vec_to"]['agg'] ) td_divided = TunDrawer(msp, tun['layer_id'], tun["vec12"]["div"], tun["vec34"]["div"], tun["from"]['div'], tun["to"]['div'], text, tun["width"], tun["head_vec_from"]['div'], tun["head_vec_to"]['div'] ) td_agg.draw_tun() td_divided.draw_tun() # if text not in text_dict.keys(): # text_dict[text]= [] # else: # text_dict[text].append(tun) # # if text not in text_list: # text_list.append(text) # td_agg.draw_tun_text() # td_divided.draw_tun_text() for tun_text in tqdm(cad_json.tun_text_dict.values(), desc=' 【巷道文字绘制中】'): ttd_agg = TunTextDrawer(msp, tun_text['layer'], tun_text['tun_line']['agg'], tun_text['text'],5) ttd_div = TunTextDrawer(msp,tun_text['layer'], tun_text['tun_line']['div'],tun_text['text'],5) ttd_agg.dra_text() ttd_div.dra_text() for sealed in tqdm(cad_json.sealed_list, desc=f' 【密闭绘制中】'): agg_w = SealedDrawer(msp, sealed["width"], sealed["layer"], sealed["center"]["agg"], sealed["route"]['agg']) div_w = SealedDrawer(msp, sealed["width"], sealed["layer"], sealed["center"]["div"], sealed["route"]['div']) agg_w.drawSealed() div_w.drawSealed() for window in tqdm(cad_json.window_list, desc=f' 【风窗绘制中】'): agg_w = WindowDrawer(msp, window["width"], window["layer"], window["center"]["agg"], window["route"]['agg']) div_w = WindowDrawer(msp, window["width"], window["layer"], window["center"]["div"], window["route"]['div']) agg_w.draw_window() div_w.draw_window() for ad_list in tqdm(cad_json.ari_duct_dict.values(), desc=f' 【风筒绘制中】'): for i in range(len(ad_list)): ad = ad_list[i] agg_ad= AirDuctDrawer(msp,ad['layer'],ad['from']['agg'],ad['to']['agg'],ad['width']) div_ad= AirDuctDrawer(msp,ad['layer'],ad['from']['div'],ad['to']['div'],ad['width']) agg_ad.draw_air_duct_list() div_ad.draw_air_duct_list() if i+1