From ea807c5365c9160526a9aec8d70be54ac654d4ea Mon Sep 17 00:00:00 2001 From: mistangl Date: Fri, 4 Jul 2025 22:10:28 +0200 Subject: [PATCH] =?UTF-8?q?Neue=20Suche=20nach=20Doppelrahmen=20=C3=BCber?= =?UTF-8?q?=20eigene=20Pufferklasse?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- lib/getpositions.py | 313 ++++++++++++++++++++++++++++---------------- 1 file changed, 201 insertions(+), 112 deletions(-) diff --git a/lib/getpositions.py b/lib/getpositions.py index b91808d..877097d 100644 --- a/lib/getpositions.py +++ b/lib/getpositions.py @@ -8,6 +8,7 @@ import re from ezdxf.addons import iterdxf import re from pathlib import Path +from shapely.geometry import Point """ @@ -33,41 +34,6 @@ def merge_two_dicts(x, y): z.update(y) return z -def get_type_of_name(name): - - SpecialKeys = ["MB", # Ventil - "MA", # Motor - "BG", # Stausensor - "BP", # Schalter Druckluft - "QM", # Ventile - "BX" # Scanner - ] - - KabelKey = ["WD", "WF"] - DropKeys = [ - "FC", # Motorschutzschalter - "PF", # Leuchtmelder - "DI", # Feedback vom Gerät - "QA", # Hauptschütz - "SF" # Drucktaster - ] - - prefix = name[:2] - if prefix in SpecialKeys: - typ = "Sensor" - - # Suche nach Kabel - elif prefix in KabelKey: - typ = "Kabel" - - # suche nach Items die wir nicht weiter verfolgen - elif prefix in DropKeys: - typ = "Schaltschrankelement" - - else: - typ = "unknown" - - return typ def get_type_of_name_cfg(name): prefix = name[:2] @@ -79,7 +45,7 @@ def get_type_of_name_cfg(name): else: return "unknown" -def get_attributes_of_insert(insert): +def get_attributes_of_insert(dinsert, dpos): """ Diese Funktion schaut nach den aktuell definierten Attributen in den Blöcken @@ -95,94 +61,223 @@ def get_attributes_of_insert(insert): """ id = "" - ld = dict() + ld = dinsert typ = 'unknown' - 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 == "NAME": # die neueren Bläcke heissen nicht IO, sondern haben einen Namen - typ = get_type_of_name_cfg(attr_text) - id = attr_text - # 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 == "IO": - typ = get_type_of_name_cfg(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)) - if attr_tag == "B": - # Suche nach Sensoren - typ = get_type_of_name_cfg(attr_text) - id = attr_text - 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)) + # die neueren Bläcke heissen nicht IO, sondern haben einen Namen + if "NAME" in dinsert: + typ = get_type_of_name_cfg(dinsert["NAME"]) + id = dinsert["NAME"] + + if "IO" in dinsert: + attr_text = dinsert["IO"] + typ = get_type_of_name_cfg(attr_text) + id = dinsert["IO"] + pos = dpos["IO"] + + if "REALE_POSITION" in dinsert and dinsert["REALE_POSITION"] == 'x': + pos = dpos["REALE_POSITION"] - if attr_tag == "REALE_POSITION" and attr_text == "x": - #typ = get_type_of_name(attr_text) - #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)) - - return (ld, id, typ) + else: + ld["pos"] = (pos[0], pos[1]) -def check_double_ids(ld, id, allSensors): - if id in allSensors and "NAME" in ld: - return True - if id in allSensors and "B" in ld and "B" in allSensors[id]: - return True - if id in allSensors and "IO" in ld and "IO" in allSensors[id]: - return True - return False + if "B" in dinsert: + attr_text = dinsert["B"] + typ = get_type_of_name_cfg(attr_text) + id = attr_text + #Position aufzeichnen und bei Bedarf später mit REAL_POS überschreiben + pos = dpos["B"] + ld["pos"] = (pos[0], pos[1]) + + return (ld, id, typ) + +class CompareBuffer: + """ein Puffer um alle als Doppelt zugewiesenen blöcke zwischenzulagern. + Anhand von ähnlicher Position und gleicher ID wird dann entschieden, + ob die Daten beider dicts zusammen gehören""" + def __init__(self): + self._wartepuffer = dict() + + def add_block(self, id, buffer): + """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): + if id in self._wartepuffer: + return self._wartepuffer[id] + else: + return [] + + def get_block_ids(self)->list: + return list(self._wartepuffer.keys()) + + + def set_block(self, id, buffer): + if id in self._wartepuffer: + del self._wartepuffer[id] + self._wartepuffer[id] = buffer + + def dict_compare(self, d1, d2): + str1 = json.dumps(d1) + str2 = json.dumps(d2) + return str1 == str2 + + def remove_block(self, id, toRemove): + """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, toRemove): + pass + else: + l.append(b) + self.set_block(id, l) + + def positions_are_close(self, dict1:dict, dict2:dict, border:float): + 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): + """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): + """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: allSensors = dict() allCables = dict() - allunknowns = dict() - doubleIds = dict() allSchaltschrank = dict() + allunknowns = list() - for insert in insert_iterable: - if insert.dxftype() != 'INSERT': - continue + WP = CompareBuffer() + all_inserts, all_positions = attribs_to_dicts(insert_iterable) - ld, id, typ = get_attributes_of_insert(insert) + for insert, pos in zip(all_inserts, all_positions): + ld, id, typ = get_attributes_of_insert(insert, pos) if typ == "Sensor": - if id and "pos" in ld and isinstance(ld["pos"], tuple) and len(ld["pos"]) == 2: - if id in allSensors: - is_double = check_double_ids(ld, id, allSensors) - if not is_double: - allSensors[id] = merge_two_dicts(allSensors[id], ld) #Kombiniert alle infos aus dxf und "pos" - else: - # gib aus wo sich beide gleichlautenden Elemente auf der Zeichnung befinden - if id not in doubleIds: - doubleIds[id] = [] - doubleIds[id].append(allSensors[id]["pos"]) # erste Position - doubleIds[id].append(ld["pos"]) # zweite Position - else: - allSensors[id] = ld + # wenn NAME enthalten ist, dann ist das ein eindeutiger Bezeichner + if "NAME" in ld: + allSensors[id] = ld + else: + # alle anderen Sachen werden dann aus mehreren Rahmen zusammen gesammelt + WP.add_block(id, ld) elif typ == "Kabel": allCables[id] = ld elif typ == "Schaltschrankelement": allSchaltschrank[id] = ld else: - allunknowns[id] = ld + allunknowns.append(ld) - return allSensors, doubleIds + # 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(allSensors, WP) + + # die noch übrigen Blöcke melden + missing_attribs, doubleIds = get_errors_double_and_attributes(WP) + + return allSensors, doubleIds, missing_attribs + +def get_errors_double_and_attributes(WP): + missing_attribs = dict() + doubleIds = 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 doubleIds: + doubleIds[id] = [] + doubleIds[id].append(block['pos']) + return missing_attribs, doubleIds + +def allocate_blocks_together(allSensors:dict, WP:CompareBuffer): + """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): + newId = create_new_id(id, block_with_sps, block_without_sps) # hier das Präfix davor + allSensors[newId] = 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, dict1, dict2): + 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"{sps_praefix}:{id}" + +def attribs_to_dicts(insert_iterable): + 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: return extract_input_positions(msp.query('INSERT')) @@ -193,7 +288,7 @@ def get_input_positions_iter(dxf_path) -> tuple: def create_mappings(positions:dict) -> tuple: unterverteiler_pfad = "" dnamen = dict() - # sammle die Sensoren mit ihren zugehörigen Unterverteilern + # s sensor2unterverteiler = dict() warnings = dict() for sensorname,v in positions.items(): @@ -205,17 +300,11 @@ def create_mappings(positions:dict) -> tuple: #print(unterverteiler_pfad) # Pfad zur Karte splitten. Dieser hat z.B. den Inhalt "=AH01+UH02-KF1FDI7" - pattern = r"^=([A-Z]+\d+)([+\-])([A-Z]+\d+)([+\-])([A-Z]+\d+.)$" - match = re.match(pattern, unterverteiler_pfad) - if match: - anlage = match.group(1) - verteiler = match.group(3) - karte = match.group(5) - # match.group(1) # AH01 - # match.group(2) # + - # match.group(3) # UH02 - # match.group(4) # - - # match.group(5) # KF1FDI7 + matches = re.findall(r'[^\-+=]+', unterverteiler_pfad.lstrip('=')) + if matches: + anlage = matches[0] + verteiler = matches[1] + karte = matches[2] else: warnings[sensorname] = f"Ungültiger Pfad in Kennzeichnung: {unterverteiler_pfad}" continue @@ -586,9 +675,9 @@ if __name__ == '__main__': if args.sensors: # Sensoren auslesen if use_iter: - res_sens, res_double = get_input_positions_iter(dxf_path) + res_sens, res_double, missing_attribs = get_input_positions_iter(dxf_path) else: - res_sens, res_double = get_input_positions(msp) + res_sens, res_double, missing_attribs = get_input_positions(msp) if args.errors and len(res_double) > 0: print("Duplicate blocks found. Writing errors-file.")