import argparse import ezdxf import json import os.path from dataclasses import dataclass, asdict, fields from dacite import from_dict from typing import List from datetime import datetime from openpyxl import Workbook import math from collections import defaultdict import configparser @dataclass class Point: x: float y: float @dataclass class Polyline: id: str s_artinr: str coords: List[Point] length: float def to_tuple(self): ret = list() for p in self.coords: ret.append( (p.x, p.y) ) return ret @dataclass class Error_Connection: name: str coords: Point @dataclass class Error_Routing: unterverteiler: str sensoren: List[str] @dataclass class Polylines: kabel: List[Polyline] errors_routing: List[Error_Routing] errors_sensors: List[Error_Connection] errors_dists: List[Error_Connection] errors_existing_dists: List[str] errors_existing_sensors: List[str] @dataclass class Sensors: name: str artinr: str pos: List[float] def add_polyline(msp, points:Polyline, dxf_attribs): pts = points.to_tuple() pline = msp.add_lwpolyline(points=pts, dxfattribs=dxf_attribs) pline.rgb = (255, 128, 0) def new_dxf(plines, out_path): """ creates a new dxf file with a polyline inside which is created by the given json file """ print("creating new .dxf ..") doc = ezdxf.new('R2018', setup=True) draw_cables(plines, doc) draw_sensors(plines, doc) draw_subdists(plines, doc) doc.saveas(out_path) print("done") def modify_original_dxf(plines, originaldxf): """ adds new layer to original .dxf-file that contains cables """ print("adding cables into original .dxf ..") doc = ezdxf.readfile(originaldxf) draw_cables(plines, doc) doc.saveas(out_path) print("done") def copy_layers_into_new(originaldxf, outpath, plines): """ creates a new dxf file with a racks, sensors, subdists from original file including cable paths """ print("copying layers (Racks, Subdistributors, ...) from original .dxf into new .dxf ..") quelle = ezdxf.readfile(originaldxf) ziel = ezdxf.new('R2018', setup=True) draw_cables(plines, ziel) draw_sensors(plines, ziel) copy_layers_into_dxf_by_filter(quelle, ziel) ziel.saveas(out_path) print("done") def draw_cables(plines, doc): msp = doc.modelspace() timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M") cable_layer = f"cables_{timestamp}" # Kabel-Layer anlegen if cable_layer not in doc.layers: doc.layers.add(name=cable_layer, color=7) dxfattribs_cable={"layer": cable_layer} # Kabel zeichnen for pl in plines.kabel: # Polyline für Kabel zeichnen add_polyline(msp, pl, dxfattribs_cable) def find_close_key(pos2sensors, x, y, tolerance=10): # !!! Toleranz nicht in Config !!! ''' Funktion überprüft ob Sensoren nahezu identisch an der gleichen Stelle liegen und legt sie in diesem fall aufeinander Wird benötigt, um zusammengehörige Sensoren gestaffelt auf dxf zu zeichen ''' for (px, py) in pos2sensors: if abs(px - x) <= tolerance and abs(py - y) <= tolerance: return (px, py) return None def draw_sensors(plines, doc): msp = doc.modelspace() timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M") sensor_layer = f"sensors_{timestamp}" # Sensor-Layer erzeugen if sensor_layer not in doc.layers: doc.layers.add(name=sensor_layer, color=5) dxfattribs_sensors={"layer": sensor_layer, "height": 100} # Sensoren nach Endpunkten gruppieren -> mehrfacheinträge gestaffelt zeichnen pos2sensors = defaultdict(list) for pl in plines.kabel: pt2 = pl.coords[-1] #Endpunkt des Kabels = Sensor Position pos_key = find_close_key(pos2sensors, pt2.x, pt2.y) if pos_key: pos2sensors[pos_key].append(pl) else: pos2sensors[(pt2.x, pt2.y)].append(pl) # Sensor Blöcke zeichnen for (x,y), pls in pos2sensors.items(): for i, pl in enumerate(pls): sensor_name = pl.id.split('-')[-1] pt1 = pl.coords[-2] pt2 = pl.coords[-1] dx = pt2.x - pt1.x dy = pt2.y - pt1.y text = msp.add_text(sensor_name, dxfattribs=dxfattribs_sensors) # Offsets für Beschriftungen offsetx = 0 offsety = 0 # Kabel Horizontal if abs(dx) > abs(dy): n = len(pls) center_offset = i - (n - 1) / 2 # Falls mehrere Sensoren -> Verschiebung so, dass Kabel immer in Mitte ankommt offsety = -80 + center_offset * 110 # Wert -80 über Try-and-Error zur Mitte der Beschriftung angepasst if dx > 0: halign = 0 # LEFT offsetx = 50 # Wert 50 durch Try-and-Error sodass etwas Abstand zu Kabelpritsche else: halign = 2 # RIGHT offsetx = -50 # Wert 50 durch Try-and-Error sodass etwas Abstand zu Kabelpritsche valign = 1 # BOTTOM # Kabel vertikal else: if dy > 0: valign = 0 # BASELINE offsety = 50 + i * 110 # nach oben anfügen else: valign = 3 # TOP offsety = -50 - i * 110 # nach unten anfügen halign = 1 # CENTER # Alignments setzen text.dxf.halign = halign text.dxf.valign = valign text.set_placement((pt2.x + offsetx, pt2.y + offsety)) def draw_subdists(plines, doc): msp = doc.modelspace() timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M") subdist_layer = f"subdists_{timestamp}" # Sensor-Layer erzeugen if subdist_layer not in doc.layers: doc.layers.add(name=subdist_layer, color=3) dxfattribs_subdists={"layer": subdist_layer, "height": 100} subdist_positions = set() for pl in plines.kabel: pt1 = pl.coords[0] # Startposition = UV-Position pos = (pt1.x, pt1.y) if pos in subdist_positions: continue subdist_positions.add(pos) subdist_name = pl.id.split('-')[0] pt2 = pl.coords[1] dx = pt2.x - pt1.x dy = pt2.y - pt1.y # Offsets für Beschriftungen offsetx = 0 offsety = 0 if abs(dx) > abs(dy): # Horizontal offsety = -80 # Wert -80 über Try-and-Error zur Mitte der Beschriftung angepasst if dx < 0: halign = 0 # LEFT offsetx = 50 # Wert 50 durch Try-and-Error sodass etwas Abstand zu Kabelpritsche else: halign = 2 # RIGHT offsetx = -50 # Wert 50 durch Try-and-Error sodass etwas Abstand zu Kabelpritsche valign = 1 # BOTTOM else: # Vertikal if dy < 0: valign = 0 # BASELINE offsety = 50 else: valign = 3 # TOP offsety = -50 halign = 1 # CENTER # Text platzieren text = msp.add_text(subdist_name, dxfattribs=dxfattribs_subdists) text.dxf.halign = halign text.dxf.valign = valign text.set_placement((pt1.x + offsetx, pt1.y + offsety)) def model_from_json(json_file): with open(json_file, encoding='utf-8') as fh: data = json.load(fh) plines = from_dict( data_class=Polylines, data=data ) return plines def parse_sensors_from_json(positions_json): with open(positions_json, encoding='utf-8') as fh: data = json.load(fh) sensors = {} for name, data in data.get("sensors", {}).items(): sensor = Sensors( name=name, artinr=data.get("ARTINR", ""), pos=data.get("pos", [0.0, 0.0]), ) sensors[name] = sensor return sensors def export_excel(plines, out_path): # Hier für Excel Export print("creating excel file with cable information ..") write_excel_from_json(plines, out_path) print("done") def write_excel_from_json(plines:Polylines, outpath:str): wb = Workbook() sens2cable = map_sensor_to_cable(plines) # Dicts für Anzahl bzw kummulierte Länge count_summary = defaultdict(int) length_summary = defaultdict(float) #Worksheet 1 - Kabellängen nach Kabel-ID ws1 = wb.active ws1.title = "Length by ID" ws1.append(["Cable-ID", "True Length (m)", "Cable-ArtNr", "Cable-Name (short)"]) for pl in plines.kabel: length = pl.length /1000 # Umrechnung von mm in m for artnr in sens2cable[pl.id]: cable_name = "" if artnr != "Kabel länger als max. Kabellänge": cable_name = cable_cfg["Sivasnummern"][artnr] ws1.append([pl.id, length, artnr, cable_name]) if "MA" in pl.id: length_summary[artnr] += math.ceil(length) # Aufrunden von z.b. 10,3 auf 11 m (volle Meter!) else: count_summary[artnr] += 1 # Worksheet 2 - Kabelnummern und Stückzahlen ws2 = wb.create_sheet("Cables SIVAS") ws2.append(["Cable-ArtNr", "Amount (pcs)", "Cumm. Length (m)"]) all_artnrs = set(count_summary.keys()) | set(length_summary.keys()) for artnr in sorted(all_artnrs): count = count_summary.get(artnr, "") cumm_length = length_summary.get(artnr, "") ws2.append([artnr, count, cumm_length]) # Abfage ob Fehler Worsheets ausgegeben werden if len(plines.errors_existing_sensors) > 0 or len(plines.errors_dists) > 0: # Worksheet 3 - Nicht an Racks gekoppeltes Equipment ws3 = wb.create_sheet("Not connected Equipment") ws3.append(["Type", "ID", "x", "y"]) for error in plines.errors_sensors: ws3.append(["Sensor / Actuator", error.name, error.coords.x, error.coords.y]) for error in plines.errors_dists: ws3.append(["Subistributor", error.name, error.coords.x, error.coords.y]) if len(plines.errors_routing) > 0: # Worksheet 4 - Fehlgeschlagenes Routing ws4 = wb.create_sheet("Routing Errors") ws4.append(["Subdistributor", "Sensor / Actuator", "Details"]) nicht_angebunden = set(e.name for e in plines.errors_sensors + plines.errors_dists) for routing_error in plines.errors_routing: uv = routing_error.unterverteiler uv_nicht_angebunden = uv in nicht_angebunden if uv in plines.errors_existing_dists: ws4.append([uv,"-", "Distributor not found in given layout"]) continue for sensor in routing_error.sensoren: sensor_nicht_angebunden = sensor in nicht_angebunden if sensor_nicht_angebunden and uv_nicht_angebunden: grund = "Subdistributor and sensor / actuator not connected to racks" elif sensor_nicht_angebunden: grund = "Sensor / actuator not connected to racks" elif uv_nicht_angebunden: grund = "Subdistributor not connected to racks" else: grund = "Failed routing (not caused by missing connection)" ws4.append([uv, sensor, grund]) wb.save(outpath) 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 copy_layers_into_dxf_by_filter(dxf_source: ezdxf.document.Drawing, dxf_target:ezdxf.document.Drawing): msp_source = dxf_source.modelspace() msp_target = dxf_target.modelspace() subdist_layers = set(config.options('GetPos-Layer_Distributors')) rack_layers = set(config.options('GetPos-Layer_Racks')) equipment_layers = set(config.options('GetPos-Layer_Equipment')) tunnel_layers = set(config.options('GetPos-Layer_Tunnel')) layernames = set() layernames.update(subdist_layers) layernames.update(rack_layers) layernames.update(equipment_layers) layernames.update(tunnel_layers) # # welche Texte existieren # for layername in layernames: # selectstr = f'MTEXT[layer=="{layername}"]' # for text in msp_source.query(selectstr): # inhalt = text.dxf.text # position = text.dxf.insert # print(f"Text: '{inhalt}' an Position: {position} auf Layer: {layername}") # text_entity = text.copy() # msp_target.add_entity(text_entity) layer_names_inside = dxf_source.layers alle_block_defs = set(dxf_source.blocks.block_names()) verwendete = {insert.dxf.name for insert in msp_source.query("INSERT")} # 1. Textstyles kopieren for style in dxf_source.styles: if style.dxf.name not in dxf_target.styles: dxf_target.styles.new(name=style.dxf.name) # 4. Filter-Layernamen bestimmen for layername in layernames: if layername not in dxf_source.layers: continue # Falls der Layer noch nicht im Zieldokument existiert, neu anlegen if layername not in dxf_target.layers: quelle_layer = dxf_source.layers.get(layername) dxf_target.layers.add( name=layername, color=quelle_layer.color, linetype=quelle_layer.dxf.linetype, lineweight=quelle_layer.dxf.lineweight ) # Alle Entities auf diesem Layer kopieren entities = msp_source.query(f"*[layer=='{layername}']") for entity in entities: msp_target.add_entity(entity.copy()) def get_cable_artnr(section, length): """ Sucht in der angegebenen Config-Section die passende Kabel-Artikelnr. für die gegebene Länge. """ if section not in cable_cfg: return None entries = cable_cfg[section] # Alle Einträge der Form: "5" = "123456789" length_keys = sorted([float(k) for k in entries.keys()]) for l in length_keys: if length <= l: return entries[str(l)] # Annahme: Keys sind ganze Zahlen in Metern return None def map_sensor_to_cable(plines): sens2cable = defaultdict(list) for pl in plines.kabel: sensor_name = pl.id.split('-')[-1] cable_length = pl.length/1000 sensor_artinr = pl.s_artinr name_prefix = sensor_name[0:2] sections = [] #Default if name_prefix == "MA": sections = ["MA"] # Sektion für Motor-Kabel elif name_prefix == "MB": sections = ["WD_Q"] # Sektion für Ventil-Kabel elif name_prefix == "QM": sections = ["WD_Q"] # Sektion für Ventil-Kabel elif sensor_name.startswith("B"): # Sensor Name beginnt mit B -> Sensor jeglicher Art if name_prefix == "BX": # Wenn "BX", dann Scanner sections = ["WF_B", "WD_I"] # Sektion für Scanner Patch-Kabel und Sektion für Standard-Sensorkabel zur Spannungsversorgung cable_length = max(0.0, cable_length -4.0) # 4 Meter abziehen, da bereits 5m Kabelschwanz an Scanner dran elif sensor_artinr == 829422026: sections = ["WD_I-829422026"] elif sensor_artinr == 720002003: sections = ["WD_I-720002003"] else: sections = ["WD_I"] if not sections: sens2cable[pl.id].append ("Kein Kabeltyp zugewiesen") continue for section in sections: cable_artnr = get_cable_artnr(section, cable_length) if cable_artnr is None: sens2cable[pl.id].append(f"Kabel länger als max. Kabellänge") else: sens2cable[pl.id].append(cable_artnr) return sens2cable if __name__ == '__main__': parser = argparse.ArgumentParser(description='draws a dxf file with the given cable coordinates', prog='drawdxf') parser.add_argument('-f', '--filename', action='store', required=True, help='this json file contains all cables and its coordinates which should be drawn. Saved with an unique timestamp', metavar='myfile.json') parser.add_argument('-p','--positions', action='store', required=True, help='this json file contains positional information written by getpositions', metavar='myfile.json') parser.add_argument('-d', '--dxf', action='store', help='this dxf drawing will be copied and the new layer with the cables will be added. Original file must be added with --origin', metavar='myfile.dxf') parser.add_argument('-c', '--copy_layer', action='store', help='copy layers of racks, sensors, distributors into a new .dxf-file. File also contains cable paths. Original file must be added with --origin', metavar='original.dxf') parser.add_argument('-n', '--new', action='store', help='create a new dxf file only with cables in it. Name is basename and a timestamp') parser.add_argument('-x', '--excel', action='store', help='create a xlsx file with cables data', metavar='allCables.xls') parser.add_argument('-o', '--origin', action='store', help='name of original .dxf file used by -d and -a', metavar='original.dxf') args = parser.parse_args() config_dir = os.environ.get("PROJECT_CFG") work_dir = os.fspath(os.environ.get('PROJECT_WORK')) json_file = args.filename (json_path, jexists) = check_file_in_work(work_dir, json_file) if not jexists: print(f"file {json_file} does not exist") parser.print_help() exit() positions_json = args.positions (positions_path, jexists) = check_file_in_work(work_dir, positions_json) if not jexists: print(f"file {positions_json} does not exist") parser.print_help() exit() plines = model_from_json(json_path) sensors = parse_sensors_from_json(positions_path) # 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")) # Config für Kabel-Atikelnummern laden cable_cfg = configparser.ConfigParser() cable_cfg.optionxform = str #Keys case-sensitive with open(os.path.join(config_dir, "kabel.cfg"), encoding="utf-8") as f: cable_cfg.read_file(f) dxf_file = args.dxf if args.dxf or args.copy_layer: if not args.origin: parser.print_help() exit() else: (origin_path, dexists) = check_file_in_work(work_dir, args.origin) if args.dxf: (dxf_path, dexists) = check_file_in_work(work_dir, dxf_file) if not dexists: print(f"file {dxf_file} does not exist") parser.print_help() exit() out_path = dxf_path res_pos = new_dxf(plines, dxf_path) if args.copy_layer: out_path = os.path.join(work_dir, args.copy_layer) res_pos = new_dxf(plines, out_path) copy_layers_into_new(origin_path, out_path, plines) if args.new: # erzeuge dxf Datei nur mit Kabeln out_path = os.path.join(work_dir, args.new) res_pos = new_dxf(plines, out_path) if args.excel: excel_path = os.path.join(work_dir, args.excel) export_excel(plines, excel_path)