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 from updateconfignames import get_sivas_teilestamm, read_bezeichner_from_csv @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, out_path): # Hier für Excel Export print("creating excel file with cable information ..") write_excel_from_json(plines, out_path) print("done") def write_excel_from_json(plines:Polylines, outpath:str): wb = Workbook() sens2cable = map_sensor_to_cable_cfg(plines) # Dicts für Anzahl bzw kummulierte Länge count_summary = defaultdict(int) length_summary = defaultdict(float) #Worksheet 1 - Kabellängen nach Kabel-ID ws1 = wb.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 = pl.length /1000 # Umrechnung von mm in m for artnr in sens2cable[pl.id]: cable_name = "" if artnr.isdigit(): cable_name = cable_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!) else: count_summary[artnr] += 1 # Worksheet 2 - Kabelnummern und Stückzahlen ws2 = wb.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 = wb.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 = wb.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 = wb.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]) wb.save(outpath) 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(plines): sens2cable = defaultdict(list) for pl in plines.kabel: sensor_name = pl.id.split('-')[-1] cable_length = pl.length/1000 sensor_artinr = pl.s_artinr name_prefix = sensor_name[0:2] sections = [] #Default if name_prefix == "MA": sections = ["MA"] # Sektion für Motor-Kabel elif name_prefix == "MB": sections = ["WD_Q"] # Sektion für Ventil-Kabel elif name_prefix == "QM": sections = ["WD_Q"] # Sektion für Ventil-Kabel elif sensor_name.startswith("B"): # Sensor Name beginnt mit B -> Sensor jeglicher Art if name_prefix == "BX": # Wenn "BX", dann Scanner sections = ["WF_B", "WD_I"] # Sektion für Scanner Patch-Kabel und Sektion für Standard-Sensorkabel zur Spannungsversorgung cable_length = max(0.0, cable_length -4.0) # 4 Meter abziehen, da bereits 5m Kabelschwanz an Scanner dran elif sensor_artinr == 829422026: sections = ["WD_I-829422026"] elif sensor_artinr == 720002003: sections = ["WD_I-720002003"] else: sections = ["WD_I"] if not sections: sens2cable[pl.id].append ("Kein Kabeltyp zugewiesen") continue for section in sections: cable_artnr = get_cable_artnr(section, cable_length) if cable_artnr is None: sens2cable[pl.id].append(f"Kabel länger als max. Kabellänge") else: sens2cable[pl.id].append(cable_artnr) return sens2cable def map_sensor_to_cable_cfg(plines): sens2cable = defaultdict(list) mapping = config_BMK["Cable-Mapping"] for pl in plines.kabel: sensor_name = 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) if not sucess: sens2cable[pl.id].append(result) else: 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")) names_cfg = configparser.ConfigParser() names_cfg.optionxform = str #Keys case-sensitive with open(os.path.join(config_dir, "bezeichner.cfg"), encoding="utf-8") as f: names_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: excel_path = os.path.join(work_dir, args.excel) export_excel(plines, excel_path)