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 """ 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 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(insert_iterable) -> tuple[dict, dict, 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: insert_iterable: An iterable of insert objects to process. Returns: A tuple containing: - all_sensors: dict of sensor IDs to sensor attribute dicts - all_schaltschrank: dict of Schaltschrankelement IDs to their attribute dicts - double_ids: dict of IDs with multiple associated blocks (potential duplicates) - missing_attribs: dict of IDs with missing or incomplete attributes """ all_sensors = dict() all_cables = dict() all_schaltschrank = dict() all_unknowns = list() wp = CompareBuffer() all_inserts, all_positions = attribs_to_dicts(insert_iterable) 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) set_single_frames_with_unique_sps(all_schaltschrank, wp) # die noch übrigen Blöcke melden missing_attribs, double_ids = get_errors_double_and_attributes(wp) return all_sensors, all_schaltschrank, double_ids, missing_attribs def get_errors_double_and_attributes(wp: CompareBuffer) -> tuple[dict, dict]: """ 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. Returns: missing_attribs: dict mapping IDs to a message about missing or incomplete attributes. double_ids: dict mapping IDs to a list of positions for blocks with duplicate IDs. """ 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']) return missing_attribs, double_ids 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) -> 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) # 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) -> str: sps_praefix = None if "SPS" in dict1: sps_praefix = dict1["SPS"] if "SPS" in dict2: sps_praefix = dict2["SPS"] if not sps_praefix: raise Exception 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 get_input_positions(msp) -> tuple[dict, dict, dict, dict]: return extract_input_positions(msp.query('INSERT')) def get_input_positions_iter(dxf_path) -> tuple[dict, dict, dict, dict]: return extract_input_positions(iterdxf.modelspace(dxf_path)) 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_positions(msp, dist2sensors: dict) -> dict: """Hole alle Positionen der Unterverteiler !!UV-Positionen bereits 'Mitte-Mitte'!!""" ret = dict() all_distributors = dist2sensors.keys() all_layers = config.items('GetPos-Layer_Distributors') for (layer, v) in all_layers: for distname in all_distributors: selectstr = f'MTEXT[layer=="{layer}"]' for text in msp.query(selectstr): match = re.search("-" + distname, text.dxf.text) if match: ret[distname] = (round(text.dxf.insert[0], 1), round(text.dxf.insert[1], 1)) return ret def get_subdistributor_positions_iter(dxf_path, dist2sensors: dict) -> dict: """Hole alle Positionen der Unterverteiler aus MTEXT-Objekten mithilfe von iterdxf.""" ret = {} all_distributors = dist2sensors.keys() all_layers = config.items('GetPos-Layer_Distributors') for entity in iterdxf.modelspace(dxf_path): if entity.dxftype() != "MTEXT": continue entity_text = entity.dxf.text entity_layer = entity.dxf.layer insert_point = entity.dxf.insert for (layer_name, _) in all_layers: if entity_layer != layer_name: continue for distname in all_distributors: if f"+{distname}" in entity_text: ret[distname] = (round(insert_point[0], 1), round(insert_point[1], 1)) return ret def get_tunnel_positions(msp) -> dict: """Hole alle Positionen aller Tunnel Ein und Ausgänge.""" all_tunnels = dict() tunnel_length = dict() all_layers = config.items('GetPos-Layer_Tunnel') for (layer, v) in all_layers: selectstr = f'MTEXT[layer=="{layer}"]' for text in msp.query(selectstr): txt = text.dxf.text pattern = r"(TUNNEL\d+)-(\d+)" match = re.search(pattern, txt) if match: pos = (round(text.dxf.insert[0], 1), round(text.dxf.insert[1], 1)) tunnelname = match.group(1) laenge = match.group(2) tunnel_length[tunnelname] = laenge if tunnelname not in all_tunnels: all_tunnels[tunnelname] = list() all_tunnels[tunnelname].append(pos) else: all_tunnels[tunnelname].append(pos) if len(all_tunnels.keys()) > 0: all_tunnels['length'] = tunnel_length return all_tunnels def get_tunnel_positions_iter(dxf_path) -> dict: """Hole alle Positionen aller Tunnel Ein- und Ausgänge mithilfe von iterdxf.""" all_tunnels = dict() tunnel_length = dict() all_layers = config.items('GetPos-Layer_Tunnel') for entity in iterdxf.modelspace(dxf_path): if entity.dxftype() != "MTEXT": continue txt = entity.dxf.text layer = entity.dxf.layer insert = entity.dxf.insert for (layer_name, _) in all_layers: if layer != layer_name: continue pattern = r"(TUNNEL\d+)-(\d+)" match = re.search(pattern, txt) if match: 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() ret["missing_distributors"] = list() ret["missing_sensors"] = list() ret["missing_tunnel"] = list() for dname in res_mappings.keys(): if dname not in res_dist: ret["missing_distributors"].append(dname) for sname, lofsensors in res_mappings.items(): for s in lofsensors: if s not in res_pos: 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 : 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() def get_input_positions_combined(source: str, use_iter: bool) -> tuple[dict, dict, dict, dict]: if use_iter: return get_input_positions_iter(source) else: return get_input_positions(source) 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() if args.sensors: # Sensoren auslesen res_sens, res_schaltschrank_elemente, res_double, missing_attribs = get_input_positions_combined(dxf_path if use_iter else msp, use_iter) if args.errors and len(res_double) > 0: print("Duplicate blocks found. Writing errors-file.") err_ids = list(res_double.keys()) write_results(to_json(list(err_ids)), work_dir, args.errors) 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 if args.console: print(to_json(res_mappings)) # Distributoren auslesen if use_iter: res_dist = get_subdistributor_positions_iter(dxf_path, res_mappings) else: res_dist = get_subdistributor_positions(msp, res_mappings) output_results['distributors'] = res_dist if args.console: print(to_json(res_dist)) # Tunnel auslesen if use_iter: res_tunnel = get_tunnel_positions_iter(dxf_path) else: res_tunnel = get_tunnel_positions(msp) 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) if args.errors and "overdefined_tunnel" in res_not_found: print("unorrect tunnels found. Writing errors-file.") err_ids = list(res_not_found["overdefined_tunnel"]) write_results(to_json(list(err_ids)), work_dir, args.errors) added_warnings = merge_two_dicts(warnings, missing_attribs) res_not_found["missing_attributes"] = added_warnings output_results["not_found"] = res_not_found output_results["double_ids"] = res_double write_results(to_json(output_results), work_dir, f"{basename}.json") else: parser.print_help()