822 lines
31 KiB
Python
822 lines
31 KiB
Python
import argparse
|
||
import configparser
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
import ezdxf
|
||
from ezdxf.addons import iterdxf
|
||
from shapely.geometry import Point
|
||
from ezdxf.lldxf.const import DXFStructureError
|
||
|
||
|
||
"""
|
||
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 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)
|
||
with open(outfile, 'w', encoding='utf-8') as fh:
|
||
fh.write(jsn_results)
|
||
print("done")
|
||
|
||
|
||
def merge_two_dicts(x: dict, y: dict) -> dict:
|
||
z = x.copy()
|
||
z.update(y)
|
||
return z
|
||
|
||
|
||
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 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(insert_iterable) -> tuple[dict, dict, 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:
|
||
insert_iterable: An iterable of insert objects to process.
|
||
|
||
Returns:
|
||
A tuple containing:
|
||
- all_sensors: dict of sensor IDs to sensor attribute dicts
|
||
- all_schaltschrank: dict of Schaltschrankelement IDs to their attribute dicts
|
||
- double_ids: dict of IDs with multiple associated blocks (potential duplicates)
|
||
- missing_attribs: dict of IDs with missing or incomplete attributes
|
||
"""
|
||
all_sensors = dict()
|
||
all_cables = dict()
|
||
all_schaltschrank = dict()
|
||
all_unknowns = list()
|
||
|
||
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)
|
||
|
||
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)
|
||
|
||
set_single_frames_with_unique_sps(all_schaltschrank, wp)
|
||
|
||
# die noch übrigen Blöcke melden
|
||
missing_attribs, double_ids = get_errors_double_and_attributes(wp)
|
||
|
||
return all_sensors, all_schaltschrank, double_ids, missing_attribs
|
||
|
||
def get_errors_double_and_attributes(wp: CompareBuffer) -> tuple[dict, dict]:
|
||
"""
|
||
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.
|
||
|
||
Returns:
|
||
missing_attribs: dict mapping IDs to a message about missing or incomplete attributes.
|
||
double_ids: dict mapping IDs to a list of positions for blocks with duplicate IDs.
|
||
"""
|
||
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'])
|
||
return missing_attribs, double_ids
|
||
|
||
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) -> 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) # 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) -> str:
|
||
sps_praefix = None
|
||
if "SPS" in dict1:
|
||
sps_praefix = dict1["SPS"]
|
||
if "SPS" in dict2:
|
||
sps_praefix = dict2["SPS"]
|
||
if not sps_praefix:
|
||
raise Exception
|
||
return f"{id_}@{sps_praefix}"
|
||
|
||
def attribs_to_dicts(insert_iterable) -> tuple[list, list]:
|
||
"""
|
||
Wandelt eine Iterable von INSERT-Objekten in zwei Listen um:
|
||
- 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 (jeweils gerundet auf eine Nachkommastelle)
|
||
|
||
Jeder Eintrag in den Listen entspricht einem INSERT-Block mit Attributen.
|
||
Blöcke ohne Attribute werden übersprungen.
|
||
"""
|
||
all_inserts = list()
|
||
all_positions = list()
|
||
for insert in insert_iterable:
|
||
if insert.dxftype() != 'INSERT':
|
||
continue
|
||
itemdata = dict()
|
||
positions = dict()
|
||
typ = 'unknown'
|
||
for attrib in insert.attribs:
|
||
if len(insert.attribs) == 0:
|
||
continue # Überspringe Blöcke ohne Attribute
|
||
attr_tag = attrib.dxf.tag
|
||
attr_text = attrib.dxf.text
|
||
pos = attrib.dxf.insert
|
||
itemdata[attr_tag] = attr_text
|
||
positions[attr_tag] = (round(pos.x, 1), round(pos.y, 1), round(pos.z, 1))
|
||
if len(itemdata) > 0:
|
||
all_inserts.append(itemdata)
|
||
all_positions.append(positions)
|
||
return all_inserts, all_positions
|
||
|
||
def get_input_positions(msp) -> tuple[dict, dict, dict, dict]:
|
||
return extract_input_positions(msp.query('INSERT'))
|
||
|
||
def get_input_positions_iter(dxf_path) -> tuple[dict, dict, dict, dict]:
|
||
return extract_input_positions(iterdxf.modelspace(dxf_path))
|
||
|
||
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"
|
||
|
||
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_positions(msp, dist2sensors: dict) -> dict:
|
||
"""Hole alle Positionen der Unterverteiler !!UV-Positionen bereits 'Mitte-Mitte'!!"""
|
||
ret = dict()
|
||
all_distributors = dist2sensors.keys()
|
||
all_layers = config.items('GetPos-Layer_Distributors')
|
||
for (layer, v) in all_layers:
|
||
for distname in all_distributors:
|
||
selectstr = f'MTEXT[layer=="{layer}"]'
|
||
for text in msp.query(selectstr):
|
||
match = re.search("-" + distname, text.dxf.text)
|
||
if match:
|
||
ret[distname] = (round(text.dxf.insert[0], 1), round(text.dxf.insert[1], 1))
|
||
return ret
|
||
|
||
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()
|
||
all_layers = config.items('GetPos-Layer_Distributors')
|
||
|
||
for entity in iterdxf.modelspace(dxf_path):
|
||
if entity.dxftype() != "MTEXT":
|
||
continue
|
||
entity_text = entity.dxf.text
|
||
entity_layer = entity.dxf.layer
|
||
insert_point = entity.dxf.insert
|
||
|
||
for (layer_name, _) in all_layers:
|
||
if entity_layer != layer_name:
|
||
continue
|
||
for distname in all_distributors:
|
||
if f"+{distname}" in entity_text:
|
||
ret[distname] = (round(insert_point[0], 1), round(insert_point[1], 1))
|
||
|
||
return ret
|
||
|
||
def get_tunnel_positions(msp) -> dict:
|
||
"""Hole alle Positionen aller Tunnel Ein und Ausgänge."""
|
||
all_tunnels = dict()
|
||
tunnel_length = dict()
|
||
all_layers = config.items('GetPos-Layer_Tunnel')
|
||
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))
|
||
tunnelname = match.group(1)
|
||
laenge = match.group(2)
|
||
tunnel_length[tunnelname] = laenge
|
||
if tunnelname not in all_tunnels:
|
||
all_tunnels[tunnelname] = list()
|
||
all_tunnels[tunnelname].append(pos)
|
||
else:
|
||
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) -> dict:
|
||
"""Hole alle Positionen aller Tunnel Ein- und Ausgänge mithilfe von iterdxf."""
|
||
all_tunnels = dict()
|
||
tunnel_length = dict()
|
||
all_layers = config.items('GetPos-Layer_Tunnel')
|
||
|
||
for entity in iterdxf.modelspace(dxf_path):
|
||
if entity.dxftype() != "MTEXT":
|
||
continue
|
||
|
||
txt = entity.dxf.text
|
||
layer = entity.dxf.layer
|
||
insert = entity.dxf.insert
|
||
|
||
for (layer_name, _) in all_layers:
|
||
if layer != layer_name:
|
||
continue
|
||
|
||
pattern = r"(TUNNEL\d+)-(\d+)"
|
||
match = re.search(pattern, txt)
|
||
if match:
|
||
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
|
||
|
||
# helper function
|
||
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) -> 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) -> 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 get_rack_positions_iter(dxf_path) -> dict:
|
||
"""Hole alle Positionen aller Kabelpritschen (Racks) mithilfe von iterdxf."""
|
||
ret = dict()
|
||
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
|
||
|
||
rack_key = f"Rack_{rack_counter}"
|
||
|
||
if entity.dxftype() == "LWPOLYLINE":
|
||
handle_lwpolyline(entity, rack_key, ret)
|
||
elif entity.dxftype() == "POLYLINE":
|
||
handle_polyline(entity, rack_key, ret)
|
||
else:
|
||
continue
|
||
|
||
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 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"))
|
||
ret = dict()
|
||
ret['all_layers'] = layer_names_inside
|
||
ret['used_blocks'] = used_block_names
|
||
ret['all_blocks'] = alle_block_defs
|
||
return ret
|
||
|
||
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: Path):
|
||
"""Hole das dxf file."""
|
||
try:
|
||
print("reading file ..", end='')
|
||
doc = ezdxf.filemanagement.readfile(filepath)
|
||
print("done")
|
||
except IOError:
|
||
print("Not a DXF file or a generic I/O error.")
|
||
sys.exit(1)
|
||
except DXFStructureError:
|
||
print("Invalid or corrupted DXF file.")
|
||
sys.exit(2)
|
||
return doc
|
||
|
||
def check_file_in_work(work_dir: Path, filename: Path) -> tuple[Path, bool]:
|
||
fexists = True
|
||
if not filename.exists():
|
||
mypath = work_dir.joinpath(filename)
|
||
ex = mypath.exists()
|
||
if not mypath.exists():
|
||
fexists = False
|
||
else:
|
||
mypath = filename
|
||
return mypath, fexists
|
||
|
||
def check_existance(res_mappings: dict, res_dist: dict, res_pos: dict) -> dict:
|
||
ret = dict()
|
||
ret["missing_distributors"] = list()
|
||
ret["missing_sensors"] = list()
|
||
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 s in lofsensors:
|
||
if s not in res_pos:
|
||
ret['missing_sensors'].append(s)
|
||
|
||
return ret
|
||
|
||
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() -> 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.")
|
||
|
||
def check_environment_var(env_str: str) -> Path:
|
||
out_path = os.environ.get(env_str)
|
||
if out_path:
|
||
return Path(out_path)
|
||
else:
|
||
print(f"Umgebungsvariable {env_str} ist nicht gesetzt oder leer.")
|
||
exit()
|
||
|
||
def get_input_positions_combined(source: str, use_iter: bool) -> tuple[dict, dict, dict, dict]:
|
||
if use_iter:
|
||
return get_input_positions_iter(source)
|
||
else:
|
||
return get_input_positions(source)
|
||
|
||
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()
|
||
|
||
if dxf_is_binary(dxf_path): # Wenn dxf eine binary ist, dann komplett parsen und modelspace anlegen
|
||
print("Given .dxf-file is binary dxf. Proceeding to read file. Watch RAM-usage.")
|
||
doc = get_dxf_file(dxf_path)
|
||
msp = doc.modelspace()
|
||
use_iter = False
|
||
else:
|
||
print("Given .dxf-file is ASCII-dxf. Proceeding to use iterative functions. Process may take longer.")
|
||
use_iter = True
|
||
|
||
res_sens = dict()
|
||
res_cables = 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()
|
||
|
||
if args.sensors:
|
||
# Sensoren auslesen
|
||
res_sens, res_schaltschrank_elemente, res_double, missing_attribs = get_input_positions_combined(dxf_path if use_iter else msp, use_iter)
|
||
|
||
if args.errors and len(res_double) > 0:
|
||
print("Duplicate blocks found. Writing errors-file.")
|
||
err_ids = list(res_double.keys())
|
||
write_results(to_json(list(err_ids)), work_dir, args.errors)
|
||
|
||
|
||
output_results['sensors'] = res_sens
|
||
output_results['schaltschrank_elemente'] = res_schaltschrank_elemente
|
||
#output_results['cables'] = res_cables
|
||
|
||
if args.console:
|
||
print(to_json(res_sens))
|
||
|
||
# Mapping zu Sensoren auslesen
|
||
(res_mappings, warnings) = create_mappings(res_sens)
|
||
output_results['mappings'] = res_mappings
|
||
if args.console:
|
||
print(to_json(res_mappings))
|
||
|
||
# Distributoren auslesen
|
||
if use_iter:
|
||
res_dist = get_subdistributor_positions_iter(dxf_path, res_mappings)
|
||
else:
|
||
res_dist = get_subdistributor_positions(msp, res_mappings)
|
||
|
||
output_results['distributors'] = res_dist
|
||
|
||
if args.console:
|
||
print(to_json(res_dist))
|
||
|
||
# Tunnel auslesen
|
||
if use_iter:
|
||
res_tunnel = get_tunnel_positions_iter(dxf_path)
|
||
else:
|
||
res_tunnel = get_tunnel_positions(msp)
|
||
|
||
output_results['tunnels'] = res_tunnel
|
||
|
||
if args.console:
|
||
print(to_json(res_tunnel))
|
||
|
||
if args.rack:
|
||
if use_iter:
|
||
res_rac = get_rack_positions_iter(dxf_path)
|
||
else:
|
||
res_rac = get_rack_positions(msp)
|
||
|
||
output_results['racks'] = res_rac
|
||
|
||
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)
|
||
added_warnings = merge_two_dicts(warnings, missing_attribs)
|
||
res_not_found["missing_attributes"] = added_warnings
|
||
output_results["not_found"] = res_not_found
|
||
output_results["double_ids"] = res_double
|
||
write_results(to_json(output_results), work_dir, f"{basename}.json")
|
||
|
||
else:
|
||
parser.print_help()
|