Files
kabellaengen/lib/getpositions.py
T

575 lines
20 KiB
Python

import argparse
import configparser
import ezdxf.document
from ezdxf import readfile
import os
import sys
import json
import re
from shapely import Point
from itertools import combinations
from ezdxf.addons import iterdxf
import re
import time
"""
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):
'''
Hier in Zukunft weniger Abfragen: IO und B und Reale_Position wird überflüssig wenn jeder Sensor nur noch eiun 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 == "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 extract_input_positions(insert_iterable) -> tuple:
allSensors = dict()
allCables = dict()
allunknowns = 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:
allSensors[id] = merge_two_dicts(allSensors[id], ld) #Kombiniert alle infos aus dxf und "pos"
else:
allSensors[id] = ld
elif typ == "Kabel":
allCables[id] = ld
elif typ == "Schaltschrankelement":
allSchaltschrank[id] = ld
else:
allunknowns[id] = ld
return allSensors, allCables
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-Z0-9]+)$"
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)
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
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,v) in all_layers:
selectstr = f'LWPOLYLINE[layer=="{layer}"]'
for e in msp.query(selectstr):
#print_polyline(e)
rack_key = f"Rack_{rack_counter}"
ret[rack_key] = list()
for x, y, start_width, end_width, bulge in e.get_points(): # Gibt Tuple (x, y, start_width, end_width, bulge)
p = [round(x,1), round(y,1)]
ret[rack_key].append(p)
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):
if entity.dxftype() != "LWPOLYLINE":
continue
layer = entity.dxf.layer
if not any(layer == cfg_layer for cfg_layer, _ in all_layers):
continue
rack_key = f"Rack_{rack_counter}"
ret[rack_key] = []
# Verwende entity.vertices() statt get_points()
for point in entity.vertices(): # (x, y, start_width, end_width, bulge)
x, y, *_ = point # wir interessieren uns nur für x und y
ret[rack_key].append([round(x, 1), round(y, 1)])
rack_counter += 1
return ret
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, filename):
fexists = True
if not os.path.exists(filename):
mypath = os.path.join(work_dir, filename)
if not os.path.exists(mypath):
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("Validating 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("continuing")
else:
print ("No inconsistencies found")
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('-n', '--scan', action='store_true', help='print all layer of racs, distributes and equiment not empty')
args = parser.parse_args()
out_dir = os.environ.get('PROJECT_DATA')
work_dir = os.environ.get('PROJECT_WORK')
config_dir = os.environ.get("PROJECT_CFG")
filename = args.filename
(dxf_path, dexists) = check_file_in_work(work_dir, filename)
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_cables = get_input_positions_iter(dxf_path)
else:
res_sens, res_cables = get_input_positions(msp)
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
write_results(to_json(output_results), work_dir, f"{basename}.json")
else:
parser.print_help()