745 lines
27 KiB
Python
745 lines
27 KiB
Python
import argparse
|
||
import configparser
|
||
import ezdxf
|
||
import os
|
||
import sys
|
||
import json
|
||
import re
|
||
from ezdxf.addons import iterdxf
|
||
import re
|
||
from pathlib import Path
|
||
from shapely.geometry import Point
|
||
|
||
|
||
"""
|
||
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(jsnResults, outdir, filename):
|
||
""" write results to a json file
|
||
"""
|
||
print("writing results to a json file ...")
|
||
outfile = os.path.join(outdir, filename)
|
||
with open(outfile, 'w', encoding='utf-8') as fh:
|
||
fh.write(jsnResults)
|
||
print("done")
|
||
|
||
def merge_two_dicts(x, y):
|
||
z = x.copy()
|
||
z.update(y)
|
||
return z
|
||
|
||
|
||
def get_type_of_name_cfg(name):
|
||
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(dinsert, dpos):
|
||
"""
|
||
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 = dinsert
|
||
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"]
|
||
|
||
if "IO" in dinsert:
|
||
attr_text = dinsert["IO"]
|
||
typ = get_type_of_name_cfg(attr_text)
|
||
id = dinsert["IO"]
|
||
pos = dpos["IO"]
|
||
|
||
if "REALE_POSITION" in dinsert and dinsert["REALE_POSITION"] == 'x':
|
||
pos = dpos["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])
|
||
|
||
if "B" in dinsert:
|
||
attr_text = dinsert["B"]
|
||
typ = get_type_of_name_cfg(attr_text)
|
||
id = attr_text
|
||
#Position aufzeichnen und bei Bedarf später mit REAL_POS überschreiben
|
||
pos = dpos["B"]
|
||
ld["pos"] = (pos[0], pos[1])
|
||
|
||
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"""
|
||
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 get_blocks(self, id):
|
||
if id in self._wartepuffer:
|
||
return self._wartepuffer[id]
|
||
else:
|
||
return []
|
||
|
||
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 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)
|
||
l = list()
|
||
for b in buffers:
|
||
if self.dict_compare(b, toRemove):
|
||
pass
|
||
else:
|
||
l.append(b)
|
||
self.set_block(id, l)
|
||
|
||
def positions_are_close(self, dict1:dict, dict2:dict, border:float):
|
||
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):
|
||
"""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)
|
||
l = list()
|
||
for b in buffers:
|
||
if "SPS" not in b:
|
||
l.append(b)
|
||
return l
|
||
|
||
def extract_input_positions(insert_iterable) -> tuple:
|
||
allSensors = dict()
|
||
allCables = dict()
|
||
allSchaltschrank = dict()
|
||
allunknowns = 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, dann ist das ein eindeutiger Bezeichner
|
||
if "NAME" in ld:
|
||
allSensors[id] = ld
|
||
else:
|
||
# alle anderen Sachen werden dann aus mehreren Rahmen zusammen gesammelt
|
||
WP.add_block(id, ld)
|
||
|
||
elif typ == "Kabel":
|
||
allCables[id] = ld
|
||
elif typ == "Schaltschrankelement":
|
||
allSchaltschrank[id] = ld
|
||
else:
|
||
allunknowns.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)
|
||
|
||
# die noch übrigen Blöcke melden
|
||
missing_attribs, doubleIds = get_errors_double_and_attributes(WP)
|
||
|
||
return allSensors, doubleIds, missing_attribs
|
||
|
||
def get_errors_double_and_attributes(WP):
|
||
missing_attribs = dict()
|
||
doubleIds = 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 doubleIds:
|
||
doubleIds[id] = []
|
||
doubleIds[id].append(block['pos'])
|
||
return missing_attribs, doubleIds
|
||
|
||
def allocate_blocks_together(allSensors: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)
|
||
|
||
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)
|
||
|
||
def create_new_id(id, dict1, dict2):
|
||
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):
|
||
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:
|
||
return extract_input_positions(msp.query('INSERT'))
|
||
|
||
def get_input_positions_iter(dxf_path) -> tuple:
|
||
return extract_input_positions(iterdxf.modelspace(dxf_path))
|
||
|
||
def create_mappings(positions:dict) -> tuple:
|
||
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:
|
||
anlage = matches[0]
|
||
verteiler = matches[1]
|
||
karte = matches[2]
|
||
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):
|
||
"""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 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)
|
||
if match:
|
||
ret[distname] = (round(text.dxf.insert[0],1), round(text.dxf.insert[1],1)) #nur x und y Koordinate in Json schreiben
|
||
return ret
|
||
|
||
def get_subdistributor_positions_iter(dxf_path, dist2sensors):
|
||
"""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):
|
||
"""hole alle Positionen aller Tunnel Ein und Ausgänge
|
||
"""
|
||
allTunnels = 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:
|
||
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
|
||
tunnelname = match.group(1)
|
||
laenge = match.group(2)
|
||
tunnel_length[tunnelname] = laenge
|
||
if not tunnelname in allTunnels:
|
||
allTunnels[tunnelname] = list()
|
||
allTunnels[tunnelname].append(pos)
|
||
else:
|
||
allTunnels[tunnelname].append(pos)
|
||
if len(allTunnels.keys()) > 0:
|
||
allTunnels['length'] = tunnel_length
|
||
return allTunnels
|
||
|
||
def get_tunnel_positions_iter(dxf_path):
|
||
"""Hole alle Positionen aller Tunnel Ein- und Ausgänge mithilfe von iterdxf."""
|
||
allTunnels = 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 allTunnels:
|
||
allTunnels[tunnelname] = []
|
||
allTunnels[tunnelname].append(pos)
|
||
tunnel_length[tunnelname] = laenge
|
||
if len(tunnel_length.keys()) > 0:
|
||
allTunnels['length'] = tunnel_length
|
||
return allTunnels
|
||
|
||
# helper function
|
||
def print_line(e):
|
||
print("LINE on layer: %s\n" % e.dxf.layer)
|
||
print("points: %s\n" % repr(e.dxf))
|
||
|
||
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" 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"""
|
||
ret = dict()
|
||
rack_counter = 1 #Zaehler für Rack Nummerierung
|
||
|
||
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}"
|
||
handle_polyline(entity, rack_key, ret)
|
||
rack_counter += 1
|
||
|
||
return ret
|
||
|
||
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
|
||
|
||
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, 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
|
||
|
||
ret[rack_key] = []
|
||
for point in entity.vertices(): # returns (x, y, start_width, end_width, bulge)
|
||
x, y, *_ = point
|
||
ret[rack_key].append([round(x, 1), round(y, 1), round(z, 1)])
|
||
|
||
def handle_polyline(entity, rack_key, ret):
|
||
"""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:ezdxf.document.Drawing):
|
||
layer_names_inside = dxf_source.layers.entries.keys()
|
||
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, 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
|
||
"""
|
||
try:
|
||
print("reading file ..", end='')
|
||
doc = ezdxf.readfile(filepath)
|
||
print("done")
|
||
except IOError:
|
||
print(f"Not a DXF file or a generic I/O error.")
|
||
sys.exit(1)
|
||
except ezdxf.DXFStructureError:
|
||
print(f"Invalid or corrupted DXF file.")
|
||
sys.exit(2)
|
||
return doc
|
||
|
||
def check_file_in_work(work_dir:Path, filename:Path):
|
||
fexists = True
|
||
if not filename.exists(): # dann schau im Work Ordner nach
|
||
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, res_dist, res_pos):
|
||
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):
|
||
with open(dxf_path, 'rb') as f:
|
||
header = f.read(22)
|
||
return b'AutoCAD Binary DXF' in header
|
||
|
||
def validate_configs():
|
||
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(",")]
|
||
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")
|
||
|
||
# 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:
|
||
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()
|
||
|
||
|
||
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 option: option # 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.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.read(os.path.join(config_dir, "kabel.cfg"))
|
||
|
||
validate_configs()
|
||
|
||
output_results = dict()
|
||
|
||
if args.sensors:
|
||
# Sensoren auslesen
|
||
if use_iter:
|
||
res_sens, res_double, missing_attribs = get_input_positions_iter(dxf_path)
|
||
else:
|
||
res_sens, res_double, missing_attribs = get_input_positions(msp)
|
||
|
||
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['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()
|