Files
kabellaengen/lib/drawdxf.py
T

598 lines
21 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
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]
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
def map_sensor_to_cable_cfg(plines):
sens2cable = defaultdict(list)
mapping = config_BMK["Cable-Mapping"]
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[:2]
# Suche nach Key in der BMK-Config
key_with_artnr = f"{name_prefix}-{sensor_artinr}"
if key_with_artnr in mapping:
section_list = mapping[key_with_artnr]
elif name_prefix in mapping:
section_list = mapping[name_prefix]
else:
sens2cable[pl.id].append("Kein Kabeltyp zugewiesen")
# Liste aus evtl. mehreren Sektionen erzeugen
sections = [s.strip for s in section_list.split(",")]
# Evtl. Kabelkürzung durchführen, fall Kabelschwanz vorhanden
if config_BMK.has_section("Length-Adjustments") and config_BMK.has_option("Length-Adjustments", name_prefix):
length_reduction = float(config_BMK.get("Length-Adjustments", name_prefix))
cable_length = max(0.0, cable_length-length_reduction)
# Kabel-Atikelnummer innerhalb der Sektion der kabel.cfg bestimmen
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('-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)
# 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-Artikelnummern 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)
# 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"))
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)