458 lines
16 KiB
Python
458 lines
16 KiB
Python
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
|
|
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]
|
|
|
|
|
|
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 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 export_excel(json_file, out_path):
|
|
# Hier für Excel Export
|
|
print("creating excel file with cable information ..")
|
|
plines = model_from_json(json_file)
|
|
write_excel_from_json(plines, out_path)
|
|
print("done")
|
|
|
|
def write_excel_from_json(plines:Polylines, outpath:str):
|
|
wb = Workbook()
|
|
length_summary = defaultdict(int)
|
|
|
|
#Worksheet 1 - Kabellängen
|
|
ws1 = wb.active
|
|
ws1.title = "Length by ID"
|
|
ws1.append(["Cable-ID", "True Length (m)", "Rounded Length (m)"])
|
|
|
|
for pl in plines.kabel:
|
|
length = pl.length /1000 # Umrechnung von mm in m
|
|
rounded_len = math.ceil(length/10)*10
|
|
length_summary[rounded_len] +=1
|
|
ws1.append([pl.id, length, int(rounded_len)])
|
|
|
|
# Worksheet 2 - Zusammengefasste Längen
|
|
ws2 = wb.create_sheet("Length Summary")
|
|
ws2.append(["Rounded Length (m)", "Number of Cables"])
|
|
|
|
for rlength in sorted(length_summary):
|
|
ws2.append([int(rlength), length_summary[rlength]])
|
|
|
|
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())
|
|
|
|
|
|
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('-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()
|
|
|
|
plines = model_from_json(json_path)
|
|
|
|
|
|
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"))
|
|
|
|
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(json_path, excel_path)
|
|
|
|
|
|
|