Michael Stangl
895 lines
34 KiB
Python
895 lines
34 KiB
Python
import argparse
|
||
import configparser
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
from shapely.geometry import Point
|
||
|
||
from error_collector import ErrorCollector
|
||
from utils import (
|
||
check_environment_var,
|
||
check_file_in_work,
|
||
get_dxf_file,
|
||
merge_two_dicts,
|
||
to_json,
|
||
write_results,
|
||
extract_attributes_with_positions as attribs_to_dicts,
|
||
)
|
||
|
||
|
||
"""
|
||
Dieses Programm:
|
||
- 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 get_type_of_name_cfg(name: str) -> str:
|
||
prefix = name[:2]
|
||
|
||
if config_BMK.has_option("Routing-Include", prefix):
|
||
return "Sensor"
|
||
elif config_BMK.has_option("Routing-Ignore", prefix):
|
||
return "Schaltschrankelement"
|
||
else:
|
||
return "unknown"
|
||
|
||
|
||
def matches_cabinet_pattern(name: str) -> [bool, str]:
|
||
"""Check if the given name matches any Cabinet-Pattern from BMK.cfg"""
|
||
if not config_BMK.has_section("Cabinet-Pattern"):
|
||
return (False, "")
|
||
|
||
patterns = [pattern for pattern, _ in config_BMK.items("Cabinet-Pattern")]
|
||
|
||
for pattern in patterns:
|
||
|
||
m = re.search(pattern, name)
|
||
if m:
|
||
res = m.group(1)
|
||
return (True, res)
|
||
|
||
return (False, "")
|
||
|
||
|
||
def matches_tunnel_pattern(name: str) -> bool:
|
||
"""Check if the given name matches any Tunnel-Pattern from BMK.cfg"""
|
||
if not config_BMK.has_section("Tunnel-Pattern"):
|
||
return False
|
||
|
||
patterns = [pattern for pattern, _ in config_BMK.items("Tunnel-Pattern")]
|
||
|
||
for pattern in patterns:
|
||
m = re.search(pattern, name)
|
||
if m:
|
||
return True
|
||
|
||
return False
|
||
|
||
|
||
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
|
||
|
||
Bei den Sensoren in den alten Layouts gibt zwei immer übereinanderliegende Blöcke mit den Attributen:
|
||
- A, B (z.B. MA0062), C, ARTINR (z.B. 790902001), BESCHR (E-Teile für SEW Motor ASE1..), MENGE, POSITION, ...
|
||
- IO (z.B. MA0062), ID , VERW (z.B. CV-M0062_0,75), BEZEICHNUNG (Motor MA0062), KENNZEICHNUNG (z.B.=A01+UH01-KF1DQ04), ...
|
||
Die Adresse für das Routing kommt aus "KENNZEICHNUNG", während die Sivas Nummer aus "ARTINR" geholt werden muss
|
||
Einmal wird B zur ID, beim anderen IO
|
||
|
||
Erdungssymbole erhalten die ID aus dem Eintrag unter "NAME"
|
||
|
||
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 = d_insert
|
||
typ = 'unknown'
|
||
|
||
# die neueren Bläcke heissen nicht IO, sondern haben einen Namen
|
||
if "IO" in d_insert:
|
||
pos = d_pos["IO"]
|
||
typ = get_type_of_name_cfg(d_insert["IO"])
|
||
|
||
# neuer Block: Enthält alles was nötig is
|
||
if "ARTINR" in d_insert and "SPS" in d_insert:
|
||
id_ = d_insert["IO"]+"@"+d_insert["SPS"]
|
||
ld["pos"] = (pos[0], pos[1])
|
||
|
||
if "REALE_POSITION" in d_insert and d_insert["REALE_POSITION"] == 'x':
|
||
pos = d_pos["REALE_POSITION"]
|
||
|
||
else:
|
||
|
||
id_ = d_insert["IO"]
|
||
# Sensoren werden später gemerged mit den anderen Blöcken des Rahmens mit A,B,C, usw
|
||
if "SPS" in d_insert and typ != "Sensor":
|
||
id_ = id_+"@"+d_insert["SPS"]
|
||
|
||
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")
|
||
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))
|
||
else:
|
||
ld["pos"] = (pos[0], pos[1])
|
||
|
||
# die neueren Blöcke heissen nicht IO, sondern haben einen Namen
|
||
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 "B" in d_insert and "IO" not in d_insert:
|
||
attr_text = d_insert["B"]
|
||
typ = get_type_of_name_cfg(attr_text)
|
||
id_ = attr_text
|
||
#Position aufzeichnen und bei Bedarf später mit REAL_POS überschreiben
|
||
pos = d_pos["B"]
|
||
ld["pos"] = (pos[0], pos[1])
|
||
|
||
return ld, id_, typ
|
||
|
||
|
||
class CompareBuffer:
|
||
"""Ein Puffer um alle als Doppelt zugewiesenen Blöcke zwischenzulagern."""
|
||
def __init__(self) -> None:
|
||
self._wartepuffer = dict()
|
||
|
||
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_: str) -> list:
|
||
if id_ in self._wartepuffer:
|
||
return self._wartepuffer[id_]
|
||
else:
|
||
return []
|
||
|
||
def get_block_ids(self) -> list[str]:
|
||
return list(self._wartepuffer.keys())
|
||
|
||
|
||
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: dict, d2: dict) -> bool:
|
||
str1 = json.dumps(d1)
|
||
str2 = json.dumps(d2)
|
||
return str1 == str2
|
||
|
||
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()
|
||
for b in buffers:
|
||
if self.dict_compare(b, to_remove):
|
||
pass
|
||
else:
|
||
l.append(b)
|
||
self.set_block(id_, l)
|
||
|
||
def positions_are_close(self, dict1: dict, dict2: dict, border: float) -> bool:
|
||
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_: str) -> list:
|
||
"""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_: str) -> list:
|
||
"""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(all_inserts, all_positions, error_collector: ErrorCollector = None) -> tuple[dict, dict]:
|
||
"""
|
||
Extracts and organizes input positions from an iterable of inserts.
|
||
|
||
This function processes a list of insert objects (e.g., from a DXF file), classifies them by type
|
||
(Sensor, Kabel, Schaltschrankelement, or unknown), and organizes them into dictionaries keyed by their IDs.
|
||
For sensors, it handles the special case where sensor information may be split across multiple blocks
|
||
(e.g., IO and A,B,C blocks) and merges them if necessary. It also collects information about missing
|
||
attributes and duplicate IDs for further error handling.
|
||
|
||
Args:
|
||
all_inserts: List of insert objects to process.
|
||
all_positions: List of position objects corresponding to inserts.
|
||
error_collector: ErrorCollector instance to collect errors and warnings.
|
||
|
||
Returns:
|
||
A tuple containing:
|
||
- all_sensors: dict of sensor IDs to sensor attribute dicts
|
||
- all_schaltschrank: dict of Schaltschrankelement IDs to their attribute dicts
|
||
"""
|
||
all_sensors = dict()
|
||
all_cables = dict()
|
||
all_schaltschrank = dict()
|
||
all_unknowns = list()
|
||
|
||
wp = CompareBuffer()
|
||
|
||
for insert, pos in zip(all_inserts, all_positions):
|
||
ld, id_, typ = get_attributes_of_insert(insert, pos)
|
||
|
||
if typ == "Sensor":
|
||
# wenn NAME enthalten ist, z.B. bei Erdungslayouts, dann ist das ein eindeutiger Bezeichner
|
||
if "NAME" in ld:
|
||
all_sensors[id_] = ld
|
||
# neuer Block der alle nötigen Infos enthält
|
||
elif "IO" in ld and "ARTINR" in ld :
|
||
all_sensors[id_] = ld
|
||
# alle anderen Sachen werden dann aus mehreren Rahmen zusammen gesammelt
|
||
else:
|
||
wp.add_block(id_, ld)
|
||
|
||
elif typ == "Kabel":
|
||
all_cables[id_] = ld
|
||
elif typ == "Schaltschrankelement":
|
||
all_schaltschrank[id_] = ld
|
||
else:
|
||
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(all_sensors, wp, error_collector)
|
||
|
||
set_single_frames_with_unique_sps(all_schaltschrank, wp)
|
||
|
||
# die noch übrigen Blöcke melden
|
||
get_errors_double_and_attributes(wp, error_collector)
|
||
|
||
return all_sensors, all_schaltschrank
|
||
|
||
def get_errors_double_and_attributes(wp: CompareBuffer, error_collector: ErrorCollector = None):
|
||
"""
|
||
Analyze the CompareBuffer for blocks that could not be merged or assigned.
|
||
|
||
This function inspects the CompareBuffer for:
|
||
- IDs with only a single associated block, which likely indicates missing or incomplete attributes.
|
||
- IDs with multiple associated blocks, which may indicate duplicate or ambiguous entries.
|
||
|
||
Args:
|
||
wp: CompareBuffer to analyze
|
||
error_collector: ErrorCollector instance to collect errors and warnings
|
||
"""
|
||
missing_attribs = dict()
|
||
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])
|
||
# alle anderen Angaben sind mindestens zwei oder mehr Blöcke mit derselben Id
|
||
else:
|
||
for block in blocks:
|
||
if id_ not in double_ids:
|
||
double_ids[id_] = []
|
||
double_ids[id_].append(block['pos'])
|
||
|
||
# Füge Fehler und Warnungen zum ErrorCollector hinzu, falls vorhanden
|
||
if error_collector:
|
||
if double_ids:
|
||
error_collector.add_errors({"double_ids": double_ids})
|
||
if missing_attribs:
|
||
error_collector.add_warnings({"missing_attributes": missing_attribs})
|
||
|
||
def set_single_frames_with_unique_sps(all_sensors: dict, wp: CompareBuffer):
|
||
all_sensors_ids = wp.get_block_ids()
|
||
for id_ in all_sensors_ids:
|
||
blks_sps = wp.get_all_sps_blocks(id_)
|
||
|
||
for block_with_sps in blks_sps:
|
||
# schaue ob es keinen Konflikt zu bestehenden Angaben
|
||
if "IO" in block_with_sps and "SPS" in block_with_sps:
|
||
sps_praefix = block_with_sps["SPS"]
|
||
new_id = f"{id_}@{sps_praefix}"
|
||
|
||
all_sensors[new_id] = block_with_sps
|
||
wp.remove_block(id_, block_with_sps)
|
||
|
||
def allocate_blocks_together(all_sensors: dict, wp: CompareBuffer, error_collector: ErrorCollector = None) -> None:
|
||
"""
|
||
Merge sensor blocks with the same ID that are split across multiple DXF blocks.
|
||
|
||
This function iterates over all sensor IDs stored in the CompareBuffer. For each ID, it looks for blocks
|
||
with and without an "SPS" prefix. If two such blocks have positions that are close to each other (within a
|
||
specified tolerance), they are considered to represent the same physical sensor. The function then merges
|
||
their information into a single dictionary, creates a new unique sensor ID by appending the SPS prefix,
|
||
and adds the merged sensor to the all_sensors dictionary. The merged blocks are then removed from the buffer.
|
||
|
||
Args:
|
||
all_sensors (dict): Dictionary to store merged sensor information, keyed by unique sensor IDs.
|
||
wp (CompareBuffer): Buffer containing blocks that could not be directly assigned, grouped by ID.
|
||
"""
|
||
#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):
|
||
new_id = create_new_id(id_, block_with_sps, block_without_sps, error_collector) # 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_: str, dict1: dict, dict2: dict, error_collector: ErrorCollector = None) -> str:
|
||
sps_praefix = None
|
||
if "SPS" in dict1:
|
||
sps_praefix = dict1["SPS"]
|
||
if "SPS" in dict2:
|
||
sps_praefix = dict2["SPS"]
|
||
if not sps_praefix:
|
||
if error_collector:
|
||
error_collector.add_errors({"missing_sps_prefix": {id_: "SPS Präfix fehlt für Block-ID"}})
|
||
return f"{id_}@UNKNOWN" # Fallback für fehlenden SPS-Präfix
|
||
return f"{id_}@{sps_praefix}"
|
||
|
||
def create_mappings(positions: dict) -> tuple[dict, dict]:
|
||
unterverteiler_pfad = ""
|
||
dnamen = dict()
|
||
# s
|
||
sensor2unterverteiler = dict()
|
||
warnings = dict()
|
||
for sensorname,v in positions.items():
|
||
if "KENNZEICHNUNG" not in v:
|
||
warnings[sensorname] = "keine KENNZEICHNUNG vorhanden"
|
||
continue
|
||
|
||
unterverteiler_pfad = v["KENNZEICHNUNG"]
|
||
#print(unterverteiler_pfad)
|
||
|
||
# Pfad zur Karte splitten. Dieser hat z.B. den Inhalt "=AH01+UH02-KF1FDI7"
|
||
matches = re.findall(r'[^\-+=]+', unterverteiler_pfad.lstrip('='))
|
||
if matches:
|
||
if len(matches) == 3:
|
||
anlage = matches[0]
|
||
verteiler = matches[1]
|
||
karte = matches[2]
|
||
else:
|
||
warnings[sensorname] = f"Ungültiger Pfad in Kennzeichnung: {unterverteiler_pfad}: +, - oder = fehlen an entsprechender Stelle, keine drei Teile sichtbar"
|
||
continue
|
||
else:
|
||
warnings[sensorname] = f"Ungültiger Pfad in Kennzeichnung: {unterverteiler_pfad}"
|
||
continue
|
||
|
||
if verteiler not in dnamen:
|
||
dnamen[verteiler] = True
|
||
|
||
sensor2unterverteiler[sensorname] = verteiler
|
||
|
||
# jetzt zu jedem Unterverteiler die zugehörigen Sensoren merken
|
||
uv2sensor = dict()
|
||
for sensorname,verteiler in sensor2unterverteiler.items():
|
||
if verteiler not in uv2sensor:
|
||
uv2sensor[verteiler] = list()
|
||
uv2sensor[verteiler].append(sensorname)
|
||
|
||
return (uv2sensor, warnings)
|
||
|
||
def get_subdistributor_position_of_symbol(d_insert: dict, d_pos: dict) -> tuple[dict, str]:
|
||
"""
|
||
Diese Funktion schaut nach den aktuell definierten Attributen in allen Unterverteiler Blöcken
|
||
Sie müssen auch UH heissen
|
||
|
||
"""
|
||
id_ = ""
|
||
ld = d_insert
|
||
|
||
# die neueren Blöcke haben einen Namen
|
||
if "NAME" in d_insert:
|
||
id_ = d_insert["NAME"]
|
||
pos = d_pos["NAME"]
|
||
ld["pos"] = (pos[0], pos[1])
|
||
|
||
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")
|
||
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_
|
||
return None, None
|
||
|
||
|
||
def get_tunnel_position_of_symbol(d_insert: dict, d_pos: dict) -> tuple[dict, str]:
|
||
"""
|
||
Diese Funktion schaut nach den aktuell definierten Attributen in allen Tunnel Blöcken
|
||
Tunnel haben einen Namen der den Tunnel-Mustern aus BMK.cfg entspricht
|
||
|
||
"""
|
||
id_ = ""
|
||
ld = d_insert
|
||
|
||
# Tunnel haben einen Namen
|
||
if "NAME" in d_insert:
|
||
id_ = d_insert["NAME"]
|
||
pos = d_pos["NAME"]
|
||
ld["pos"] = (pos[0], pos[1])
|
||
|
||
# Länge des Tunnels aus dem LAENGE Attribut extrahieren
|
||
if "LAENGE" in d_insert:
|
||
ld["laenge"] = d_insert["LAENGE"]
|
||
|
||
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")
|
||
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_
|
||
return None, None
|
||
|
||
def get_subdistributor_positions_from_symbols(all_inserts:list, all_positions:list, dist2sensors: dict) -> dict:
|
||
"""Ermittelt Unterverteiler-Positionen aus allen Symbolen
|
||
"""
|
||
ret = {}
|
||
all_distributors = dist2sensors.keys()
|
||
for insert, pos in zip(all_inserts, all_positions):
|
||
if "NAME" not in insert: # Unterverteiler haben immer einen eindeutigen Namen
|
||
continue
|
||
ld, id_ = get_subdistributor_position_of_symbol(insert, pos)
|
||
|
||
# Check if id_ matches Cabinet-Pattern from BMK.cfg
|
||
(matches, res) = matches_cabinet_pattern(id_)
|
||
if not matches:
|
||
continue
|
||
|
||
if res in dist2sensors:
|
||
ret[res] = ld["pos"]
|
||
|
||
return ret
|
||
|
||
|
||
def get_tunnel_positions_from_symbols(all_inserts: list, all_positions: list, error_collector: ErrorCollector = None) -> dict:
|
||
"""Ermittelt Tunnel-Positionen aus allen Symbolen
|
||
|
||
Args:
|
||
all_inserts: Liste von Dictionaries mit Attribut-Tags und deren Textwerten
|
||
all_positions: Liste von Dictionaries mit Attribut-Tags und deren (x, y, z)-Positionen
|
||
error_collector: ErrorCollector instance to collect errors and warnings
|
||
|
||
Returns:
|
||
Dictionary mit Tunnelnamen als Keys und Listen von Positionen als Values
|
||
Format: {'Tunnel1': [(x1, y1), (x2, y2)], ..., 'length': {'Tunnel1': 'laenge_wert', ...}}
|
||
"""
|
||
ret = {}
|
||
tunnel_length = {}
|
||
tunnel_missing_length = {}
|
||
|
||
for insert, pos in zip(all_inserts, all_positions):
|
||
if "NAME" not in insert: # Tunnel haben immer einen eindeutigen Namen
|
||
continue
|
||
|
||
ld, id_ = get_tunnel_position_of_symbol(insert, pos)
|
||
if not ld or not id_:
|
||
continue
|
||
|
||
# Check if id_ matches Tunnel-Pattern from BMK.cfg
|
||
if not matches_tunnel_pattern(id_):
|
||
continue
|
||
|
||
# Sammle alle Positionen für den gleichen Tunnelnamen
|
||
if id_ not in ret:
|
||
ret[id_] = []
|
||
ret[id_].append(ld["pos"])
|
||
|
||
# Sammle Längeninformation falls vorhanden, sonst Fehler melden
|
||
if "laenge" in ld and ld["laenge"]:
|
||
tunnel_length[id_] = ld["laenge"]
|
||
else:
|
||
error_msg = f"Tunnel '{id_}' hat keine LAENGE-Angabe. Bitte LAENGE-Attribut im DXF-Symbol setzen."
|
||
print(f"FEHLER: {error_msg}")
|
||
tunnel_missing_length[id_] = error_msg
|
||
|
||
# Fehlende LAENGE als Fehler melden (stoppt die Verarbeitung)
|
||
if error_collector and tunnel_missing_length:
|
||
error_collector.add_errors({"tunnel_missing_length": tunnel_missing_length})
|
||
|
||
# Füge Längeninformation hinzu, falls Tunnel gefunden wurden
|
||
if len(tunnel_length) > 0:
|
||
ret['length'] = tunnel_length
|
||
|
||
return ret
|
||
|
||
|
||
def get_subdistributor_positions_from_entities(entities, dist2sensors: dict) -> dict:
|
||
"""Ermittelt Unterverteiler-Positionen aus einer beliebigen Entity-Iterable.
|
||
|
||
Erwartet eine Iterable von DXF-Entities (z. B. aus `msp.query("MTEXT")`)
|
||
und filtert nach den in der Config erlaubten Layern. Es werden beide bisher
|
||
verwendeten Suchmuster in der MTEXT-Zeile unterstützt ("-<distname>" und
|
||
"+<distname>").
|
||
"""
|
||
ret = {}
|
||
all_distributors = dist2sensors.keys()
|
||
allowed_layers = {layer for (layer, _) in config.items('GetPos-Layer_Distributors')}
|
||
|
||
for entity in entities:
|
||
if entity.dxftype() != "MTEXT":
|
||
continue
|
||
layer = entity.dxf.layer
|
||
if layer not in allowed_layers:
|
||
continue
|
||
text = entity.dxf.text
|
||
insert_point = entity.dxf.insert
|
||
|
||
for distname in all_distributors:
|
||
if f"-{distname}" in text or f"+{distname}" in text:
|
||
ret[distname] = (round(insert_point[0], 1), round(insert_point[1], 1))
|
||
|
||
return ret
|
||
|
||
def get_tunnel_positions_from_entities(entities) -> dict:
|
||
"""Ermittelt Tunnel-Ein/Ausgangs-Positionen aus einer beliebigen Entity-Iterable.
|
||
|
||
Erwartet eine Iterable von DXF-Entities (z. B. aus `msp.query("MTEXT")`)
|
||
und filtert nach den in der Config erlaubten Layern. Erkennt Tunnel anhand
|
||
des Musters "TUNNEL<nr>-<laenge>" und sammelt pro Tunnelname die gefundenen
|
||
Positionen sowie die Länge.
|
||
"""
|
||
all_tunnels = dict()
|
||
tunnel_length = dict()
|
||
allowed_layers = {layer for (layer, _) in config.items('GetPos-Layer_Tunnel')}
|
||
|
||
for entity in entities:
|
||
if entity.dxftype() != "MTEXT":
|
||
continue
|
||
if entity.dxf.layer not in allowed_layers:
|
||
continue
|
||
|
||
txt = entity.dxf.text
|
||
insert = entity.dxf.insert
|
||
|
||
pattern = r"(TUNNEL\d+)-(\d+)"
|
||
match = re.search(pattern, txt)
|
||
if not match:
|
||
continue
|
||
|
||
tunnelname = match.group(1)
|
||
laenge = match.group(2)
|
||
pos = (round(insert[0], 1), round(insert[1], 1))
|
||
|
||
if tunnelname not in all_tunnels:
|
||
all_tunnels[tunnelname] = []
|
||
all_tunnels[tunnelname].append(pos)
|
||
tunnel_length[tunnelname] = laenge
|
||
|
||
if len(tunnel_length.keys()) > 0:
|
||
all_tunnels['length'] = tunnel_length
|
||
return all_tunnels
|
||
|
||
|
||
def get_rack_positions(msp) -> dict:
|
||
"""Hole alle Positionen aller Kabelpritschen und nummeriere Racks."""
|
||
ret = dict()
|
||
rack_counter = 1
|
||
all_layers = list(config.items('GetPos-Layer_Racks'))
|
||
|
||
for layer, _ in all_layers:
|
||
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
|
||
|
||
pl_query = f'POLYLINE[layer=="{layer}"]'
|
||
for entity in msp.query(pl_query):
|
||
rack_key = f"Rack_{rack_counter}"
|
||
handle_polyline(entity, rack_key, ret)
|
||
rack_counter += 1
|
||
|
||
return 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] = []
|
||
for point in entity.vertices():
|
||
x, y, *_ = point
|
||
ret[rack_key].append([round(x, 1), round(y, 1), round(z, 1)])
|
||
|
||
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:
|
||
x = vertex.dxf.location.x
|
||
y = vertex.dxf.location.y
|
||
z = vertex.dxf.location.z
|
||
ret[rack_key].append([round(x, 1), round(y, 1), round(z, 1)])
|
||
|
||
|
||
def check_rack_z_coordinates(res_racks: dict, error_collector, config) -> None:
|
||
"""
|
||
Prüft die z-Koordinaten aller Racks und gibt eine Warnung aus,
|
||
wenn die Differenz zwischen min und max größer als der konfigurierte Schwellwert ist.
|
||
|
||
Args:
|
||
res_racks: Dictionary mit Rack-Daten (key: rack_name, value: list of coordinates)
|
||
error_collector: ErrorCollector Instanz zum Hinzufügen von Warnungen
|
||
config: ConfigParser Objekt mit allgemein.cfg
|
||
"""
|
||
if not res_racks:
|
||
return
|
||
|
||
# Lese Schwellwert aus Config (Standard: 2000.0 mm)
|
||
max_height_diff = config.getfloat("Racks", "MaximalTotalHeightDifferences", fallback=2000.0)
|
||
|
||
all_z_coords = []
|
||
|
||
# Sammle alle z-Koordinaten aus allen Racks
|
||
for rack_name, coordinates in res_racks.items():
|
||
if isinstance(coordinates, list):
|
||
for coord in coordinates:
|
||
if isinstance(coord, (list, tuple)) and len(coord) >= 3:
|
||
all_z_coords.append(coord[2])
|
||
elif isinstance(coord, dict) and 'z' in coord:
|
||
all_z_coords.append(coord['z'])
|
||
|
||
if not all_z_coords:
|
||
return
|
||
|
||
# Berechne Min und Max
|
||
min_z = min(all_z_coords)
|
||
max_z = max(all_z_coords)
|
||
diff = max_z - min_z
|
||
|
||
# Prüfe, ob Differenz > Schwellwert
|
||
if diff > max_height_diff:
|
||
warning_msg = f"The z-coordinates of the racks differ strong between each other from {min_z:.1f} to {max_z:.1f}"
|
||
error_collector.add_warnings({"rack_z_coordinate_deviation": warning_msg})
|
||
#print(f"WARNING: {warning_msg}")
|
||
|
||
def check_existance(res_mappings: dict, res_dist: dict, res_pos: dict, res_tunnel: dict) -> dict:
|
||
ret = dict()
|
||
|
||
for dname in res_mappings.keys():
|
||
if dname not in res_dist:
|
||
if "missing_distributors" not in ret:
|
||
ret["missing_distributors"] = list()
|
||
ret["missing_distributors"].append(dname)
|
||
for sname, lofsensors in res_mappings.items():
|
||
for s in lofsensors:
|
||
if s not in res_pos:
|
||
if "missing_sensors" not in ret:
|
||
ret["missing_sensors"] = list()
|
||
ret['missing_sensors'].append(s)
|
||
for tname in res_tunnel.keys():
|
||
if tname == "length": # dict enthält auch die Länge aller Tunnels in einem Key
|
||
continue
|
||
if len(res_tunnel[tname]) < 2 :
|
||
if "missing_tunnel" not in ret:
|
||
ret["missing_tunnel"] = list()
|
||
ret["missing_tunnel"].append(tname)
|
||
if len(res_tunnel[tname]) > 2 :
|
||
if "overdefined_tunnel" not in ret:
|
||
ret["overdefined_tunnel"] = list()
|
||
ret["overdefined_tunnel"].append(tname)
|
||
return ret
|
||
|
||
def validate_configs() -> None:
|
||
errors = []
|
||
|
||
print("\nValidating given configs: Checking for inconsistency.")
|
||
|
||
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'")
|
||
|
||
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(",")]
|
||
for section in sections:
|
||
if not config_cables.has_section(section):
|
||
errors.append(f"Cable-Section '{section}' from Cable-Mapping ({mapping_key}) missing in kabel.cfg")
|
||
|
||
if config_BMK.has_section("Length-Adjustments"):
|
||
for prefix, value in config_BMK.items("Length-Adjustments"):
|
||
try:
|
||
f = float(value)
|
||
if f < 0:
|
||
errors.append(f"Negative Value in Length-Adjustments for {prefix}: {value}")
|
||
except ValueError:
|
||
errors.append(f"Invalid Value in Length-Adjustments for {prefix}: {value}")
|
||
|
||
if errors:
|
||
print("Inconsistencies found:")
|
||
for e in errors:
|
||
print(f"- {e}")
|
||
print("\ncontinuing with routing process")
|
||
else:
|
||
print("No inconsistencies found. Continuing with routing process.")
|
||
|
||
if __name__ == '__main__':
|
||
parser = argparse.ArgumentParser(description='fetches the x/y positions from a dxf file', prog='getpositions')
|
||
parser.add_argument('-f', '--filename', action='store', required=True, default="ST_6300_Steuerungstestlayout1_neueBloecke.dwg", help='which file should be fetched', metavar='myfile.dxf')
|
||
parser.add_argument('-s', '--sensors', action='store_true', help='fetch all position of sensors, motors, actors and subdistributors')
|
||
parser.add_argument('-r', '--rack', action='store_true', help='fetch all positions of all cable racks')
|
||
parser.add_argument('-w', '--write', action='store', help='write results into a json file')
|
||
parser.add_argument('-c', '--console', action='store_true', help='print results to output')
|
||
parser.add_argument('-e', '--errors', action='store', help='write an error file in case of double defined items in layout')
|
||
parser.add_argument('-n', '--scan', action='store_true', help='print all layer of racs, distributes and equiment not empty')
|
||
|
||
args = parser.parse_args()
|
||
|
||
out_dir = check_environment_var('PROJECT_DATA')
|
||
work_dir = check_environment_var('PROJECT_WORK')
|
||
config_dir = check_environment_var("PROJECT_CFG")
|
||
|
||
filename = Path(args.filename)
|
||
if not filename.suffix == ".dxf":
|
||
print("only available for .dxf files")
|
||
exit()
|
||
|
||
(dxf_path, dexists) = check_file_in_work(work_dir, filename)
|
||
if dexists == False:
|
||
print("no such file ")
|
||
parser.print_help()
|
||
exit()
|
||
|
||
doc = get_dxf_file(dxf_path)
|
||
msp = doc.modelspace()
|
||
|
||
res_sens = dict()
|
||
res_dist = dict()
|
||
res_rac = dict()
|
||
res_mappings = dict()
|
||
|
||
if args.sensors or args.dists or args.rack:
|
||
|
||
# Allgemeine Config Laden
|
||
config = configparser.ConfigParser(allow_no_value=True, delimiters=("="))
|
||
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 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 optionstr: optionstr
|
||
config_cables.read(os.path.join(config_dir, "kabel.cfg"))
|
||
|
||
validate_configs()
|
||
|
||
output_results = dict()
|
||
|
||
# ErrorCollector für alle Fehler und Warnungen initialisieren
|
||
error_collector = ErrorCollector()
|
||
|
||
if args.sensors:
|
||
# Sensoren auslesen aus den Symbolen
|
||
|
||
all_inserts, all_positions = attribs_to_dicts(msp)
|
||
res_sens, res_schaltschrank_elemente = extract_input_positions(all_inserts, all_positions, error_collector)
|
||
|
||
output_results['sensors'] = res_sens
|
||
output_results['schaltschrank_elemente'] = res_schaltschrank_elemente
|
||
|
||
if args.console:
|
||
print(to_json(res_sens))
|
||
|
||
# Mapping zu Sensoren auslesen
|
||
(res_mappings, warnings) = create_mappings(res_sens)
|
||
output_results['mappings'] = res_mappings
|
||
# Mapping-Warnungen zum ErrorCollector hinzufügen
|
||
if warnings:
|
||
error_collector.add_warnings({"mapping_warnings": warnings})
|
||
if args.console:
|
||
print(to_json(res_mappings))
|
||
|
||
# Distributoren auslesen (generisch über Entities)
|
||
entities = msp.query('MTEXT')
|
||
# die Infos aus den Texten (alter Stil)
|
||
res_dist = get_subdistributor_positions_from_entities(entities, res_mappings)
|
||
# die Infos aus den Blöcken
|
||
symbol_dists = get_subdistributor_positions_from_symbols(all_inserts, all_positions, res_mappings)
|
||
res_dist = merge_two_dicts(res_dist, symbol_dists)
|
||
output_results['distributors'] = res_dist
|
||
if args.console:
|
||
print(to_json(res_dist))
|
||
|
||
# Tunnel auslesen (generisch über Entities)
|
||
t_entities = msp.query('MTEXT')
|
||
# die Infos aus den Texten (alter Stil)
|
||
res_tunnel = get_tunnel_positions_from_entities(t_entities)
|
||
# die Infos aus den Blöcken (neuer Stil)
|
||
symbol_tunnels = get_tunnel_positions_from_symbols(all_inserts, all_positions, error_collector)
|
||
res_tunnel = merge_two_dicts(res_tunnel, symbol_tunnels)
|
||
output_results['tunnels'] = res_tunnel
|
||
if args.console:
|
||
print(to_json(res_tunnel))
|
||
|
||
if args.rack:
|
||
res_rac = get_rack_positions(msp)
|
||
|
||
output_results['racks'] = res_rac
|
||
|
||
# Prüfe z-Koordinaten der Racks auf starke Abweichungen
|
||
check_rack_z_coordinates(res_rac, error_collector, config)
|
||
|
||
if args.console:
|
||
print(to_json(res_rac))
|
||
|
||
|
||
if args.write:
|
||
basename = os.path.splitext(args.write)[0]
|
||
|
||
res_not_found = check_existance(res_mappings, res_dist, res_sens, res_tunnel)
|
||
|
||
# Alle weiteren Fehler zum ErrorCollector hinzufügen
|
||
error_collector.add_errors(res_not_found)
|
||
|
||
# Warnings immer in output_results schreiben (auch wenn leer)
|
||
output_results["warnings"] = error_collector.warnings if error_collector.warnings else {}
|
||
|
||
# ErrorCollector schreibt Datei nur wenn Fehler/Warnungen vorhanden
|
||
if args.errors:
|
||
error_collector.write_errorfile(work_dir, args.errors)
|
||
|
||
# Für Kompatibilität mit Ausgabe-JSON: Fehler und Warnungen aus ErrorCollector holen
|
||
all_issues = error_collector.get_all_issues()
|
||
if "errors" in all_issues:
|
||
output_results["not_found"] = all_issues["errors"].get("not_found", {})
|
||
if "double_ids" in all_issues["errors"]:
|
||
output_results["double_ids"] = all_issues["errors"]["double_ids"]
|
||
|
||
|
||
write_results(to_json(output_results), work_dir, f"{basename}.json")
|
||
|
||
else:
|
||
parser.print_help()
|