From f2a6f0e35dc046d02b490183803d7737ed7181e5 Mon Sep 17 00:00:00 2001 From: mistangl Date: Fri, 6 Jun 2025 17:02:20 +0200 Subject: [PATCH] Kabeldaten auch extrahiert --- lib/getpositions.py | 199 ++++++++++++++++++++++++-------------------- 1 file changed, 107 insertions(+), 92 deletions(-) diff --git a/lib/getpositions.py b/lib/getpositions.py index 8c0fac6..682d6f7 100644 --- a/lib/getpositions.py +++ b/lib/getpositions.py @@ -36,115 +36,128 @@ def merge_two_dicts(x, y): z.update(y) return z -def get_input_positions(msp: ezdxf.document.Drawing.modelspace): - """hole alle Positionen der Eingänge !!Sensor Positionen erst nach Offsett-Addition "Mitte-Mitte"!! - """ - allIds = dict() +def get_attributes_of_insert(insert): + SpecialKeys = ["MB", # Separator - "MA", # Motor - "BG", # Stausensor - "FC" # Motorschutzschalter + "MA", # Motor + "BG", # Stausensor + "BP" # Schalter Druckluft ] - # Über alle Blockreferenzen (INSERT) im Modelspace laufen - for insert in msp.query('INSERT'): + KabelKey = "WD_" + DropKeys = [ + "FC", # Motorschutzschalter + "PF", # Leuchtmelder + "SF", + "DI", # Feedback vom Gerät + "QA", # Hauptschütz + "SF" # Drucktaster + ] + id = "" + ld = dict() + for attrib in insert.attribs: + attr_tag = attrib.dxf.tag + attr_text = attrib.dxf.text if len(insert.attribs) == 0: continue # Überspringe Blöcke ohne Attribute - id = "" - ld = dict() - for attrib in insert.attribs: - attr_tag = attrib.dxf.tag - attr_text = attrib.dxf.text - if len(insert.attribs) == 0: - continue # Überspringe Blöcke ohne Attribute - #print(f"Attribut Name: {attrib.dxf.tag}, Wert: {attrib.dxf.text}") - ld[attr_tag] = attr_text - if attr_tag == "IO": + #print(f"Attribut Name: {attrib.dxf.tag}, Wert: {attrib.dxf.text}") + ld[attr_tag] = attr_text + typ = 'unknown' + if attr_tag == "IO": + id = attr_text + #print(f"-- coord io {id}--: {attrib.dxf.insert}") # position des Blocks + pos = attrib.dxf.insert #Position aufzeichnen und bei Bedarf später mit REAL_POS überschreiben + ld["pos"] = (round(pos.x, 1), round(pos.y, 1)) + typ = "Sensor" + + if attr_tag == "B": + # Suche nach Sensoren + for spec in SpecialKeys: + if spec in attr_text: + id = attr_text + typ = "Sensor" + + # Suche nach Kabel + if KabelKey in attr_text: id = attr_text - #print(f"-- coord io {id}--: {attrib.dxf.insert}") # position des Blocks - pos = attrib.dxf.insert #Position aufzeichnen und bei Bedarf später mit REAL_POS überschreiben - ld["pos"] = (round(pos.x, 1), round(pos.y, 1)) + typ = "Kabel" + + # suche nach Items die wir nicht weiter verfolgen + for dropkey in DropKeys: + if dropkey in attr_text: + typ = "Schaltschrankelement" + id = attr_text + return {}, id, typ - if attr_tag == "B": - for spec in SpecialKeys: - if spec in attr_text: - id = attr_text - #print(f"-- coord {attrib.dxf.text} --: {attrib.dxf.insert}") - if attr_tag == "REALE_POSITION" and attr_text == "x": - #print(f"-- coord real --: {attrib.dxf.insert}") - pos = attrib.dxf.insert #Position Ecke unten links von "x"-Marker auslesen - - # 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)) + if attr_tag == "REALE_POSITION" and attr_text == "x": + #print(f"-- coord real --: {attrib.dxf.insert}") + pos = attrib.dxf.insert #Position Ecke unten links von "x"-Marker auslesen + + # 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)) + typ = "Sensor" + + return ld, id, typ +def get_input_positions(msp: ezdxf.document.Drawing.modelspace)-> tuple: + """hole alle Positionen der Eingänge !!Sensor Positionen erst nach Offsett-Addition "Mitte-Mitte"!! + """ + allSensors = dict() + allCables = dict() - # Nur wenn eine ID vorhanden ist, und eine gültige Position existiert + # Über alle Blockreferenzen (INSERT) im Modelspace laufen + for insert in msp.query('INSERT'): + # Überspringe Blöcke ohne Attribute + if len(insert.attribs) == 0: + continue + + ld, id, typ = get_attributes_of_insert(insert) + + # Nur wenn eine ID vorhanden ist, und eine gültige Position existiert + if typ == "Sensor": if id and "pos" in ld and isinstance(ld["pos"], tuple) and len(ld["pos"]) == 2: - if id in allIds: - allIds[id] = merge_two_dicts(allIds[id], ld) #Kombiniert alle infos aus dxf und "pos" + if id in allSensors: + allSensors[id] = merge_two_dicts(allSensors[id], ld) #Kombiniert alle infos aus dxf und "pos" else: - allIds[id] = ld - return allIds + allSensors[id] = ld + elif typ == "Kabel": + allCables[id] = ld + else: + # Falls das Objekt ein Teil des Schaltschranks o.ä ist, dann nichts merken + pass + return allSensors, allCables -def get_input_positions_iter(dxf_path) -> dict: +def get_input_positions_iter(dxf_path) -> tuple: """ Iterative Version für große DXF-Dateien mit iterdxf (kein Context Manager!). """ - - SpecialKeys = ["MB", "MA", "BG", "FC"] - allIds = dict() - + allSensors = dict() + allCables = dict() for insert in iterdxf.modelspace(dxf_path): if insert.dxftype() != 'INSERT': continue - attribs = {att.dxf.tag: att.dxf.text for att in insert.attribs or []} - id = "" - ld = dict() + ld, id, typ = get_attributes_of_insert(insert) - for attrib in insert.attribs: - attr_tag = attrib.dxf.tag - attr_text = attrib.dxf.text - if len(insert.attribs) == 0: - continue # Überspringe Blöcke ohne Attribute - #print(f"Attribut Name: {attrib.dxf.tag}, Wert: {attrib.dxf.text}") - ld[attr_tag] = attr_text - if attr_tag == "IO": - id = attr_text - #print(f"-- coord io {id}--: {attrib.dxf.insert}") # position des Blocks - pos = attrib.dxf.insert #Position aufzeichnen und bei Bedarf später mit REAL_POS überschreiben - ld["pos"] = (round(pos.x, 1), round(pos.y, 1)) - - if attr_tag == "B": - for spec in SpecialKeys: - if spec in attr_text: - id = attr_text - #print(f"-- coord {attrib.dxf.text} --: {attrib.dxf.insert}") - if attr_tag == "REALE_POSITION" and attr_text == "x": - #print(f"-- coord real --: {attrib.dxf.insert}") - pos = attrib.dxf.insert #Position Ecke unten links von "x"-Marker auslesen - - # 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") - pos_midx = pos.x + breite_marker*0.5 - pos_midy = pos.y + hoehe_marker*0.5 - - ld["pos"] = (round(pos_midx, 1), round(pos_midy, 1)) - - # Nur wenn eine ID vorhanden ist, und eine gültige Position existiert + # Nur wenn eine ID vorhanden ist, und eine gültige Position existiert + if typ == "Sensor": if id and "pos" in ld and isinstance(ld["pos"], tuple) and len(ld["pos"]) == 2: - if id in allIds: - allIds[id] = merge_two_dicts(allIds[id], ld) #Kombiniert alle infos aus dxf und "pos" + if id in allSensors: + allSensors[id] = merge_two_dicts(allSensors[id], ld) #Kombiniert alle infos aus dxf und "pos" else: - allIds[id] = ld + allSensors[id] = ld + elif typ == "Kabel": + allCables[id] = ld + else: + # Falls das Objekt ein Teil des Schaltschranks o.ä ist, dann nichts merken + pass - return allIds + return allSensors, allCables def create_mappings(positions:dict) -> dict: unterverteiler_pfad = "" @@ -435,7 +448,8 @@ if __name__ == '__main__': - res_pos = dict() + res_sens = dict() + res_cables = dict() res_dist = dict() res_rac = dict() res_mappings = dict() @@ -452,17 +466,18 @@ if __name__ == '__main__': if args.sensors: # Sensoren auslesen if use_iter: - res_pos = get_input_positions_iter(dxf_path) + res_sens, res_cables = get_input_positions_iter(dxf_path) else: - res_pos = get_input_positions(msp) + res_sens, res_cables = get_input_positions(msp) - output_results['sensors'] = res_pos + output_results['sensors'] = res_sens + output_results['cables'] = res_cables if args.console: - print(to_json(res_pos)) + print(to_json(res_sens)) # Mapping zu Sensoren auslesen - (res_mappings, warnings) = create_mappings(res_pos) + (res_mappings, warnings) = create_mappings(res_sens) output_results['mappings'] = res_mappings if args.console: print(to_json(res_mappings)) @@ -503,7 +518,7 @@ if __name__ == '__main__': if args.write: basename = os.path.splitext(args.write)[0] - res_not_found = check_existance(res_mappings, res_dist, res_pos) + res_not_found = check_existance(res_mappings, res_dist, res_sens) output_results["not_found"] = res_not_found write_results(to_json(output_results), work_dir, f"{basename}.json") zeitende = time.time()