655 lines
24 KiB
Python
655 lines
24 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
|
||
|
||
|
||
"""
|
||
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(name):
|
||
|
||
SpecialKeys = ["MB", # Ventil
|
||
"MA", # Motor
|
||
"BG", # Stausensor
|
||
"BP", # Schalter Druckluft
|
||
"QM", # Ventile
|
||
"BX" # Scanner
|
||
]
|
||
|
||
KabelKey = ["WD", "WF"]
|
||
DropKeys = [
|
||
"FC", # Motorschutzschalter
|
||
"PF", # Leuchtmelder
|
||
"DI", # Feedback vom Gerät
|
||
"QA", # Hauptschütz
|
||
"SF" # Drucktaster
|
||
]
|
||
|
||
prefix = name[:2]
|
||
if prefix in SpecialKeys:
|
||
typ = "Sensor"
|
||
|
||
# Suche nach Kabel
|
||
elif prefix in KabelKey:
|
||
typ = "Kabel"
|
||
|
||
# suche nach Items die wir nicht weiter verfolgen
|
||
elif prefix in DropKeys:
|
||
typ = "Schaltschrankelement"
|
||
|
||
else:
|
||
typ = "unknown"
|
||
|
||
return typ
|
||
|
||
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(insert):
|
||
"""
|
||
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 = dict()
|
||
typ = 'unknown'
|
||
for attrib in insert.attribs:
|
||
attr_tag = attrib.dxf.tag
|
||
attr_text = attrib.dxf.text
|
||
if len(insert.attribs) == 0:
|
||
continue # Überspringe Blöcke ohne Attribute
|
||
#print(f"Attribut Name: {attrib.dxf.tag}, Wert: {attrib.dxf.text}")
|
||
ld[attr_tag] = attr_text
|
||
|
||
if attr_tag == "NAME": # die neueren Bläcke heissen nicht IO, sondern haben einen Namen
|
||
typ = get_type_of_name_cfg(attr_text)
|
||
id = attr_text
|
||
# pos = attrib.dxf.insert #Position aufzeichnen und bei Bedarf später mit REAL_POS überschreiben
|
||
# ld["pos"] = (round(pos.x, 1), round(pos.y, 1))
|
||
if attr_tag == "IO":
|
||
typ = get_type_of_name_cfg(attr_text)
|
||
id = attr_text
|
||
#print(f"-- coord io {id}--: {attrib.dxf.insert}") # position des Blocks
|
||
pos = attrib.dxf.insert #Position aufzeichnen und bei Bedarf später mit REAL_POS überschreiben
|
||
ld["pos"] = (round(pos.x, 1), round(pos.y, 1))
|
||
|
||
if attr_tag == "B":
|
||
# Suche nach Sensoren
|
||
typ = get_type_of_name_cfg(attr_text)
|
||
id = attr_text
|
||
pos = attrib.dxf.insert #Position aufzeichnen und bei Bedarf später mit REAL_POS überschreiben
|
||
ld["pos"] = (round(pos.x, 1), round(pos.y, 1))
|
||
|
||
if attr_tag == "REALE_POSITION" and attr_text == "x":
|
||
#typ = get_type_of_name(attr_text)
|
||
#print(f"-- coord real --: {attrib.dxf.insert}")
|
||
pos = attrib.dxf.insert #Position Ecke unten links von "x"-Marker auslesen
|
||
|
||
# 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, typ)
|
||
|
||
def check_double_ids(ld, id, allSensors):
|
||
if id in allSensors and "NAME" in ld:
|
||
return True
|
||
if id in allSensors and "B" in ld and "B" in allSensors[id]:
|
||
return True
|
||
if id in allSensors and "IO" in ld and "IO" in allSensors[id]:
|
||
return True
|
||
return False
|
||
|
||
def extract_input_positions(insert_iterable) -> tuple:
|
||
allSensors = dict()
|
||
allCables = dict()
|
||
allunknowns = dict()
|
||
doubleIds = dict()
|
||
allSchaltschrank = dict()
|
||
|
||
for insert in insert_iterable:
|
||
if insert.dxftype() != 'INSERT':
|
||
continue
|
||
|
||
ld, id, typ = get_attributes_of_insert(insert)
|
||
|
||
if typ == "Sensor":
|
||
if id and "pos" in ld and isinstance(ld["pos"], tuple) and len(ld["pos"]) == 2:
|
||
if id in allSensors:
|
||
is_double = check_double_ids(ld, id, allSensors)
|
||
if not is_double:
|
||
allSensors[id] = merge_two_dicts(allSensors[id], ld) #Kombiniert alle infos aus dxf und "pos"
|
||
else:
|
||
# gib aus wo sich beide gleichlautenden Elemente auf der Zeichnung befinden
|
||
if id not in doubleIds:
|
||
doubleIds[id] = []
|
||
doubleIds[id].append(allSensors[id]["pos"]) # erste Position
|
||
doubleIds[id].append(ld["pos"]) # zweite Position
|
||
else:
|
||
allSensors[id] = ld
|
||
|
||
elif typ == "Kabel":
|
||
allCables[id] = ld
|
||
elif typ == "Schaltschrankelement":
|
||
allSchaltschrank[id] = ld
|
||
else:
|
||
allunknowns[id] = ld
|
||
|
||
return allSensors, doubleIds
|
||
|
||
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()
|
||
# sammle die Sensoren mit ihren zugehörigen Unterverteilern
|
||
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"
|
||
pattern = r"^=([A-Z]+\d+)([+\-])([A-Z]+\d+)([+\-])([A-Z]+\d+.)$"
|
||
match = re.match(pattern, unterverteiler_pfad)
|
||
if match:
|
||
anlage = match.group(1)
|
||
verteiler = match.group(3)
|
||
karte = match.group(5)
|
||
# match.group(1) # AH01
|
||
# match.group(2) # +
|
||
# match.group(3) # UH02
|
||
# match.group(4) # -
|
||
# match.group(5) # KF1FDI7
|
||
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 = get_input_positions_iter(dxf_path)
|
||
else:
|
||
res_sens, res_double = 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)
|
||
res_not_found["missing_attributes"] = 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()
|