From 7c1827908eeaae5d1268cea1e457536a223bdf93 Mon Sep 17 00:00:00 2001 From: mistangl Date: Mon, 14 Jul 2025 17:27:32 +0200 Subject: [PATCH] PEP 8 styleguide mit curser angewandt --- lib/getpositions.py | 286 +++++++++++++++++++++----------------------- 1 file changed, 134 insertions(+), 152 deletions(-) diff --git a/lib/getpositions.py b/lib/getpositions.py index dc05ef4..eafbb7e 100644 --- a/lib/getpositions.py +++ b/lib/getpositions.py @@ -20,15 +20,15 @@ Dieses Programm: - erzeugt daraus eine .json Datei im Work Ordner """ -def write_results(jsnResults, outdir, filename): - """ write results to a json file - """ +def write_results(jsn_results, out_dir, filename): + """Write results to a JSON file.""" print("writing results to a json file ...") - outfile = os.path.join(outdir, filename) + outfile = os.path.join(out_dir, filename) with open(outfile, 'w', encoding='utf-8') as fh: - fh.write(jsnResults) + fh.write(jsn_results) print("done") + def merge_two_dicts(x, y): z = x.copy() z.update(y) @@ -45,7 +45,8 @@ def get_type_of_name_cfg(name): else: return "unknown" -def get_attributes_of_insert(dinsert, dpos): + +def get_attributes_of_insert(d_insert, d_pos): """ Diese Funktion schaut nach den aktuell definierten Attributen in den Blöcken @@ -60,25 +61,25 @@ def get_attributes_of_insert(dinsert, dpos): 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 = dinsert + id_ = "" + ld = d_insert typ = 'unknown' # 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"] - pos = dpos["NAME"] + 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 "IO" in dinsert: - attr_text = dinsert["IO"] + if "IO" in d_insert: + attr_text = d_insert["IO"] typ = get_type_of_name_cfg(attr_text) - id = dinsert["IO"] - pos = dpos["IO"] + id_ = d_insert["IO"] + pos = d_pos["IO"] - if "REALE_POSITION" in dinsert and dinsert["REALE_POSITION"] == 'x': - pos = dpos["REALE_POSITION"] + 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") @@ -89,60 +90,58 @@ def get_attributes_of_insert(dinsert, dpos): else: ld["pos"] = (pos[0], pos[1]) - if "B" in dinsert: - attr_text = dinsert["B"] + if "B" in d_insert: + attr_text = d_insert["B"] typ = get_type_of_name_cfg(attr_text) - id = attr_text + id_ = attr_text #Position aufzeichnen und bei Bedarf später mit REAL_POS überschreiben - pos = dpos["B"] + pos = d_pos["B"] ld["pos"] = (pos[0], pos[1]) - return (ld, id, typ) + 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""" + """Ein Puffer um alle als Doppelt zugewiesenen Blöcke zwischenzulagern.""" 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 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] + def get_blocks(self, id_): + if id_ in self._wartepuffer: + return self._wartepuffer[id_] else: return [] - def get_block_ids(self)->list: + 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 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) + def remove_block(self, id_, to_remove): + """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): + if self.dict_compare(b, to_remove): pass else: l.append(b) - self.set_block(id, l) + self.set_block(id_, l) def positions_are_close(self, dict1:dict, dict2:dict, border:float): pos1 = None @@ -160,18 +159,18 @@ class CompareBuffer: 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) + 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) + 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: @@ -179,77 +178,77 @@ class CompareBuffer: return l def extract_input_positions(insert_iterable) -> tuple: - allSensors = dict() - allCables = dict() - allSchaltschrank = dict() - allunknowns = list() + all_sensors = dict() + all_cables = dict() + all_schaltschrank = dict() + all_unknowns = list() - WP = CompareBuffer() + 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) + ld, id_, typ = get_attributes_of_insert(insert, pos) if typ == "Sensor": # wenn NAME enthalten ist, dann ist das ein eindeutiger Bezeichner if "NAME" in ld: - allSensors[id] = ld + all_sensors[id_] = ld else: # alle anderen Sachen werden dann aus mehreren Rahmen zusammen gesammelt - WP.add_block(id, ld) + wp.add_block(id_, ld) elif typ == "Kabel": - allCables[id] = ld + all_cables[id_] = ld elif typ == "Schaltschrankelement": - allSchaltschrank[id] = ld + all_schaltschrank[id_] = ld else: - allunknowns.append(ld) + 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(allSensors, WP) + allocate_blocks_together(all_sensors, wp) # die noch übrigen Blöcke melden - missing_attribs, doubleIds = get_errors_double_and_attributes(WP) + missing_attribs, double_ids = get_errors_double_and_attributes(wp) - return allSensors, doubleIds, missing_attribs + return all_sensors, double_ids, missing_attribs -def get_errors_double_and_attributes(WP): +def get_errors_double_and_attributes(wp): missing_attribs = dict() - doubleIds = dict() - for id in WP.get_block_ids(): - blocks = WP.get_blocks(id) + 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]) + 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 + if id_ not in double_ids: + double_ids[id_] = [] + double_ids[id_].append(block['pos']) + return missing_attribs, double_ids -def allocate_blocks_together(allSensors:dict, WP:CompareBuffer): +def allocate_blocks_together(all_sensors: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) + 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) + 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, dict1, dict2): +def create_new_id(id_, dict1, dict2): sps_praefix = None if "SPS" in dict1: sps_praefix = dict1["SPS"] @@ -257,7 +256,7 @@ def create_new_id(id, dict1, dict2): sps_praefix = dict2["SPS"] if not sps_praefix: raise Exception - return f"{id}@{sps_praefix}" + return f"{id_}@{sps_praefix}" def attribs_to_dicts(insert_iterable): all_inserts = list() @@ -326,20 +325,17 @@ def create_mappings(positions:dict) -> tuple: return (uv2sensor, warnings) def get_subdistributor_positions(msp, dist2sensors): - """hole alle Positionen der Unterverteiler !!UV-Positionen bereits "Mitte-Mitte"!! - """ + """Hole alle Positionen der Unterverteiler !!UV-Positionen bereits 'Mitte-Mitte'!!""" ret = dict() - # Alle Texte auf Layer "xy" all_distributors = dist2sensors.keys() all_layers = config.items('GetPos-Layer_Distributors') - for (layer,v) in all_layers: + for (layer, v) in all_layers: for distname in all_distributors: selectstr = f'MTEXT[layer=="{layer}"]' for text in msp.query(selectstr): - #print(f"Text auf Layer 'Busverteiler-Kennzeichnung': {text.dxf.text}") - match = re.search("-"+distname, text.dxf.text) + match = re.search("-" + distname, text.dxf.text) if match: - ret[distname] = (round(text.dxf.insert[0],1), round(text.dxf.insert[1],1)) #nur x und y Koordinate in Json schreiben + ret[distname] = (round(text.dxf.insert[0], 1), round(text.dxf.insert[1], 1)) return ret def get_subdistributor_positions_iter(dxf_path, dist2sensors): @@ -365,37 +361,34 @@ def get_subdistributor_positions_iter(dxf_path, dist2sensors): return ret def get_tunnel_positions(msp): - """hole alle Positionen aller Tunnel Ein und Ausgänge - """ - allTunnels = dict() + """Hole alle Positionen aller Tunnel Ein und Ausgänge.""" + all_tunnels = dict() tunnel_length = dict() - # Alle Text mit "Tunnel" als Inhalt auf Layer "xy" all_layers = config.items('GetPos-Layer_Tunnel') - for (layer,v) in all_layers: + 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)) #nur x und y Koordinate in Json schreiben + 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 not tunnelname in allTunnels: - allTunnels[tunnelname] = list() - allTunnels[tunnelname].append(pos) + if tunnelname not in all_tunnels: + all_tunnels[tunnelname] = list() + all_tunnels[tunnelname].append(pos) else: - allTunnels[tunnelname].append(pos) - if len(allTunnels.keys()) > 0: - allTunnels['length'] = tunnel_length - return allTunnels + 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): """Hole alle Positionen aller Tunnel Ein- und Ausgänge mithilfe von iterdxf.""" - allTunnels = dict() + all_tunnels = dict() tunnel_length = dict() - all_layers = config.items('GetPos-Layer_Tunnel') for entity in iterdxf.modelspace(dxf_path): @@ -417,44 +410,39 @@ def get_tunnel_positions_iter(dxf_path): laenge = match.group(2) pos = (round(insert[0], 1), round(insert[1], 1)) - if tunnelname not in allTunnels: - allTunnels[tunnelname] = [] - allTunnels[tunnelname].append(pos) + if tunnelname not in all_tunnels: + all_tunnels[tunnelname] = [] + all_tunnels[tunnelname].append(pos) tunnel_length[tunnelname] = laenge if len(tunnel_length.keys()) > 0: - allTunnels['length'] = tunnel_length - return allTunnels + all_tunnels['length'] = tunnel_length + return all_tunnels # helper function def print_line(e): - print("LINE on layer: %s\n" % e.dxf.layer) - print("points: %s\n" % repr(e.dxf)) + print(f"LINE on layer: {e.dxf.layer}\n") + print(f"points: {repr(e.dxf)}\n") def print_polyline(e): - print("POLYLINE on layer: %s\n" % e.dxf.layer) - #print("points: %s\n" % repr(e.dxf)) - #print("y point: %s\n" % e.dxf.y) - for x, y, start_width, end_width, bulge in e.get_points(): # Gibt Tuple (x, y, start_width, end_width, bulge) + 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): - """hole alle Positionen aller Kabelpritschen und nummeriere Racks""" + """Hole alle Positionen aller Kabelpritschen und nummeriere Racks.""" ret = dict() - rack_counter = 1 #Zaehler für Rack Nummerierung - + rack_counter = 1 all_layers = list(config.items('GetPos-Layer_Racks')) - + for layer, _ in all_layers: - # Suche nach LWPOLYLINE 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 - # Suche nach klassischer POLYLINE pl_query = f'POLYLINE[layer=="{layer}"]' for entity in msp.query(pl_query): rack_key = f"Rack_{rack_counter}" @@ -466,18 +454,17 @@ def get_rack_positions(msp): def get_rack_positions_iter(dxf_path): """Hole alle Positionen aller Kabelpritschen (Racks) mithilfe von iterdxf.""" ret = dict() - rack_counter = 1 # Zähler für Rack-Nummerierung - + 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 + continue rack_key = f"Rack_{rack_counter}" - + if entity.dxftype() == "LWPOLYLINE": handle_lwpolyline(entity, rack_key, ret) elif entity.dxftype() == "POLYLINE": @@ -491,10 +478,9 @@ def get_rack_positions_iter(dxf_path): def handle_lwpolyline(entity, rack_key, ret): """Verarbeitet eine 2D LWPOLYLINE mit globalem Z-Wert (elevation).""" - z = getattr(entity.dxf, "elevation", 0.0) # hole "Erhebung" (gilt für gesamte Polyline!) aus Attributen - + z = getattr(entity.dxf, "elevation", 0.0) ret[rack_key] = [] - for point in entity.vertices(): # returns (x, y, start_width, end_width, bulge) + for point in entity.vertices(): x, y, *_ = point ret[rack_key].append([round(x, 1), round(y, 1), round(z, 1)]) @@ -507,9 +493,9 @@ def handle_polyline(entity, rack_key, ret): z = vertex.dxf.location.z ret[rack_key].append([round(x, 1), round(y, 1), round(z, 1)]) -def scan(dxf_source:ezdxf.document.Drawing): - layer_names_inside = dxf_source.layers.entries.keys() - alle_block_defs = set(dxf_source.blocks.block_names()) +def scan(dxf_source): + 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 @@ -521,30 +507,29 @@ def to_json(d, 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): - """hole das dxf file - """ + """Hole das dxf file.""" try: print("reading file ..", end='') doc = ezdxf.filemanagement.readfile(filepath) print("done") except IOError: - print(f"Not a DXF file or a generic I/O error.") + print("Not a DXF file or a generic I/O error.") sys.exit(1) except ezdxf.DXFStructureError: - print(f"Invalid or corrupted DXF file.") + print("Invalid or corrupted DXF file.") sys.exit(2) return doc -def check_file_in_work(work_dir:Path, filename:Path): +def check_file_in_work(work_dir: Path, filename: Path): fexists = True - if not filename.exists(): # dann schau im Work Ordner nach + if not filename.exists(): mypath = work_dir.joinpath(filename) ex = mypath.exists() if not mypath.exists(): fexists = False else: mypath = filename - return (mypath, fexists) + return mypath, fexists def check_existance(res_mappings, res_dist, res_pos): ret = dict() @@ -553,7 +538,7 @@ def check_existance(res_mappings, res_dist, res_pos): 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 sname, lofsensors in res_mappings.items(): for s in lofsensors: if s not in res_pos: ret['missing_sensors'].append(s) @@ -566,17 +551,15 @@ def dxf_is_binary(dxf_path): return b'AutoCAD Binary DXF' in header def validate_configs(): - errors= [] + errors = [] print("\nValidating given configs: Checking for inconsistency.") - # 1. Alle Prefixes aus Routing_include müssen in Cable_Mapping stehen (nur BMK.cfg) 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'") - # 2. Jeder Eintrag in Cable-Mapping → Sektionen in kabel.cfg prüfen (Abgleich BMK.cfg und kabel.cfg) 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(",")] @@ -584,7 +567,6 @@ def validate_configs(): if not config_cables.has_section(section): errors.append(f"Cable-Section '{section}' from Cable-Mapping ({mapping_key}) missing in kabel.cfg") - # 3. Länge in Length-Adjustments muss float >= 0 sein if config_BMK.has_section("Length-Adjustments"): for prefix, value in config_BMK.items("Length-Adjustments"): try: @@ -600,9 +582,9 @@ def validate_configs(): print(f"- {e}") print("\ncontinuing with routing process") else: - print ("No inconsistencies found. Continuing with routing process.") + print("No inconsistencies found. Continuing with routing process.") -def check_environment_var(env_str:str) -> Path: +def check_environment_var(env_str: str) -> Path: out_path = os.environ.get(env_str) if out_path: return Path(out_path) @@ -657,17 +639,17 @@ if __name__ == '__main__': # Allgemeine Config Laden config = configparser.ConfigParser(allow_no_value=True, delimiters=("=")) - config.optionxform = lambda option: option # preserve case for letters + 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 option: option # preserve case for letters + 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 option: option + config_cables.optionxform = lambda optionstr: optionstr config_cables.read(os.path.join(config_dir, "kabel.cfg")) validate_configs()