import argparse import configparser import json import os import re import sys from pathlib import Path import ezdxf from ezdxf.addons import iterdxf from shapely.geometry import Point from ezdxf.lldxf.const import DXFStructureError """ Dieses Programm: - liest die dxf Datei und holt sich von den Layern der dxf Datei die Positionen + der Motoren, Sensoren und Aktoren + der Unterverteiler + der Polylinien der Kabelpritschen - erzeugt daraus eine .json Datei im Work Ordner """ class ErrorCollector: """Sammelt alle Fehler und Warnungen während der Verarbeitung.""" def __init__(self): self.errors = {} self.warnings = {} def add_errors(self, error_dict): """Fügt Fehler aus einem Dictionary hinzu.""" for key, value in error_dict.items(): if key in self.errors: if isinstance(self.errors[key], list) and isinstance(value, list): self.errors[key].extend(value) else: self.errors[key] = [self.errors[key], value] else: self.errors[key] = value def add_warnings(self, warning_dict): """Fügt Warnungen aus einem Dictionary hinzu.""" for key, value in warning_dict.items(): if key in self.warnings: if isinstance(self.warnings[key], list) and isinstance(value, list): self.warnings[key].extend(value) else: self.warnings[key] = [self.warnings[key], value] else: self.warnings[key] = value def has_errors_or_warnings(self): """Prüft ob Fehler oder Warnungen vorhanden sind.""" return bool(self.errors) or bool(self.warnings) def has_errors(self): """Prüft ob Fehlervorhanden sind.""" return bool(self.errors) def get_all_issues(self): """Gibt alle Fehler und Warnungen als Dictionary zurück.""" result = {} if self.errors: result["errors"] = self.errors if self.warnings: result["warnings"] = self.warnings return result def write_errorfile(self, work_dir: Path, filename: str): """Schreibt Fehlerdatei nur wenn Fehler oder Warnungen vorhanden sind.""" if self.has_errors(): print("Fehler gefunden. Schreibe Error-Datei.") all_issues = self.get_all_issues() write_results(to_json(all_issues), work_dir, filename) else: print("Keine Fehler oder Warnungen gefunden. Keine Error-Datei geschrieben.") def write_results(jsn_results: str, out_dir: Path, filename: str) -> None: """Write results to a JSON file.""" print("writing results to a json file ...") outfile = os.path.join(out_dir, filename) with open(outfile, 'w', encoding='utf-8') as fh: fh.write(jsn_results) print("done") def merge_two_dicts(x: dict, y: dict) -> dict: z = x.copy() z.update(y) return z def get_type_of_name_cfg(name: str) -> str: prefix = name[:2] if config_BMK.has_option("Routing-Include", prefix): return "Sensor" elif config_BMK.has_option("Routing-Ignore", prefix): return "Schaltschrankelement" else: return "unknown" def matches_cabinet_pattern(name: str) -> [bool, str]: """Check if the given name matches any Cabinet-Pattern from BMK.cfg""" if not config_BMK.has_section("Cabinet-Pattern"): return (False, "") patterns = [pattern for pattern, _ in config_BMK.items("Cabinet-Pattern")] for pattern in patterns: m = re.search(pattern, name) if m: res = m.group(1) return (True, res) return (False, "") def matches_tunnel_pattern(name: str) -> bool: """Check if the given name matches any Tunnel-Pattern from BMK.cfg""" if not config_BMK.has_section("Tunnel-Pattern"): return False patterns = [pattern for pattern, _ in config_BMK.items("Tunnel-Pattern")] for pattern in patterns: m = re.search(pattern, name) if m: return True return False def get_attributes_of_insert(d_insert: dict, d_pos: dict) -> tuple[dict, str, str]: """ Diese Funktion schaut nach den aktuell definierten Attributen in den Blöcken Bei den Sensoren in den alten Layouts gibt zwei immer übereinanderliegende Blöcke mit den Attributen: - A, B (z.B. MA0062), C, ARTINR (z.B. 790902001), BESCHR (E-Teile für SEW Motor ASE1..), MENGE, POSITION, ... - IO (z.B. MA0062), ID , VERW (z.B. CV-M0062_0,75), BEZEICHNUNG (Motor MA0062), KENNZEICHNUNG (z.B.=A01+UH01-KF1DQ04), ... Die Adresse für das Routing kommt aus "KENNZEICHNUNG", während die Sivas Nummer aus "ARTINR" geholt werden muss Einmal wird B zur ID, beim anderen IO Erdungssymbole erhalten die ID aus dem Eintrag unter "NAME" Hier in Zukunft weniger Abfragen: IO und B und Reale_Position wird überflüssig wenn jeder Sensor nur noch ein Block mit allen Attributen! """ id_ = "" ld = d_insert typ = 'unknown' # die neueren Bläcke heissen nicht IO, sondern haben einen Namen if "IO" in d_insert: pos = d_pos["IO"] typ = get_type_of_name_cfg(d_insert["IO"]) # neuer Block: Enthält alles was nötig is if "ARTINR" in d_insert and "SPS" in d_insert: id_ = d_insert["IO"]+"@"+d_insert["SPS"] ld["pos"] = (pos[0], pos[1]) if "REALE_POSITION" in d_insert and d_insert["REALE_POSITION"] == 'x': pos = d_pos["REALE_POSITION"] else: id_ = d_insert["IO"] # Sensoren werden später gemerged mit den anderen Blöcken des Rahmens mit A,B,C, usw if "SPS" in d_insert and typ != "Sensor": id_ = id_+"@"+d_insert["SPS"] if "REALE_POSITION" in d_insert and d_insert["REALE_POSITION"] == 'x': pos = d_pos["REALE_POSITION"] # Hoehe und Breite von "x" addieren, um Mittelpunkt zu finden breite_marker = config.getfloat("GetPos-Geom-Sensor", "Breite") hoehe_marker = config.getfloat("GetPos-Geom-Sensor", "Hoehe") midx = pos[0] + breite_marker * 0.5 midy = pos[1] + hoehe_marker * 0.5 ld["pos"] = (round(midx, 1), round(midy, 1)) else: ld["pos"] = (pos[0], pos[1]) # die neueren Blöcke heissen nicht IO, sondern haben einen Namen if "NAME" in d_insert: typ = get_type_of_name_cfg(d_insert["NAME"]) id_ = d_insert["NAME"] pos = d_pos["NAME"] ld["pos"] = (pos[0], pos[1]) if "B" in d_insert and "IO" not in d_insert: attr_text = d_insert["B"] typ = get_type_of_name_cfg(attr_text) id_ = attr_text #Position aufzeichnen und bei Bedarf später mit REAL_POS überschreiben pos = d_pos["B"] ld["pos"] = (pos[0], pos[1]) return ld, id_, typ class CompareBuffer: """Ein Puffer um alle als Doppelt zugewiesenen Blöcke zwischenzulagern.""" def __init__(self) -> None: self._wartepuffer = dict() def add_block(self, id_: str, buffer: dict) -> None: """Adds one block under the buffer list under this id.""" if id_ not in self._wartepuffer: self._wartepuffer[id_] = list() self._wartepuffer[id_].append(buffer) def get_blocks(self, id_: str) -> list: if id_ in self._wartepuffer: return self._wartepuffer[id_] else: return [] def get_block_ids(self) -> list[str]: return list(self._wartepuffer.keys()) def set_block(self, id_: str, buffer: list) -> None: if id_ in self._wartepuffer: del self._wartepuffer[id_] self._wartepuffer[id_] = buffer def dict_compare(self, d1: dict, d2: dict) -> bool: str1 = json.dumps(d1) str2 = json.dumps(d2) return str1 == str2 def remove_block(self, id_: str, to_remove: dict) -> None: """Removes this block from the list under the given id.""" buffers = self.get_blocks(id_) l = list() for b in buffers: if self.dict_compare(b, to_remove): pass else: l.append(b) self.set_block(id_, l) def positions_are_close(self, dict1: dict, dict2: dict, border: float) -> bool: pos1 = None pos2 = None if "pos" in dict1: pos1 = dict1["pos"] if "pos" in dict2: pos2 = dict2["pos"] if not (pos1 and pos2): return False p1 = Point(pos1[0], pos1[1]) p2 = Point(pos2[0], pos2[1] ) dist = p1.distance(p2) if dist < border: return True return False def get_all_sps_blocks(self, id_: str) -> list: """Gives back all blocks with SPS inside to the given id.""" buffers = self.get_blocks(id_) l = list() for b in buffers: if "SPS" in b: l.append(b) return l def get_non_sps_blocks(self, id_: str) -> list: """Gives back all blocks without SPS inside to the given id.""" buffers = self.get_blocks(id_) l = list() for b in buffers: if "SPS" not in b: l.append(b) return l def extract_input_positions(all_inserts, all_positions, error_collector: ErrorCollector = None) -> tuple[dict, dict]: """ Extracts and organizes input positions from an iterable of inserts. This function processes a list of insert objects (e.g., from a DXF file), classifies them by type (Sensor, Kabel, Schaltschrankelement, or unknown), and organizes them into dictionaries keyed by their IDs. For sensors, it handles the special case where sensor information may be split across multiple blocks (e.g., IO and A,B,C blocks) and merges them if necessary. It also collects information about missing attributes and duplicate IDs for further error handling. Args: all_inserts: List of insert objects to process. all_positions: List of position objects corresponding to inserts. error_collector: ErrorCollector instance to collect errors and warnings. Returns: A tuple containing: - all_sensors: dict of sensor IDs to sensor attribute dicts - all_schaltschrank: dict of Schaltschrankelement IDs to their attribute dicts """ all_sensors = dict() all_cables = dict() all_schaltschrank = dict() all_unknowns = list() wp = CompareBuffer() for insert, pos in zip(all_inserts, all_positions): ld, id_, typ = get_attributes_of_insert(insert, pos) if typ == "Sensor": # wenn NAME enthalten ist, z.B. bei Erdungslayouts, dann ist das ein eindeutiger Bezeichner if "NAME" in ld: all_sensors[id_] = ld # neuer Block der alle nötigen Infos enthält elif "IO" in ld and "ARTINR" in ld : all_sensors[id_] = ld # alle anderen Sachen werden dann aus mehreren Rahmen zusammen gesammelt else: wp.add_block(id_, ld) elif typ == "Kabel": all_cables[id_] = ld elif typ == "Schaltschrankelement": all_schaltschrank[id_] = ld else: all_unknowns.append(ld) # spezialbehandlung der Sensoren, da diese in IO und A,B,C Blöcke geteilt sind # Die Funktion sucht die übereinanderliegenen Elemente und baut ein Dict daraus allocate_blocks_together(all_sensors, wp, error_collector) set_single_frames_with_unique_sps(all_schaltschrank, wp) # die noch übrigen Blöcke melden get_errors_double_and_attributes(wp, error_collector) return all_sensors, all_schaltschrank def get_errors_double_and_attributes(wp: CompareBuffer, error_collector: ErrorCollector = None): """ Analyze the CompareBuffer for blocks that could not be merged or assigned. This function inspects the CompareBuffer for: - IDs with only a single associated block, which likely indicates missing or incomplete attributes. - IDs with multiple associated blocks, which may indicate duplicate or ambiguous entries. Args: wp: CompareBuffer to analyze error_collector: ErrorCollector instance to collect errors and warnings """ missing_attribs = dict() double_ids = dict() for id_ in wp.get_block_ids(): blocks = wp.get_blocks(id_) # einzelne Blöcke deren Zuordnung nicht gelungen ist, fehlen Angaben in den Attributen, z.B. SPS, B, u.ä. if len(blocks) == 1: given_keys = str(blocks[0].keys()) missing_attribs[id_] = f"Nur ein Block und/oder fehlende Attribute: {given_keys}" wp.remove_block(id_, blocks[0]) # alle anderen Angaben sind mindestens zwei oder mehr Blöcke mit derselben Id else: for block in blocks: if id_ not in double_ids: double_ids[id_] = [] double_ids[id_].append(block['pos']) # Füge Fehler und Warnungen zum ErrorCollector hinzu, falls vorhanden if error_collector: if double_ids: error_collector.add_errors({"double_ids": double_ids}) if missing_attribs: error_collector.add_warnings({"missing_attributes": missing_attribs}) def set_single_frames_with_unique_sps(all_sensors: dict, wp: CompareBuffer): all_sensors_ids = wp.get_block_ids() for id_ in all_sensors_ids: blks_sps = wp.get_all_sps_blocks(id_) for block_with_sps in blks_sps: # schaue ob es keinen Konflikt zu bestehenden Angaben if "IO" in block_with_sps and "SPS" in block_with_sps: sps_praefix = block_with_sps["SPS"] new_id = f"{id_}@{sps_praefix}" all_sensors[new_id] = block_with_sps wp.remove_block(id_, block_with_sps) def allocate_blocks_together(all_sensors: dict, wp: CompareBuffer, error_collector: ErrorCollector = None) -> None: """ Merge sensor blocks with the same ID that are split across multiple DXF blocks. This function iterates over all sensor IDs stored in the CompareBuffer. For each ID, it looks for blocks with and without an "SPS" prefix. If two such blocks have positions that are close to each other (within a specified tolerance), they are considered to represent the same physical sensor. The function then merges their information into a single dictionary, creates a new unique sensor ID by appending the SPS prefix, and adds the merged sensor to the all_sensors dictionary. The merged blocks are then removed from the buffer. Args: all_sensors (dict): Dictionary to store merged sensor information, keyed by unique sensor IDs. wp (CompareBuffer): Buffer containing blocks that could not be directly assigned, grouped by ID. """ #geht alle gemerkten Sensoren durch die gleich heissen. # Falls ein SPS Präfix angegeben wird, wird es zur Id hinzugefügt und als neuer Name gemerkt all_sensors_ids = wp.get_block_ids() for id_ in all_sensors_ids: blks_sps = wp.get_all_sps_blocks(id_) blks_other = wp.get_non_sps_blocks(id_) for block_with_sps in blks_sps: for block_without_sps in blks_other: # Vergleiche alle Blöcke mit SPS und denen ohne auf die gleiche Position if wp.positions_are_close(block_with_sps, block_without_sps, 1000): new_id = create_new_id(id_, block_with_sps, block_without_sps, error_collector) # hier das Präfix davor all_sensors[new_id] = merge_two_dicts(block_without_sps, block_with_sps) #Kombiniert alle infos aus dxf und "pos" wp.remove_block(id_, block_with_sps) wp.remove_block(id_, block_without_sps) def create_new_id(id_: str, dict1: dict, dict2: dict, error_collector: ErrorCollector = None) -> str: sps_praefix = None if "SPS" in dict1: sps_praefix = dict1["SPS"] if "SPS" in dict2: sps_praefix = dict2["SPS"] if not sps_praefix: if error_collector: error_collector.add_errors({"missing_sps_prefix": {id_: "SPS Präfix fehlt für Block-ID"}}) return f"{id_}@UNKNOWN" # Fallback für fehlenden SPS-Präfix return f"{id_}@{sps_praefix}" def attribs_to_dicts(insert_iterable) -> tuple[list, list]: """ Wandelt eine Iterable von INSERT-Objekten in zwei Listen um: - all_inserts: Liste von Dictionaries mit Attribut-Tags und deren Textwerten - all_positions: Liste von Dictionaries mit Attribut-Tags und deren (x, y, z)-Positionen (jeweils gerundet auf eine Nachkommastelle) Jeder Eintrag in den Listen entspricht einem INSERT-Block mit Attributen. Blöcke ohne Attribute werden übersprungen. """ all_inserts = list() all_positions = list() for insert in insert_iterable: if insert.dxftype() != 'INSERT': continue itemdata = dict() positions = dict() typ = 'unknown' for attrib in insert.attribs: if len(insert.attribs) == 0: continue # Überspringe Blöcke ohne Attribute attr_tag = attrib.dxf.tag attr_text = attrib.dxf.text pos = attrib.dxf.insert itemdata[attr_tag] = attr_text positions[attr_tag] = (round(pos.x, 1), round(pos.y, 1), round(pos.z, 1)) if len(itemdata) > 0: all_inserts.append(itemdata) all_positions.append(positions) return all_inserts, all_positions def create_mappings(positions: dict) -> tuple[dict, dict]: unterverteiler_pfad = "" dnamen = dict() # s sensor2unterverteiler = dict() warnings = dict() for sensorname,v in positions.items(): if "KENNZEICHNUNG" not in v: warnings[sensorname] = "keine KENNZEICHNUNG vorhanden" continue unterverteiler_pfad = v["KENNZEICHNUNG"] #print(unterverteiler_pfad) # Pfad zur Karte splitten. Dieser hat z.B. den Inhalt "=AH01+UH02-KF1FDI7" matches = re.findall(r'[^\-+=]+', unterverteiler_pfad.lstrip('=')) if matches: if len(matches) == 3: anlage = matches[0] verteiler = matches[1] karte = matches[2] else: warnings[sensorname] = f"Ungültiger Pfad in Kennzeichnung: {unterverteiler_pfad}: +, - oder = fehlen an entsprechender Stelle" else: warnings[sensorname] = f"Ungültiger Pfad in Kennzeichnung: {unterverteiler_pfad}" continue if verteiler not in dnamen: dnamen[verteiler] = True sensor2unterverteiler[sensorname] = verteiler # jetzt zu jedem Unterverteiler die zugehörigen Sensoren merken uv2sensor = dict() for sensorname,verteiler in sensor2unterverteiler.items(): if verteiler not in uv2sensor: uv2sensor[verteiler] = list() uv2sensor[verteiler].append(sensorname) return (uv2sensor, warnings) def get_subdistributor_position_of_symbol(d_insert: dict, d_pos: dict) -> tuple[dict, str]: """ Diese Funktion schaut nach den aktuell definierten Attributen in allen Unterverteiler Blöcken Sie müssen auch UH heissen """ id_ = "" ld = d_insert # die neueren Blöcke haben einen Namen if "NAME" in d_insert: id_ = d_insert["NAME"] pos = d_pos["NAME"] ld["pos"] = (pos[0], pos[1]) if "REALE_POSITION" in d_insert and d_insert["REALE_POSITION"] == 'x': pos = d_pos["REALE_POSITION"] # Hoehe und Breite von "x" addieren, um Mittelpunkt zu finden breite_marker = config.getfloat("GetPos-Geom-Sensor", "Breite") hoehe_marker = config.getfloat("GetPos-Geom-Sensor", "Hoehe") midx = pos[0] + breite_marker * 0.5 midy = pos[1] + hoehe_marker * 0.5 ld["pos"] = (round(midx, 1), round(midy, 1)) return ld, id_ return None, None def get_tunnel_position_of_symbol(d_insert: dict, d_pos: dict) -> tuple[dict, str]: """ Diese Funktion schaut nach den aktuell definierten Attributen in allen Tunnel Blöcken Tunnel haben einen Namen der den Tunnel-Mustern aus BMK.cfg entspricht """ id_ = "" ld = d_insert # Tunnel haben einen Namen if "NAME" in d_insert: id_ = d_insert["NAME"] pos = d_pos["NAME"] ld["pos"] = (pos[0], pos[1]) # Länge des Tunnels aus dem LAENGE Attribut extrahieren if "LAENGE" in d_insert: ld["laenge"] = d_insert["LAENGE"] if "REALE_POSITION" in d_insert and d_insert["REALE_POSITION"] == 'x': pos = d_pos["REALE_POSITION"] # Hoehe und Breite von "x" addieren, um Mittelpunkt zu finden breite_marker = config.getfloat("GetPos-Geom-Sensor", "Breite") hoehe_marker = config.getfloat("GetPos-Geom-Sensor", "Hoehe") midx = pos[0] + breite_marker * 0.5 midy = pos[1] + hoehe_marker * 0.5 ld["pos"] = (round(midx, 1), round(midy, 1)) return ld, id_ return None, None def get_subdistributor_positions_from_symbols(all_inserts:list, all_positions:list, dist2sensors: dict) -> dict: """Ermittelt Unterverteiler-Positionen aus allen Symbolen """ ret = {} all_distributors = dist2sensors.keys() for insert, pos in zip(all_inserts, all_positions): if "NAME" not in insert: # Unterverteiler haben immer einen eindeutigen Namen continue ld, id_ = get_subdistributor_position_of_symbol(insert, pos) # Check if id_ matches Cabinet-Pattern from BMK.cfg (matches, res) = matches_cabinet_pattern(id_) if not matches: continue if res in dist2sensors: ret[res] = ld["pos"] return ret def get_tunnel_positions_from_symbols(all_inserts: list, all_positions: list) -> dict: """Ermittelt Tunnel-Positionen aus allen Symbolen Args: all_inserts: Liste von Dictionaries mit Attribut-Tags und deren Textwerten all_positions: Liste von Dictionaries mit Attribut-Tags und deren (x, y, z)-Positionen Returns: Dictionary mit Tunnelnamen als Keys und Listen von Positionen als Values Format: {'Tunnel1': [(x1, y1), (x2, y2)], ..., 'length': {'Tunnel1': 'laenge_wert', ...}} """ ret = {} tunnel_length = {} default_length = "5" # Default-Länge in Metern for insert, pos in zip(all_inserts, all_positions): if "NAME" not in insert: # Tunnel haben immer einen eindeutigen Namen continue ld, id_ = get_tunnel_position_of_symbol(insert, pos) if not ld or not id_: continue # Check if id_ matches Tunnel-Pattern from BMK.cfg if not matches_tunnel_pattern(id_): continue # Sammle alle Positionen für den gleichen Tunnelnamen if id_ not in ret: ret[id_] = [] ret[id_].append(ld["pos"]) # Sammle Längeninformation falls vorhanden, sonst Default verwenden if "laenge" in ld and ld["laenge"]: tunnel_length[id_] = ld["laenge"] else: print(f"WARNUNG: Tunnel '{id_}' hat keine LAENGE-Angabe. Verwende Default-Länge: {default_length}m") tunnel_length[id_] = default_length # Füge Längeninformation hinzu, falls Tunnel gefunden wurden if len(tunnel_length) > 0: ret['length'] = tunnel_length return ret def get_subdistributor_positions_from_entities(entities, dist2sensors: dict) -> dict: """Ermittelt Unterverteiler-Positionen aus einer beliebigen Entity-Iterable. Erwartet eine Iterable von DXF-Entities (z. B. aus `msp.query("MTEXT")` oder `iterdxf.modelspace(dxf_path)`) und filtert nach den in der Config erlaubten Layern. Es werden beide bisher verwendeten Suchmuster in der MTEXT-Zeile unterstützt ("-" und "+"). """ ret = {} all_distributors = dist2sensors.keys() allowed_layers = {layer for (layer, _) in config.items('GetPos-Layer_Distributors')} for entity in entities: if entity.dxftype() != "MTEXT": continue layer = entity.dxf.layer if layer not in allowed_layers: continue text = entity.dxf.text insert_point = entity.dxf.insert for distname in all_distributors: if f"-{distname}" in text or f"+{distname}" in text: ret[distname] = (round(insert_point[0], 1), round(insert_point[1], 1)) return ret def get_tunnel_positions_from_entities(entities) -> dict: """Ermittelt Tunnel-Ein/Ausgangs-Positionen aus einer beliebigen Entity-Iterable. Erwartet eine Iterable von DXF-Entities (z. B. aus `msp.query("MTEXT")` oder `iterdxf.modelspace(dxf_path)`) und filtert nach den in der Config erlaubten Layern. Erkennt Tunnel anhand des Musters "TUNNEL-" und sammelt pro Tunnelname die gefundenen Positionen sowie die Länge. """ all_tunnels = dict() tunnel_length = dict() allowed_layers = {layer for (layer, _) in config.items('GetPos-Layer_Tunnel')} for entity in entities: if entity.dxftype() != "MTEXT": continue if entity.dxf.layer not in allowed_layers: continue txt = entity.dxf.text insert = entity.dxf.insert pattern = r"(TUNNEL\d+)-(\d+)" match = re.search(pattern, txt) if not match: continue tunnelname = match.group(1) laenge = match.group(2) pos = (round(insert[0], 1), round(insert[1], 1)) if tunnelname not in all_tunnels: all_tunnels[tunnelname] = [] all_tunnels[tunnelname].append(pos) tunnel_length[tunnelname] = laenge if len(tunnel_length.keys()) > 0: all_tunnels['length'] = tunnel_length return all_tunnels # helper function def print_line(e) -> None: print(f"LINE on layer: {e.dxf.layer}\n") print(f"points: {repr(e.dxf)}\n") def print_polyline(e) -> None: print(f"POLYLINE on layer: {e.dxf.layer}\n") for x, y, start_width, end_width, bulge in e.get_points(): print(f" Punkt: ({x}, {y}), Startbreite: ({start_width}, Endbreite: {end_width})") if e.is_closed: print("Diese Polyline ist geschlossen.") def get_rack_positions(msp) -> dict: """Hole alle Positionen aller Kabelpritschen und nummeriere Racks.""" ret = dict() rack_counter = 1 all_layers = list(config.items('GetPos-Layer_Racks')) for layer, _ in all_layers: lw_query = f'LWPOLYLINE[layer=="{layer}"]' for entity in msp.query(lw_query): rack_key = f"Rack_{rack_counter}" handle_lwpolyline(entity, rack_key, ret) rack_counter += 1 pl_query = f'POLYLINE[layer=="{layer}"]' for entity in msp.query(pl_query): rack_key = f"Rack_{rack_counter}" handle_polyline(entity, rack_key, ret) rack_counter += 1 return ret def get_rack_positions_iter(dxf_path) -> dict: """Hole alle Positionen aller Kabelpritschen (Racks) mithilfe von iterdxf.""" ret = dict() rack_counter = 1 all_layers = config.items('GetPos-Layer_Racks') for entity in iterdxf.modelspace(dxf_path): layer = entity.dxf.layer if not any(layer == cfg_layer for cfg_layer, _ in all_layers): continue rack_key = f"Rack_{rack_counter}" if entity.dxftype() == "LWPOLYLINE": handle_lwpolyline(entity, rack_key, ret) elif entity.dxftype() == "POLYLINE": handle_polyline(entity, rack_key, ret) else: continue rack_counter += 1 return ret def handle_lwpolyline(entity, rack_key: str, ret: dict) -> None: """Verarbeitet eine 2D LWPOLYLINE mit globalem Z-Wert (elevation).""" z = getattr(entity.dxf, "elevation", 0.0) ret[rack_key] = [] for point in entity.vertices(): x, y, *_ = point ret[rack_key].append([round(x, 1), round(y, 1), round(z, 1)]) def handle_polyline(entity, rack_key: str, ret: dict) -> None: """Verarbeitet eine klassische POLYLINE – inklusive 3D-Polylinien mit individuellen Z-Werten.""" ret[rack_key] = [] for vertex in entity.vertices: x = vertex.dxf.location.x y = vertex.dxf.location.y z = vertex.dxf.location.z ret[rack_key].append([round(x, 1), round(y, 1), round(z, 1)]) def scan(dxf_source) -> dict: layer_names_inside = list(dxf_source.layers.names()) alle_block_defs = set(dxf_source.blocks.block_names()) used_block_names = set(insert.dxf.name for insert in dxf_source.modelspace().query("INSERT")) ret = dict() ret['all_layers'] = layer_names_inside ret['used_blocks'] = used_block_names ret['all_blocks'] = alle_block_defs return ret def to_json(d: object, pretty: bool = True) -> str: return json.dumps(d, indent=2 if pretty else None, ensure_ascii=False, default=str) #ensure_ascii false für darstellung von "ue" def get_dxf_file(filepath: Path): """Hole das dxf file.""" try: print("reading file ..", end='') doc = ezdxf.filemanagement.readfile(filepath) print("done") except IOError: print("Not a DXF file or a generic I/O error.") sys.exit(1) except DXFStructureError: print("Invalid or corrupted DXF file.") sys.exit(2) return doc def check_file_in_work(work_dir: Path, filename: Path) -> tuple[Path, bool]: fexists = True if not filename.exists(): mypath = work_dir.joinpath(filename) ex = mypath.exists() if not mypath.exists(): fexists = False else: mypath = filename return mypath, fexists def check_existance(res_mappings: dict, res_dist: dict, res_pos: dict, res_tunnel: dict) -> dict: ret = dict() for dname in res_mappings.keys(): if dname not in res_dist: if "missing_distributors" not in ret: ret["missing_distributors"] = list() ret["missing_distributors"].append(dname) for sname, lofsensors in res_mappings.items(): for s in lofsensors: if s not in res_pos: if "missing_sensors" not in ret: ret["missing_sensors"] = list() ret['missing_sensors'].append(s) for tname in res_tunnel.keys(): if tname == "length": # dict enthält auch die Länge aller Tunnels in einem Key continue if len(res_tunnel[tname]) < 2 : if "missing_tunnel" not in ret: ret["missing_tunnel"] = list() ret["missing_tunnel"].append(tname) if len(res_tunnel[tname]) > 2 : if "overdefined_tunnel" not in ret: ret["overdefined_tunnel"] = list() ret["overdefined_tunnel"].append(tname) return ret def dxf_is_binary(dxf_path: Path) -> bool: with open(dxf_path, 'rb') as f: header = f.read(22) return b'AutoCAD Binary DXF' in header def validate_configs() -> None: errors = [] print("\nValidating given configs: Checking for inconsistency.") if config_BMK.has_section("Routing-Include") and config_BMK.has_section("Cable-Mapping"): for prefix in config_BMK.options("Routing-Include"): if prefix not in config_BMK["Cable-Mapping"]: errors.append(f"No Cable-Mapping for Prefix '{prefix}' within 'Routing-Include'") if config_BMK.has_section("Cable-Mapping"): for mapping_key, value in config_BMK.items("Cable-Mapping"): sections = [s.strip() for s in value.split(",")] for section in sections: if not config_cables.has_section(section): errors.append(f"Cable-Section '{section}' from Cable-Mapping ({mapping_key}) missing in kabel.cfg") if config_BMK.has_section("Length-Adjustments"): for prefix, value in config_BMK.items("Length-Adjustments"): try: f = float(value) if f < 0: errors.append(f"Negative Value in Length-Adjustments for {prefix}: {value}") except ValueError: errors.append(f"Invalid Value in Length-Adjustments for {prefix}: {value}") if errors: print("Inconsistencies found:") for e in errors: print(f"- {e}") print("\ncontinuing with routing process") else: print("No inconsistencies found. Continuing with routing process.") def check_environment_var(env_str: str) -> Path: out_path = os.environ.get(env_str) if out_path: return Path(out_path) else: print(f"Umgebungsvariable {env_str} ist nicht gesetzt oder leer.") exit() if __name__ == '__main__': parser = argparse.ArgumentParser(description='fetches the x/y positions from a dxf file', prog='getpositions') parser.add_argument('-f', '--filename', action='store', required=True, default="ST_6300_Steuerungstestlayout1_neueBloecke.dwg", help='which file should be fetched', metavar='myfile.dxf') parser.add_argument('-s', '--sensors', action='store_true', help='fetch all position of sensors, motors, actors and subdistributors') parser.add_argument('-r', '--rack', action='store_true', help='fetch all positions of all cable racks') parser.add_argument('-w', '--write', action='store', help='write results into a json file') parser.add_argument('-c', '--console', action='store_true', help='print results to output') parser.add_argument('-e', '--errors', action='store', help='write an error file in case of double defined items in layout') parser.add_argument('-n', '--scan', action='store_true', help='print all layer of racs, distributes and equiment not empty') args = parser.parse_args() out_dir = check_environment_var('PROJECT_DATA') work_dir = check_environment_var('PROJECT_WORK') config_dir = check_environment_var("PROJECT_CFG") filename = Path(args.filename) if not filename.suffix == ".dxf": print("only available for .dxf files") exit() (dxf_path, dexists) = check_file_in_work(work_dir, filename) if dexists == False: print("no such file ") parser.print_help() exit() if dxf_is_binary(dxf_path): # Wenn dxf eine binary ist, dann komplett parsen und modelspace anlegen print("Given .dxf-file is binary dxf. Proceeding to read file. Watch RAM-usage.") doc = get_dxf_file(dxf_path) msp = doc.modelspace() use_iter = False else: print("Given .dxf-file is ASCII-dxf. Proceeding to use iterative functions. Process may take longer.") use_iter = True res_sens = dict() res_cables = dict() res_dist = dict() res_rac = dict() res_mappings = dict() if args.sensors or args.dists or args.rack: # Allgemeine Config Laden config = configparser.ConfigParser(allow_no_value=True, delimiters=("=")) config.optionxform = lambda optionstr: optionstr # preserve case for letters config.read(os.path.join(config_dir, "allgemein.cfg")) # Betriebsmittelkennzeichnungs-Config laden config_BMK = configparser.ConfigParser(allow_no_value=True, delimiters=("=")) config_BMK.optionxform = lambda optionstr: optionstr # preserve case for letters config_BMK.read(os.path.join(config_dir, "BMK.cfg")) # Kabel-Config laden config_cables = configparser.ConfigParser(allow_no_value=True, delimiters=("=")) config_cables.optionxform = lambda optionstr: optionstr config_cables.read(os.path.join(config_dir, "kabel.cfg")) validate_configs() output_results = dict() # ErrorCollector für alle Fehler und Warnungen initialisieren error_collector = ErrorCollector() if args.sensors: # Sensoren auslesen aus den Symbolen if use_iter: all_inserts, all_positions = attribs_to_dicts(iterdxf.modelspace(dxf_path)) else: all_inserts, all_positions = attribs_to_dicts(msp) res_sens, res_schaltschrank_elemente = extract_input_positions(all_inserts, all_positions, error_collector) output_results['sensors'] = res_sens output_results['schaltschrank_elemente'] = res_schaltschrank_elemente #output_results['cables'] = res_cables if args.console: print(to_json(res_sens)) # Mapping zu Sensoren auslesen (res_mappings, warnings) = create_mappings(res_sens) output_results['mappings'] = res_mappings # Mapping-Warnungen zum ErrorCollector hinzufügen if warnings: error_collector.add_warnings({"mapping_warnings": warnings}) if args.console: print(to_json(res_mappings)) # Distributoren auslesen (generisch über Entities) if use_iter: entities = iterdxf.modelspace(dxf_path) else: entities = msp.query('MTEXT') # die Infos aus den Texten (alter Stil) res_dist = get_subdistributor_positions_from_entities(entities, res_mappings) # die Infos aus den Blöcken symbol_dists = get_subdistributor_positions_from_symbols(all_inserts, all_positions, res_mappings) res_dist = merge_two_dicts(res_dist, symbol_dists) output_results['distributors'] = res_dist if args.console: print(to_json(res_dist)) # Tunnel auslesen (generisch über Entities) if use_iter: t_entities = iterdxf.modelspace(dxf_path) else: t_entities = msp.query('MTEXT') # die Infos aus den Texten (alter Stil) res_tunnel = get_tunnel_positions_from_entities(t_entities) # die Infos aus den Blöcken (neuer Stil) symbol_tunnels = get_tunnel_positions_from_symbols(all_inserts, all_positions) res_tunnel = merge_two_dicts(res_tunnel, symbol_tunnels) output_results['tunnels'] = res_tunnel if args.console: print(to_json(res_tunnel)) if args.rack: if use_iter: res_rac = get_rack_positions_iter(dxf_path) else: res_rac = get_rack_positions(msp) output_results['racks'] = res_rac if args.console: print(to_json(res_rac)) if args.write: basename = os.path.splitext(args.write)[0] res_not_found = check_existance(res_mappings, res_dist, res_sens, res_tunnel) # Alle weiteren Fehler zum ErrorCollector hinzufügen error_collector.add_errors(res_not_found) # ErrorCollector schreibt Datei nur wenn Fehler/Warnungen vorhanden if args.errors: error_collector.write_errorfile(work_dir, args.errors) # Für Kompatibilität mit Ausgabe-JSON: Fehler und Warnungen aus ErrorCollector holen all_issues = error_collector.get_all_issues() if "errors" in all_issues: output_results["not_found"] = all_issues["errors"].get("not_found", {}) if "double_ids" in all_issues["errors"]: output_results["double_ids"] = all_issues["errors"]["double_ids"] if "missing_attributes" in all_issues.get("warnings", {}): output_results["not_found"]["missing_attributes"] = all_issues["warnings"]["missing_attributes"] write_results(to_json(output_results), work_dir, f"{basename}.json") else: parser.print_help()