m.stangl/kabellaengen
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Typen Ein und Rückgabewerte dazu gemacht. PEP 8 styleguide angewandt

Browse Source
This commit is contained in:
Michael Stangl
2025-07-14 17:52:39 +02:00
parent 455ad888f2
commit a2d22500c4
2 changed files with 80 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bin/_setenv_local.bat
+29
View File
@@ -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
lib/getpositions.py
+51 -46
View File
@@ -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.")