529 lines
18 KiB
Python
529 lines
18 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_attributes_of_insert(insert):
|
|
|
|
SpecialKeys = ["MB", # Separator
|
|
"MA", # Motor
|
|
"BG", # Stausensor
|
|
"BP" # Schalter Druckluft
|
|
]
|
|
|
|
KabelKey = "WD_"
|
|
DropKeys = [
|
|
"FC", # Motorschutzschalter
|
|
"PF", # Leuchtmelder
|
|
"SF",
|
|
"DI", # Feedback vom Gerät
|
|
"QA", # Hauptschütz
|
|
"SF" # Drucktaster
|
|
]
|
|
id = ""
|
|
ld = dict()
|
|
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
|
|
typ = 'unknown'
|
|
if attr_tag == "IO":
|
|
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))
|
|
typ = "Sensor"
|
|
|
|
if attr_tag == "B":
|
|
# Suche nach Sensoren
|
|
for spec in SpecialKeys:
|
|
if spec in attr_text:
|
|
id = attr_text
|
|
typ = "Sensor"
|
|
|
|
# Suche nach Kabel
|
|
if KabelKey in attr_text:
|
|
id = attr_text
|
|
typ = "Kabel"
|
|
|
|
# suche nach Items die wir nicht weiter verfolgen
|
|
for dropkey in DropKeys:
|
|
if dropkey in attr_text:
|
|
typ = "Schaltschrankelement"
|
|
id = attr_text
|
|
return {}, id, typ
|
|
|
|
if attr_tag == "REALE_POSITION" and attr_text == "x":
|
|
#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))
|
|
typ = "Sensor"
|
|
|
|
return ld, id, typ
|
|
|
|
def get_input_positions(msp: ezdxf.document.Drawing.modelspace)-> tuple:
|
|
"""hole alle Positionen der Eingänge !!Sensor Positionen erst nach Offsett-Addition "Mitte-Mitte"!!
|
|
"""
|
|
allSensors = dict()
|
|
allCables = dict()
|
|
|
|
# Über alle Blockreferenzen (INSERT) im Modelspace laufen
|
|
for insert in msp.query('INSERT'):
|
|
# Überspringe Blöcke ohne Attribute
|
|
if len(insert.attribs) == 0:
|
|
continue
|
|
|
|
ld, id, typ = get_attributes_of_insert(insert)
|
|
|
|
# Nur wenn eine ID vorhanden ist, und eine gültige Position existiert
|
|
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
|
|
else:
|
|
# Falls das Objekt ein Teil des Schaltschranks o.ä ist, dann nichts merken
|
|
pass
|
|
return allSensors, allCables
|
|
|
|
def get_input_positions_iter(dxf_path) -> tuple:
|
|
"""
|
|
Iterative Version für große DXF-Dateien mit iterdxf (kein Context Manager!).
|
|
"""
|
|
allSensors = dict()
|
|
allCables = dict()
|
|
|
|
for insert in iterdxf.modelspace(dxf_path):
|
|
if insert.dxftype() != 'INSERT':
|
|
continue
|
|
|
|
ld, id, typ = get_attributes_of_insert(insert)
|
|
|
|
# Nur wenn eine ID vorhanden ist, und eine gültige Position existiert
|
|
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
|
|
else:
|
|
# Falls das Objekt ein Teil des Schaltschranks o.ä ist, dann nichts merken
|
|
pass
|
|
|
|
return allSensors, allCables
|
|
|
|
def create_mappings(positions:dict) -> dict:
|
|
unterverteiler_pfad = ""
|
|
dnamen = dict()
|
|
# sammle die Sensoren mit ihren zugehörigen Unterverteilern
|
|
sensor2unterverteiler = dict()
|
|
warnings = list()
|
|
for sensorname,v in positions.items():
|
|
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.append(unterverteiler_pfad)
|
|
continue
|
|
|
|
if not verteiler 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
|
|
# for mtext in msp.query("MTEXT"):
|
|
# print(f"Inhalt: {mtext.text}")
|
|
# print(f"Layer: {mtext.dxf.layer}")
|
|
# print(f"Einfügepunkt: {mtext.dxf.insert}")
|
|
# print("---")
|
|
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
|
|
|
|
|
|
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)
|
|
|
|
zeitanfang = time.time()
|
|
|
|
if dxf_is_binary(dxf_path): # Wenn dxf eine binary ist, dann komplett parsen und modelspace anlegen
|
|
doc = get_dxf_file(dxf_path)
|
|
msp = doc.modelspace()
|
|
use_iter = False
|
|
else:
|
|
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:
|
|
|
|
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"))
|
|
|
|
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)
|
|
output_results["not_found"] = res_not_found
|
|
write_results(to_json(output_results), work_dir, f"{basename}.json")
|
|
zeitende = time.time()
|
|
print(zeitende-zeitanfang)
|
|
|
|
else:
|
|
parser.print_help()
|