import argparse import ezdxf import json import os.path from dataclasses import dataclass, asdict, fields from dacite import from_dict from typing import List, Dict from datetime import datetime from openpyxl import Workbook import math from collections import defaultdict import configparser import updateconfignames as uc @dataclass class Point: x: float y: float @dataclass class Polyline: id: str s_artinr: str coords: List[Point] length: float def to_tuple(self): ret = list() for p in self.coords: ret.append( (p.x, p.y) ) return ret @dataclass # Fehlgeschlagene Anbindung von einem Sensor / Dist zu einem Rack class Error_Connection: name: str coords: Point @dataclass # Felgeschlagene Verbindung von einem Dist zu Sensor(en) aus beliebigem Grund class Error_Routing: unterverteiler: str sensoren: List[str] @dataclass class Polylines: kabel: List[Polyline] errors_routing: List[Error_Routing] errors_sensors: List[Error_Connection] errors_dists: List[Error_Connection] errors_dists_not_in_layout: List[str] errors_sensors_not_in_layout: List[str] errors_missing_attributes: Dict[str, str] def add_polyline(msp, points:Polyline, dxf_attribs): pts = points.to_tuple() pline = msp.add_lwpolyline(points=pts, dxfattribs=dxf_attribs) pline.rgb = (255, 128, 0) def new_dxf(plines, out_path): """ creates a new dxf file with a polyline inside which is created by the given json file """ print("creating new .dxf ..") doc = ezdxf.new('R2018', setup=True) draw_cables(plines, doc) draw_sensors(plines, doc) draw_subdists(plines, doc) doc.saveas(out_path) print("done") def modify_original_dxf(plines, originaldxf): """ adds new layer to original .dxf-file that contains cables """ print("adding cables into original .dxf ..") doc = ezdxf.readfile(originaldxf) draw_cables(plines, doc) doc.saveas(out_path) print("done") def copy_layers_into_new(originaldxf, outpath, plines): """ creates a new dxf file with a racks, sensors, subdists from original file including cable paths """ print("copying layers (Racks, Subdistributors, ...) from original .dxf into new .dxf ..") quelle = ezdxf.readfile(originaldxf) ziel = ezdxf.new('R2018', setup=True) draw_cables(plines, ziel) draw_sensors(plines, ziel) copy_layers_into_dxf_by_filter(quelle, ziel) ziel.saveas(out_path) print("done") def draw_cables(plines, doc): msp = doc.modelspace() timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M") cable_layer = f"cables_{timestamp}" # Kabel-Layer anlegen if cable_layer not in doc.layers: doc.layers.add(name=cable_layer, color=7) dxfattribs_cable={"layer": cable_layer} # Kabel zeichnen for pl in plines.kabel: # Polyline für Kabel zeichnen add_polyline(msp, pl, dxfattribs_cable) def find_close_key(pos2sensors, x, y, tolerance=10): # !!! Toleranz nicht in Config !!! ''' Funktion überprüft ob Sensoren nahezu identisch an der gleichen Stelle liegen und legt sie in diesem fall aufeinander Wird benötigt, um zusammengehörige Sensoren gestaffelt auf dxf zu zeichen ''' for (px, py) in pos2sensors: if abs(px - x) <= tolerance and abs(py - y) <= tolerance: return (px, py) return None def draw_sensors(plines, doc): msp = doc.modelspace() timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M") sensor_layer = f"sensors_{timestamp}" # Sensor-Layer erzeugen if sensor_layer not in doc.layers: doc.layers.add(name=sensor_layer, color=5) dxfattribs_sensors={"layer": sensor_layer, "height": 100} # Sensoren nach Endpunkten gruppieren -> mehrfacheinträge gestaffelt zeichnen pos2sensors = defaultdict(list) for pl in plines.kabel: pt2 = pl.coords[-1] #Endpunkt des Kabels = Sensor Position pos_key = find_close_key(pos2sensors, pt2.x, pt2.y) if pos_key: pos2sensors[pos_key].append(pl) else: pos2sensors[(pt2.x, pt2.y)].append(pl) # Sensor Blöcke zeichnen for (x,y), pls in pos2sensors.items(): for i, pl in enumerate(pls): sensor_name = pl.id.split('-')[-1] pt1 = pl.coords[-2] pt2 = pl.coords[-1] dx = pt2.x - pt1.x dy = pt2.y - pt1.y text = msp.add_text(sensor_name, dxfattribs=dxfattribs_sensors) # Offsets für Beschriftungen offsetx = 0 offsety = 0 # Kabel Horizontal if abs(dx) > abs(dy): n = len(pls) center_offset = i - (n - 1) / 2 # Falls mehrere Sensoren -> Verschiebung so, dass Kabel immer in Mitte ankommt offsety = -80 + center_offset * 110 # Wert -80 über Try-and-Error zur Mitte der Beschriftung angepasst if dx > 0: halign = 0 # LEFT offsetx = 50 # Wert 50 durch Try-and-Error sodass etwas Abstand zu Kabelpritsche else: halign = 2 # RIGHT offsetx = -50 # Wert 50 durch Try-and-Error sodass etwas Abstand zu Kabelpritsche valign = 1 # BOTTOM # Kabel vertikal else: if dy > 0: valign = 0 # BASELINE offsety = 50 + i * 110 # nach oben anfügen else: valign = 3 # TOP offsety = -50 - i * 110 # nach unten anfügen halign = 1 # CENTER # Alignments setzen text.dxf.halign = halign text.dxf.valign = valign text.set_placement((pt2.x + offsetx, pt2.y + offsety)) def draw_subdists(plines, doc): msp = doc.modelspace() timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M") subdist_layer = f"subdists_{timestamp}" # Sensor-Layer erzeugen if subdist_layer not in doc.layers: doc.layers.add(name=subdist_layer, color=3) dxfattribs_subdists={"layer": subdist_layer, "height": 100} subdist_positions = set() for pl in plines.kabel: pt1 = pl.coords[0] # Startposition = UV-Position pos = (pt1.x, pt1.y) if pos in subdist_positions: continue subdist_positions.add(pos) subdist_name = pl.id.split('-')[0] pt2 = pl.coords[1] dx = pt2.x - pt1.x dy = pt2.y - pt1.y # Offsets für Beschriftungen offsetx = 0 offsety = 0 if abs(dx) > abs(dy): # Horizontal offsety = -80 # Wert -80 über Try-and-Error zur Mitte der Beschriftung angepasst if dx < 0: halign = 0 # LEFT offsetx = 50 # Wert 50 durch Try-and-Error sodass etwas Abstand zu Kabelpritsche else: halign = 2 # RIGHT offsetx = -50 # Wert 50 durch Try-and-Error sodass etwas Abstand zu Kabelpritsche valign = 1 # BOTTOM else: # Vertikal if dy < 0: valign = 0 # BASELINE offsety = 50 else: valign = 3 # TOP offsety = -50 halign = 1 # CENTER # Text platzieren text = msp.add_text(subdist_name, dxfattribs=dxfattribs_subdists) text.dxf.halign = halign text.dxf.valign = valign text.set_placement((pt1.x + offsetx, pt1.y + offsety)) def model_from_json(json_file): with open(json_file, encoding='utf-8') as fh: data = json.load(fh) plines = from_dict( data_class=Polylines, data=data ) return plines def parse_sensors_from_json(positions_json): with open(positions_json, encoding='utf-8') as fh: data = json.load(fh) sensors = {} for name, data in data.get("sensors", {}).items(): sensor = Sensors( name=name, artinr=data.get("ARTINR", ""), pos=data.get("pos", [0.0, 0.0]), ) sensors[name] = sensor return sensors def export_excel(plines, sens2cable, out_path): # Hier für Excel Export print("creating excel file with cable information ..") write_excel_from_json(plines, sens2cable, out_path) print("done") def get_all_artnrs(plines: Polylines, sens2cable: dict) -> set: all_artnrs = set() for pl in plines.kabel: if pl.s_artinr: all_artnrs.add(pl.s_artinr) for cablelist in sens2cable.values(): for artnr in cablelist: if isinstance(artnr, str) and artnr.isdigit(): all_artnrs.add(artnr) return all_artnrs def mark_missings(all_artnrs): if "Sivasnummern" not in bezeichner_cfg: bezeichner_cfg["Sivasnummern"] = {} if "Missing" not in bezeichner_cfg: bezeichner_cfg["Missing"] = {} for artnr in all_artnrs: if artnr not in bezeichner_cfg["Sivasnummern"]: bezeichner_cfg["Missing"][artnr] = "" def write_excel_from_json(plines:Polylines, sens2cable: dict, outpath:str, with_bom = True): wb1 = Workbook() # Dicts für Anzahl bzw kummulierte Länge count_summary = defaultdict(int) length_summary = defaultdict(float) sensor_summary = defaultdict(int) uv_sensor_count = defaultdict(lambda: defaultdict(int)) uv_artnr_count = defaultdict(lambda: defaultdict(int)) uv_artnr_length = defaultdict(lambda: defaultdict(float)) seen_sensors = set() uv_sensor_mapping = defaultdict(dict) #Worksheet 1 - Kabellängen nach Kabel-ID ws1 = wb1.active ws1.title = "Length by ID" ws1.append(["Cable-ID", "True Length (m)", "Cable-ArtNr", "Cable-Name (short)"]) ws1.column_dimensions['A'].width = 18 ws1.column_dimensions['B'].width = 15 ws1.column_dimensions['C'].width = 15 ws1.column_dimensions['D'].width = 25 for pl in plines.kabel: length = round(pl.length /1000 , 1) # Umrechnung von mm in m uv_name = pl.id.split("-")[0] sensor_name = pl.id.split("-")[-1] # Globaler Sensor-Count (für erstes Sheet) if sensor_name not in seen_sensors: seen_sensors.add(sensor_name) sensor_summary[pl.s_artinr] += 1 # UV-spezifische Sensor-Mapping-Zwischenspeicherung if sensor_name not in uv_sensor_mapping[uv_name]: uv_sensor_mapping[uv_name][sensor_name] = pl.s_artinr for artnr in sens2cable[pl.id]: cable_name = "" if artnr.isdigit(): cable_name = bezeichner_cfg["Sivasnummern"][artnr] ws1.append([pl.id, length, artnr, cable_name]) if "MA" in pl.id: length_summary[artnr] += math.ceil(length) # Aufrunden von z.b. 10,3 auf 11 m (volle Meter!) uv_artnr_length[uv_name][artnr] += math.ceil(length) else: count_summary[artnr] += 1 uv_artnr_count[uv_name][artnr] += 1 # UV-spezifische Sensorzählung (robust, 1x je Sensor pro UV) for uv_name, sensor_dict in uv_sensor_mapping.items(): for sensor_name, artnr in sensor_dict.items(): uv_sensor_count[uv_name][artnr] += 1 # Worksheet 2 - Kabelnummern und Stückzahlen ws2 = wb1.create_sheet("Cables SIVAS") ws2.append(["Cable-ArtNr", "Amount (pcs)", "Cumm. Length (m)"]) ws2.column_dimensions['A'].width = 20 ws2.column_dimensions['B'].width = 12 ws2.column_dimensions['C'].width = 15 all_artnrs = set(count_summary.keys()) | set(length_summary.keys()) for artnr in sorted(all_artnrs): count = count_summary.get(artnr, "") cumm_length = length_summary.get(artnr, "") ws2.append([artnr, count, cumm_length]) # Abfage ob Fehler Worsheets ausgegeben werden if len(plines.errors_sensors) > 0 or len(plines.errors_dists) > 0: # Worksheet 3 - Nicht an Racks gekoppeltes Equipment ws3 = wb1.create_sheet("ERR-Equipment-Connection") ws3.append(["Type", "ID", "x", "y"]) ws3.column_dimensions['A'].width = 20 # nicht angebundene Sensoren for error in plines.errors_sensors: ws3.append(["Sensor / Actuator", error.name, error.coords.x, error.coords.y]) # nicht angebundene Distributoren for error in plines.errors_dists: ws3.append(["Subistributor", error.name, error.coords.x, error.coords.y]) if len(plines.errors_routing) > 0: # Worksheet 4 - Fehlgeschlagenes Routing ws4 = wb1.create_sheet("ERR-Routing") ws4.append(["Subdistributor", "Sensor / Actuator", "Details"]) ws4.column_dimensions['C'].width = 35 nicht_angebunden = set(e.name for e in plines.errors_sensors + plines.errors_dists) for routing_error in plines.errors_routing: uv = routing_error.unterverteiler uv_nicht_angebunden = uv in nicht_angebunden # Wenn Distributor nicht in Layout gefunden, dann nur diesen Fehler ausgeben. Alle Verbindungen hier fehlgeschlagen if uv in plines.errors_dists_not_in_layout: ws4.append([uv,"-", "Distributor not found in given layout."]) continue for sensor in routing_error.sensoren: sensor_nicht_angebunden = sensor in nicht_angebunden if sensor_nicht_angebunden and uv_nicht_angebunden: grund = "Subdistributor and sensor / actuator not connected to racks" elif sensor_nicht_angebunden: grund = "Sensor / actuator not connected to racks" elif uv_nicht_angebunden: grund = "Subdistributor not connected to racks" else: grund = "Failed routing (not caused by missing connection)" ws4.append([uv, sensor, grund]) if len(plines.errors_missing_attributes) > 0: # Worksheet 5 - Fehlende Attribute -> kein Routing ws5 = wb1.create_sheet("ERR-Attributes") ws5.append(["ID", "Error Detail"]) ws5.column_dimensions['B'].width = 35 for sname, err_msg in plines.errors_missing_attributes.items(): ws5.append([sname, err_msg]) wb1.save(outpath) print("Cable-Summary exported to Excel file") if with_bom: wb2 = Workbook() ws1 = wb2.active ws1.title = "BOM" ws1.append(["Art.-Number", "Name (SIVAS)", "Amount (pcs)", "Length (m)"]) # Sensoren separat auflisten for artnr in sorted(sensor_summary.keys()): name = bezeichner_cfg["Sivasnummern"].get(artnr, "") count = sensor_summary[artnr] if artnr == "": name = "Keine Artikelnummer vergeben. Layout prüfen." elif name == "": name = f"Kein Eintrag zu Art.-Nr: {artnr} in bezeichner.cfg." ws1.append([ artnr, name, count, "" # Sensoren haben keine Länge ]) # Kabel separat auflisten for artnr in sorted(set(count_summary.keys()) | set(length_summary.keys())): name = bezeichner_cfg["Sivasnummern"].get(artnr, "") count = count_summary.get(artnr, "") length = length_summary.get(artnr, "") if artnr == "": name = "Keine Artikelnummer vergeben. Layout prüfen." elif name == "": name = f"Kein Eintrag zu Art.-Nr: {artnr} in bezeichner.cfg." ws1.append([ artnr, name, count, length ]) # Neues Sheet für UV-spezifische Übersicht ws_uv = wb2.create_sheet("BOM by UV") ws_uv.append(["UV", "Art.-Number", "Name (SIVAS)", "Amount (pcs)", "Length (m)"]) all_uvs = set(uv_sensor_count.keys()) | set(uv_artnr_count.keys()) | set(uv_artnr_length.keys()) for uv in sorted(all_uvs): all_artnrs = set() all_artnrs.update(uv_sensor_count.get(uv, {}).keys()) all_artnrs.update(uv_artnr_count.get(uv, {}).keys()) all_artnrs.update(uv_artnr_length.get(uv, {}).keys()) for artnr in sorted(all_artnrs): sensor_amount = uv_sensor_count.get(uv, {}).get(artnr, 0) cable_count = uv_artnr_count.get(uv, {}).get(artnr, 0) cable_length = uv_artnr_length.get(uv, {}).get(artnr, 0) name = bezeichner_cfg["Sivasnummern"].get(artnr, "") # Sensoren separat eintragen (Anzahl) if sensor_amount > 0: ws_uv.append([ uv, artnr, name, sensor_amount, "" ]) # a) Nur Anzahl, wenn kein "MA" im Namen if cable_count > 0 and cable_length == 0: ws_uv.append([ uv, artnr, name, cable_count, "" ]) # b) Nur Länge, wenn Kabel mit "MA" (Meterware) if cable_length > 0: ws_uv.append([ uv, artnr, name, "", cable_length ]) bom_path = outpath.replace("_cables.xlsx", "_BOM.xlsx") wb2.save(bom_path) print(f"BOM saved as an excel file") def check_file_in_work(work_dir, filename): fexists = True if not os.path.exists(filename): mypath = os.path.join(work_dir, filename) if not os.path.exists(mypath): fexists = False else: mypath = filename return (mypath, fexists) def copy_layers_into_dxf_by_filter(dxf_source: ezdxf.document.Drawing, dxf_target:ezdxf.document.Drawing): msp_source = dxf_source.modelspace() msp_target = dxf_target.modelspace() subdist_layers = set(config.options('GetPos-Layer_Distributors')) rack_layers = set(config.options('GetPos-Layer_Racks')) equipment_layers = set(config.options('GetPos-Layer_Equipment')) tunnel_layers = set(config.options('GetPos-Layer_Tunnel')) layernames = set() layernames.update(subdist_layers) layernames.update(rack_layers) layernames.update(equipment_layers) layernames.update(tunnel_layers) # # welche Texte existieren # for layername in layernames: # selectstr = f'MTEXT[layer=="{layername}"]' # for text in msp_source.query(selectstr): # inhalt = text.dxf.text # position = text.dxf.insert # print(f"Text: '{inhalt}' an Position: {position} auf Layer: {layername}") # text_entity = text.copy() # msp_target.add_entity(text_entity) layer_names_inside = dxf_source.layers alle_block_defs = set(dxf_source.blocks.block_names()) verwendete = {insert.dxf.name for insert in msp_source.query("INSERT")} # 1. Textstyles kopieren for style in dxf_source.styles: if style.dxf.name not in dxf_target.styles: dxf_target.styles.new(name=style.dxf.name) # 4. Filter-Layernamen bestimmen for layername in layernames: if layername not in dxf_source.layers: continue # Falls der Layer noch nicht im Zieldokument existiert, neu anlegen if layername not in dxf_target.layers: quelle_layer = dxf_source.layers.get(layername) dxf_target.layers.add( name=layername, color=quelle_layer.color, linetype=quelle_layer.dxf.linetype, lineweight=quelle_layer.dxf.lineweight ) # Alle Entities auf diesem Layer kopieren entities = msp_source.query(f"*[layer=='{layername}']") for entity in entities: msp_target.add_entity(entity.copy()) def get_cable_artnr(section, length): """ Sucht in der angegebenen Config-Section die passende Kabel-Artikelnr. für die gegebene Länge. Übergibt bei Treffer True, artnr bei Fehler: False, errmsg """ # Existiert über BMK vergebene Kabel-Kennzeichnung in kabel.cfg? if section not in cable_cfg: return False, f"Keine Kabelkennzeichnung '{section}' in kabel.cfg" entries = cable_cfg[section] length_keys = sorted([float(k) for k in entries.keys()]) for l in length_keys: if length <= l: return True, entries[str(l)] return False, f"Kabel länger als max. Kabellänge in Sektion '{section}'" def map_sensor_to_cable_cfg(plines): sens2cable = defaultdict(list) mapping = config_BMK["Cable-Mapping"] for pl in plines.kabel: sensor_name = '-'.join(pl.id.split('-')[1:]) cable_length = round(pl.length/1000, 1) sensor_artinr = pl.s_artinr name_prefix = sensor_name[:2] # Suche nach Key in der BMK-Config key_with_artnr = f"{name_prefix}-{sensor_artinr}" # Spezialfälle über "Key mit Artikelnummer" abgleichen if key_with_artnr in mapping: section_list = mapping[key_with_artnr] elif name_prefix in mapping: # Standardzuweisung section_list = mapping[name_prefix] else: sens2cable[pl.id].append("Kein Kabeltyp zugewiesen (BMK.cfg)") # Liste aus evtl. mehreren Sektionen erzeugen sections = [s.strip() for s in section_list.split(",")] # Evtl. Kabelkürzung durchführen, falls Kabelschwanz vorhanden if config_BMK.has_section("Length-Adjustments") and config_BMK.has_option("Length-Adjustments", name_prefix): length_reduction = float(config_BMK.get("Length-Adjustments", name_prefix)) cable_length = max(0.0, cable_length-length_reduction) # Kabel-Atikelnummer innerhalb der Sektion der kabel.cfg bestimmen for section in sections: sucess, result = get_cable_artnr(section, cable_length) sens2cable[pl.id].append(result) return sens2cable if __name__ == '__main__': parser = argparse.ArgumentParser(description='draws a dxf file with the given cable coordinates', prog='drawdxf') parser.add_argument('-f', '--filename', action='store', required=True, help='this json file contains all cables and its coordinates which should be drawn. Saved with an unique timestamp', metavar='myfile.json') parser.add_argument('-d', '--dxf', action='store', help='this dxf drawing will be copied and the new layer with the cables will be added. Original file must be added with --origin', metavar='myfile.dxf') parser.add_argument('-c', '--copy_layer', action='store', help='copy layers of racks, sensors, distributors into a new .dxf-file. File also contains cable paths. Original file must be added with --origin', metavar='original.dxf') parser.add_argument('-n', '--new', action='store', help='create a new dxf file only with cables in it. Name is basename and a timestamp') parser.add_argument('-x', '--excel', action='store', help='create a xlsx file with cables data', metavar='allCables.xls') parser.add_argument('-o', '--origin', action='store', help='name of original .dxf file used by -d and -a', metavar='original.dxf') parser.add_argument('-l', '--local', action='store_true', help='using only local data for naming of article numbers. If not set: fetching names from SIVAS.') args = parser.parse_args() config_dir = os.environ.get("PROJECT_CFG") work_dir = os.fspath(os.environ.get('PROJECT_WORK')) json_file = args.filename (json_path, jexists) = check_file_in_work(work_dir, json_file) if not jexists: print(f"file {json_file} does not exist") parser.print_help() exit() plines = model_from_json(json_path) # Allgemeine Config laden config = configparser.ConfigParser(allow_no_value=True, delimiters=("=")) config.optionxform = lambda option: option # preserve case for letters config.read(os.path.join(config_dir, "allgemein.cfg")) # Config für Kabel-Artikelnummern laden cable_cfg = configparser.ConfigParser() cable_cfg.optionxform = str #Keys case-sensitive with open(os.path.join(config_dir, "kabel.cfg"), encoding="utf-8") as f: cable_cfg.read_file(f) # Betriebsmittelkennzeichnungs-Config laden config_BMK = configparser.ConfigParser(allow_no_value=True, delimiters=("=")) config_BMK.optionxform = lambda option: option # preserve case for letters config_BMK.read(os.path.join(config_dir, "BMK.cfg")) bezeichner_cfg = configparser.ConfigParser() bezeichner_cfg.optionxform = str #Keys case-sensitive with open(os.path.join(config_dir, "bezeichner.cfg"), encoding="utf-8") as f: bezeichner_cfg.read_file(f) dxf_file = args.dxf if args.dxf or args.copy_layer: if not args.origin: parser.print_help() exit() else: (origin_path, dexists) = check_file_in_work(work_dir, args.origin) if args.dxf: (dxf_path, dexists) = check_file_in_work(work_dir, dxf_file) if not dexists: print(f"file {dxf_file} does not exist") parser.print_help() exit() out_path = dxf_path res_pos = new_dxf(plines, dxf_path) if args.copy_layer: out_path = os.path.join(work_dir, args.copy_layer) res_pos = new_dxf(plines, out_path) copy_layers_into_new(origin_path, out_path, plines) if args.new: # erzeuge dxf Datei nur mit Kabeln out_path = os.path.join(work_dir, args.new) res_pos = new_dxf(plines, out_path) if args.excel: # 1. Kabelzuordnung sens2cable = map_sensor_to_cable_cfg(plines) # 2. verwendete Artikelnummer (Sensoren + Kabel) all_artnrs = get_all_artnrs(plines, sens2cable) # 3. Fehlende Bezeichner in bezeichner.cfg markieren mark_missings(all_artnrs) # 4. Bezeichner ggf. aus SIVAS abrufen if args.local == False: config_path=os.path.join(config_dir, "bezeichner.cfg") uc.update_bezeichner(bezeichner_cfg, config_path) # 5. Excel schreiben excel_path = os.path.join(work_dir, args.excel) write_excel_from_json(plines, sens2cable, excel_path)