import argparse import ezdxf import json import os.path from dataclasses import dataclass, asdict, fields from dacite import from_dict from typing import List from datetime import datetime from openpyxl import Workbook import math from collections import defaultdict import configparser @dataclass class Point: x: float y: float @dataclass class Polyline: id: 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 class Error_Connection: name: str coords: Point @dataclass 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_existing_dists: List[str] errors_existing_sensors: List[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 dxf ..") doc = ezdxf.new('R2018', setup=True) create_cables(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 to original .dxf ..") doc = ezdxf.readfile(originaldxf) create_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 from original .dxf into new .dxf ..") quelle = ezdxf.readfile(originaldxf) ziel = ezdxf.new('R2018', setup=True) create_cables(plines, ziel) copy_layers_into_dxf_by_filter(quelle, ziel) ziel.saveas(out_path) print("done") def create_cables(plines, doc): msp = doc.modelspace() # # Beispiel für Text und Polyline # add_text(msp) # Layer mit Zeitstempel erstellen (z. B. "Layer_2025-05-07_14-30") timestamp = datetime.now().strftime("Kabel_%Y-%m-%d_%H-%M") if timestamp not in doc.layers: doc.layers.add(name=timestamp, color=7) dxfattribs={"layer": timestamp} for pl in plines.kabel: add_polyline(msp, pl, dxfattribs) 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 export_excel(json_file, out_path): # Hier für Excel Export print("creating excel file ..") plines = model_from_json(json_file) write_excel_from_json(plines, out_path) print("done") def write_excel_from_json(plines:Polylines, outpath:str): wb = Workbook() length_summary = defaultdict(int) #Worksheet 1 - Kabellängen ws1 = wb.active ws1.title = "Kabellängen" ws1.append(["Kabel-ID", "Länge aus Json (m)", "Gerundete Länge (m)"]) for pl in plines.kabel: length = pl.length /1000 # Umrechnung von mm in m rounded_len = math.ceil(length/10)*10 length_summary[rounded_len] +=1 ws1.append([pl.id, length, int(rounded_len)]) # Worksheet 2 - Zusammengefasste Längen ws2 = wb.create_sheet("Längenübersicht") ws2.append(["Gerundete Länge (m)", "Anzahl Kabel"]) for rlength in sorted(length_summary): ws2.append([int(rlength), length_summary[rlength]]) # Worksheet 3 - Nicht an Racks gekoppeltes Equipment ws3 = wb.create_sheet("Equipment nicht angebunden") ws3.append(["Typ", "Name", "x", "y"]) for error in plines.errors_sensors: ws3.append(["Sensor / Aktor", error.name, error.coords.x, error.coords.y]) for error in plines.errors_dists: ws3.append(["Unterverteiler", error.name, error.coords.x, error.coords.y]) # Worksheet 4 - Fehlgeschlagenes Routing ws4 = wb.create_sheet("Routingfehler") ws4.append(["Unterverteiler", "Sensor- / Aktorname", "Grund"]) 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 if uv in plines.errors_existing_dists: ws4.append([uv,"-", "Distributor not found in layout"]) continue for sensor in routing_error.sensoren: sensor_nicht_angebunden = sensor in nicht_angebunden if sensor_nicht_angebunden and uv_nicht_angebunden: grund = "Sensor und Unterverteiler nicht angebunden" elif sensor_nicht_angebunden: grund = "Sensor / Aktor nicht angebunden" elif uv_nicht_angebunden: grund = "Unterverteiler nicht angebunden" else: grund = "Fehlgeschlagenes Routing trotz Anbindung" ws4.append([uv, sensor, grund]) 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 collect_text_entities(msp): texts = [] for entity in msp.query("TEXT MTEXT"): try: text = entity.dxf.text pos = tuple(round(c, 4) for c in entity.dxf.insert) layer = entity.dxf.layer texts.append((text, pos, layer)) except Exception as e: print(f"Fehler beim Lesen eines Textobjekts: {e}") return texts def compare_texts(source_texts, target_texts): missing_in_target = [t for t in source_texts if t not in target_texts] extra_in_target = [t for t in target_texts if t not in source_texts] if not missing_in_target and not extra_in_target: print("✅ Alle TEXT/MTEXT-Objekte wurden korrekt kopiert.") else: if missing_in_target: print("\n❌ Fehlende Texte im Ziel:") for text in missing_in_target: print(f" - {text}") if extra_in_target: print("\n⚠️ Zusätzliche Texte im Ziel:") for text in extra_in_target: print(f" - {text}") def collect_block_references(doc): msp = doc.modelspace() blocks = [] for insert in msp.query("INSERT"): try: name = insert.dxf.name pos = tuple(round(c, 4) for c in insert.dxf.insert) layer = insert.dxf.layer blocks.append((name, pos, layer)) except Exception as e: print(f"Fehler beim Lesen eines INSERT-Objekts: {e}") return blocks def serialize_entity(e): """Wandelt eine Entity in ein vergleichbares Tupel um""" dxfattribs = {} for key in e.dxf.__dir__(): try: value = getattr(e.dxf, key) if callable(value) or key.startswith("_"): continue if isinstance(value, (list, tuple)): value = tuple(round(v, 4) for v in value) elif isinstance(value, float): value = round(value, 4) dxfattribs[key] = value except Exception: continue return (e.dxftype(), dxfattribs) def compare_blocks(source_doc, target_doc, used_block_names): for name in used_block_names: if name not in source_doc.blocks or name not in target_doc.blocks: print(f"❌ Block '{name}' fehlt in einer der Dateien.") continue source_block = source_doc.blocks.get(name) target_block = target_doc.blocks.get(name) source_entities = [serialize_entity(e) for e in source_block] target_entities = [serialize_entity(e) for e in target_block] missing_in_target = [e for e in source_entities if e not in target_entities] extra_in_target = [e for e in target_entities if e not in source_entities] if not missing_in_target and not extra_in_target: print(f"✅ Block '{name}' ist identisch.") else: print(f"\n🔍 Block '{name}' hat Unterschiede:") if missing_in_target: print(f" ❌ Fehlt im Ziel:") for e in missing_in_target: print(f" - {e}") if extra_in_target: print(f" ⚠️ Zusätzlich im Ziel:") for e in extra_in_target: print(f" - {e}") 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) ''' # 2. alle Blockdefinitionen kopieren die benötigt werden used_blocks = {ins.dxf.name for ins in msp_source.query("INSERT")} for name in used_blocks: # skip vorhandene if name in dxf_target.blocks: continue src_block = dxf_source.blocks.get(name) # neues BLOCK-Objekt anlegen mit Base-Point new_block = dxf_target.blocks.new(name=name, base_point=src_block.base_point) # **ALLE** Entities im Block kopieren for ent in src_block: new_block.add_entity(ent.copy()) # 3. Jetzt die INSERTs in den Modelspace kopieren for ins in msp_source.query("INSERT"): new_ins = ins.copy() msp_target.add_entity(new_ins) ''' # 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()) # texts_source = collect_text_entities(msp_source) # texts_target = collect_text_entities(msp_target) # compare_texts(texts_source, texts_target) # used_block_names = set(insert.dxf.name for insert in dxf_source.modelspace().query("INSERT")) # compare_blocks(dxf_source, dxf_target, used_block_names) 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') 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) 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")) 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(json_path, excel_path)