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

PEP 8 styleguide mit curser angewandt

Browse Source
This commit is contained in:
Michael Stangl
2025-07-14 17:27:32 +02:00
parent 024819df5c
commit 7c1827908e
1 changed files with 134 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/getpositions.py
+134 -152
View File
@@ -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()