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 """ 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_input_positions(msp: ezdxf.document.Drawing.modelspace): """hole alle Positionen der Eingänge !!Sensor Positionen erst nach Offsett-Addition "Mitte-Mitte"!! """ allIds = dict() SpecialKeys = ["MB", # Separator "MA", # Motor "BG", # Stausensor "FC" # Motorschutzschalter ] # Über alle Blockreferenzen (INSERT) im Modelspace laufen for insert in msp.query('INSERT'): if len(insert.attribs) == 0: continue # Überspringe Blöcke ohne Attribute id = "" ld = dict() for attrib in insert.attribs: if len(insert.attribs) == 0: continue # Überspringe Blöcke ohne Attribute #print(f"Attribut Name: {attrib.dxf.tag}, Wert: {attrib.dxf.text}") ld[attrib.dxf.tag] = attrib.dxf.text if attrib.dxf.tag == "IO": id = attrib.dxf.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 attrib.dxf.tag == "B": for spec in SpecialKeys: if spec in attrib.dxf.text: id = attrib.dxf.text #print(f"-- coord {attrib.dxf.text} --: {attrib.dxf.insert}") if attrib.dxf.tag == "REALE_POSITION" and attrib.dxf.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("Sensor_Marker", "Breite") hoehe_marker = config.getfloat("Sensor_Marker", "Hoehe") pos_midx = pos.x + breite_marker*0.5 pos_midy = pos.y + hoehe_marker*0.5 ld["pos"] = (round(pos_midx, 1), round(pos_midy, 1)) # Nur wenn eine ID vorhanden ist, und eine gültige Position existiert if id and "pos" in ld and isinstance(ld["pos"], tuple) and len(ld["pos"]) == 2: if id in allIds: allIds[id] = merge_two_dicts(allIds[id], ld) #Kombiniert alle infos aus dxf und "pos" else: allIds[id] = ld return allIds 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"!! """ # # Über alle Texte laufen # for text in msp.query('MTEXT'): # print(f"Inhalt: {text.dxf.text}") # print(f"Layer: {text.dxf.layer}") # print(f"Einfügepunkt: {text.dxf.insert}") # print(f"Breite: {text.dxf.width}") # # print("Farbe:", text.dxf.color) # Achtung: 256 = "ByLayer" # # print("Höhe (height):", text.dxf.height) # # print("Rotation (degrees):", text.dxf.rotation) # # print("Breitenfaktor (width factor):", text.dxf.width) # # print("Style (Schriftart):", text.dxf.style) # # print("Handle:", text.dxf.handle) # print("---") ret = dict() # Alle Texte auf Layer "xy" all_distributors = dist2sensors.keys() all_layers = config.items('Layer_Busverteiler') 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 # 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('Layer_Pritschen')) 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 # iterate over all entities in modelspace # for e in msp: # if e.dxftype() == "LINE": # print_line(e) # if e.dxftype() == "LWPOLYLINE": # print_polyline(e) 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) 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') 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) doc = get_dxf_file(dxf_path) # type: ignore msp = doc.modelspace() res_pos = 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, "getpositions.cfg")) output_results = dict() if args.sensors: res_pos = get_input_positions(msp) output_results['sensors'] = res_pos if args.console: print(to_json(res_pos)) (res_mappings, warnings) = create_mappings(res_pos) output_results['mappings'] = res_mappings if args.console: print(to_json(res_mappings)) res_dist = get_subdistributor_positions(msp, res_mappings) output_results['distributors'] = res_dist if args.console: print(to_json(res_dist)) if args.rack: 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] write_results(to_json(output_results), work_dir, f"{basename}.json") else: parser.print_help()