From a2d22500c42c8467ccd71dbbf8e0f620fa83737f Mon Sep 17 00:00:00 2001 From: mistangl Date: Mon, 14 Jul 2025 17:52:39 +0200 Subject: [PATCH] =?UTF-8?q?Typen=20Ein=20und=20R=C3=BCckgabewerte=20dazu?= =?UTF-8?q?=20gemacht.=20PEP=208=20styleguide=20angewandt?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- bin/_setenv_local.bat | 29 +++++++++++++ lib/getpositions.py | 97 +++++++++++++++++++++++-------------------- 2 files changed, 80 insertions(+), 46 deletions(-) create mode 100644 bin/_setenv_local.bat diff --git a/bin/_setenv_local.bat b/bin/_setenv_local.bat new file mode 100644 index 0000000..a0ef4b0 --- /dev/null +++ b/bin/_setenv_local.bat @@ -0,0 +1,29 @@ +@echo off +REM ~dp0 steht für das Verzeichnis, in der diese Datei liegt +pushd %~dp0\.. + +set PROJECT=C:\10-Develop\gitrepos\kabellaengen +set PROJECT_BIN=%PROJECT%\bin +set PROJECT_CFG=%PROJECT%\cfg +set PROJECT_DOC=%PROJECT%\doc +set PROJECT_LIB=%PROJECT%\lib +set PROJECT_DATA=%PROJECT%\data +set PROJECT_WORK=%PROJECT%\work +set PROJECT_TEST=%PROJECT%\testdata + +set SIVAS_TEILESTAMM=\\195.243.223.3\sivas\jit\programme\KSbExcelExportSivasTeilestamm.exe +set SIVAS_EXCEL_EXPORT_DIR=%PROJECT_DATA% + +if not exist %PROJECT%\work mkdir %PROJECT%\work +if not exist %PROJECT%\data mkdir %PROJECT%\data + +set INSTALL_DIR="%ONEDRIVE%\Desktop\Kabeltool" + +set PATH=%PROJECT_BIN%;%PATH% + + +popd +goto :eof + + + diff --git a/lib/getpositions.py b/lib/getpositions.py index eafbb7e..df07d3e 100644 --- a/lib/getpositions.py +++ b/lib/getpositions.py @@ -1,26 +1,31 @@ import argparse import configparser -import ezdxf -import os -import sys import json +import os import re -from ezdxf.addons import iterdxf +import sys from pathlib import Path + +import ezdxf import ezdxf.filemanagement +from ezdxf.addons import iterdxf from shapely.geometry import Point +# Fix import for DXFStructureError +from ezdxf.lldxf.const import DXFStructureError + """ Dieses Programm: -- liest die dxf Datei und holt sich von den Layern der dxf Datei die Positionen +- 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, out_dir, filename): + + +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) @@ -29,13 +34,13 @@ def write_results(jsn_results, out_dir, filename): print("done") -def merge_two_dicts(x, y): +def merge_two_dicts(x: dict, y: dict) -> dict: z = x.copy() z.update(y) return z -def get_type_of_name_cfg(name): +def get_type_of_name_cfg(name: str) -> str: prefix = name[:2] if config_BMK.has_option("Routing-Include", prefix): @@ -46,7 +51,7 @@ def get_type_of_name_cfg(name): return "unknown" -def get_attributes_of_insert(d_insert, d_pos): +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 @@ -103,36 +108,36 @@ def get_attributes_of_insert(d_insert, d_pos): class CompareBuffer: """Ein Puffer um alle als Doppelt zugewiesenen Blöcke zwischenzulagern.""" - def __init__(self): + def __init__(self) -> None: self._wartepuffer = dict() - def add_block(self, id_, buffer): + 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_): + def get_blocks(self, id_: str) -> list: if id_ in self._wartepuffer: return self._wartepuffer[id_] else: return [] - def get_block_ids(self) -> list: + def get_block_ids(self) -> list[str]: return list(self._wartepuffer.keys()) - def set_block(self, id_, buffer): + 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, d2): + 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_, to_remove): + 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() @@ -143,7 +148,7 @@ class CompareBuffer: l.append(b) self.set_block(id_, l) - def positions_are_close(self, dict1:dict, dict2:dict, border:float): + def positions_are_close(self, dict1: dict, dict2: dict, border: float) -> bool: pos1 = None pos2 = None if "pos" in dict1: @@ -159,7 +164,7 @@ class CompareBuffer: return True return False - def get_all_sps_blocks(self, id_): + 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() @@ -168,7 +173,7 @@ class CompareBuffer: l.append(b) return l - def get_non_sps_blocks(self, id_): + 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() @@ -177,7 +182,7 @@ class CompareBuffer: l.append(b) return l -def extract_input_positions(insert_iterable) -> tuple: +def extract_input_positions(insert_iterable) -> tuple[dict, dict, dict]: all_sensors = dict() all_cables = dict() all_schaltschrank = dict() @@ -213,7 +218,7 @@ def extract_input_positions(insert_iterable) -> tuple: return all_sensors, double_ids, missing_attribs -def get_errors_double_and_attributes(wp): +def get_errors_double_and_attributes(wp: CompareBuffer) -> tuple[dict, dict]: missing_attribs = dict() double_ids = dict() for id_ in wp.get_block_ids(): @@ -231,7 +236,7 @@ def get_errors_double_and_attributes(wp): double_ids[id_].append(block['pos']) return missing_attribs, double_ids -def allocate_blocks_together(all_sensors:dict, wp:CompareBuffer): +def allocate_blocks_together(all_sensors: dict, wp: CompareBuffer) -> None: """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() @@ -248,7 +253,7 @@ def allocate_blocks_together(all_sensors:dict, wp:CompareBuffer): 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_: str, dict1: dict, dict2: dict) -> str: sps_praefix = None if "SPS" in dict1: sps_praefix = dict1["SPS"] @@ -258,7 +263,7 @@ def create_new_id(id_, dict1, dict2): raise Exception return f"{id_}@{sps_praefix}" -def attribs_to_dicts(insert_iterable): +def attribs_to_dicts(insert_iterable) -> tuple[list, list]: all_inserts = list() all_positions = list() for insert in insert_iterable: @@ -280,13 +285,13 @@ def attribs_to_dicts(insert_iterable): all_positions.append(positions) return all_inserts, all_positions -def get_input_positions(msp) -> tuple: +def get_input_positions(msp) -> tuple[dict, dict, dict]: return extract_input_positions(msp.query('INSERT')) -def get_input_positions_iter(dxf_path) -> tuple: +def get_input_positions_iter(dxf_path) -> tuple[dict, dict, dict]: return extract_input_positions(iterdxf.modelspace(dxf_path)) -def create_mappings(positions:dict) -> tuple: +def create_mappings(positions: dict) -> tuple[dict, dict]: unterverteiler_pfad = "" dnamen = dict() # s @@ -324,7 +329,7 @@ def create_mappings(positions:dict) -> tuple: return (uv2sensor, warnings) -def get_subdistributor_positions(msp, dist2sensors): +def get_subdistributor_positions(msp, dist2sensors: dict) -> dict: """Hole alle Positionen der Unterverteiler !!UV-Positionen bereits 'Mitte-Mitte'!!""" ret = dict() all_distributors = dist2sensors.keys() @@ -338,7 +343,7 @@ def get_subdistributor_positions(msp, dist2sensors): ret[distname] = (round(text.dxf.insert[0], 1), round(text.dxf.insert[1], 1)) return ret -def get_subdistributor_positions_iter(dxf_path, dist2sensors): +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() @@ -360,7 +365,7 @@ def get_subdistributor_positions_iter(dxf_path, dist2sensors): return ret -def get_tunnel_positions(msp): +def get_tunnel_positions(msp) -> dict: """Hole alle Positionen aller Tunnel Ein und Ausgänge.""" all_tunnels = dict() tunnel_length = dict() @@ -385,7 +390,7 @@ def get_tunnel_positions(msp): all_tunnels['length'] = tunnel_length return all_tunnels -def get_tunnel_positions_iter(dxf_path): +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() @@ -419,18 +424,18 @@ def get_tunnel_positions_iter(dxf_path): return all_tunnels # helper function -def print_line(e): +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): +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): +def get_rack_positions(msp) -> dict: """Hole alle Positionen aller Kabelpritschen und nummeriere Racks.""" ret = dict() rack_counter = 1 @@ -451,7 +456,7 @@ def get_rack_positions(msp): return ret -def get_rack_positions_iter(dxf_path): +def get_rack_positions_iter(dxf_path) -> dict: """Hole alle Positionen aller Kabelpritschen (Racks) mithilfe von iterdxf.""" ret = dict() rack_counter = 1 @@ -476,7 +481,7 @@ def get_rack_positions_iter(dxf_path): return ret -def handle_lwpolyline(entity, rack_key, 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] = [] @@ -484,7 +489,7 @@ def handle_lwpolyline(entity, rack_key, ret): x, y, *_ = point ret[rack_key].append([round(x, 1), round(y, 1), round(z, 1)]) -def handle_polyline(entity, rack_key, ret): +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: @@ -493,7 +498,7 @@ 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): +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")) @@ -503,10 +508,10 @@ def scan(dxf_source): ret['all_blocks'] = alle_block_defs return ret -def to_json(d, pretty: bool = True) -> str: +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): +def get_dxf_file(filepath: Path): """Hole das dxf file.""" try: print("reading file ..", end='') @@ -515,12 +520,12 @@ def get_dxf_file(filepath): except IOError: print("Not a DXF file or a generic I/O error.") sys.exit(1) - except ezdxf.DXFStructureError: + except DXFStructureError: 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) -> tuple[Path, bool]: fexists = True if not filename.exists(): mypath = work_dir.joinpath(filename) @@ -531,7 +536,7 @@ def check_file_in_work(work_dir: Path, filename: Path): mypath = filename return mypath, fexists -def check_existance(res_mappings, res_dist, res_pos): +def check_existance(res_mappings: dict, res_dist: dict, res_pos: dict) -> dict: ret = dict() ret["missing_distributors"] = list() ret["missing_sensors"] = list() @@ -545,12 +550,12 @@ def check_existance(res_mappings, res_dist, res_pos): return ret -def dxf_is_binary(dxf_path): +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(): +def validate_configs() -> None: errors = [] print("\nValidating given configs: Checking for inconsistency.")