1028 lines
37 KiB
Python
1028 lines
37 KiB
Python
import argparse
|
|
import ezdxf
|
|
import json
|
|
import os.path
|
|
from dataclasses import dataclass, asdict, field
|
|
from dacite import from_dict
|
|
from typing import List, Dict
|
|
from datetime import datetime
|
|
from openpyxl import Workbook
|
|
import math
|
|
from collections import defaultdict
|
|
import configparser
|
|
import updateconfignames as uc
|
|
from pathlib import Path
|
|
|
|
from error_collector import ErrorCollector, write_json_file
|
|
|
|
|
|
@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
|
|
# Fehlgeschlagene Anbindung von einem Sensor / Dist zu einem Rack
|
|
class Error_Connection:
|
|
name: str
|
|
coords: Point
|
|
|
|
@dataclass
|
|
# Felgeschlagene Verbindung von einem Dist zu Sensor(en) aus beliebigem Grund
|
|
class Error_Routing:
|
|
unterverteiler: str
|
|
sensoren: List[str]
|
|
|
|
@dataclass
|
|
class Coordinate:
|
|
x: float
|
|
y: float
|
|
z: float
|
|
|
|
@dataclass
|
|
class RackGeometry:
|
|
length: float
|
|
coordinates: List[Coordinate]
|
|
|
|
|
|
@dataclass
|
|
class Polylines:
|
|
"""Enthält alle Kabel-Polylinien und Rack-Geometrien (ohne Fehlerbehandlung)."""
|
|
kabel: List[Polyline]
|
|
rack_geometry: Dict[str, RackGeometry] = field(default_factory=dict)
|
|
|
|
|
|
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):
|
|
"""
|
|
Erstellt eine neue DXF-Datei mit den gegebenen Polylinien und speichert sie unter dem angegebenen Pfad.
|
|
|
|
Args:
|
|
plines (Polylines): Die Polylinien- und Geometriedaten, die gezeichnet werden sollen.
|
|
out_path (str): Der Pfad, unter dem die neue DXF-Datei gespeichert wird.
|
|
"""
|
|
doc = ezdxf.new('R2018', setup=True)
|
|
doc.header['$INSUNITS'] = 4 # Millimeter
|
|
draw_cables(plines, doc)
|
|
draw_sensors(plines, doc)
|
|
draw_subdists(plines, doc)
|
|
draw_racks(plines, doc)
|
|
|
|
doc.saveas(out_path)
|
|
print("Cable-Routes exported to new dxf-file")
|
|
|
|
def modify_original_dxf(plines, originaldxf):
|
|
"""
|
|
Fügt einer bestehenden DXF-Datei einen neuen Layer mit Kabeln hinzu und speichert das Ergebnis.
|
|
|
|
Args:
|
|
plines (Polylines): Die Polylinien- und Geometriedaten, die gezeichnet werden sollen.
|
|
originaldxf (str): Pfad zur bestehenden DXF-Datei, die modifiziert werden soll.
|
|
"""
|
|
print("adding cables into original .dxf ..")
|
|
|
|
doc = ezdxf.readfile(originaldxf)
|
|
draw_cables(plines, doc)
|
|
|
|
doc.saveas(out_path)
|
|
print("done")
|
|
|
|
|
|
def draw_cables(plines, doc):
|
|
"""
|
|
Zeichnet alle Kabel-Polylinien in das DXF-Dokument auf einen eigenen Layer.
|
|
|
|
Args:
|
|
plines (Polylines): Die Polylinien- und Geometriedaten.
|
|
doc (ezdxf.document.Drawing): Das zu bearbeitende DXF-Dokument.
|
|
"""
|
|
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 draw_racks(plines, doc):
|
|
"""
|
|
Zeichnet die Racks als 3D-Polylinien und platziert Rack-Namen als Text im DXF-Dokument.
|
|
|
|
Args:
|
|
plines (Polylines): Die Polylinien- und Geometriedaten.
|
|
doc (ezdxf.document.Drawing): Das zu bearbeitende DXF-Dokument.
|
|
"""
|
|
msp = doc.modelspace()
|
|
timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M")
|
|
rack_layer = f"racks_{timestamp}"
|
|
|
|
# Rack-Layer anlegen (Farbe 3 = grün z.B.)
|
|
if rack_layer not in doc.layers:
|
|
doc.layers.add(name=rack_layer, color=3, lineweight=200) # lineweight für Dicke
|
|
|
|
dxfattribs_rack = {
|
|
"layer": rack_layer,
|
|
"color": 3,
|
|
"lineweight": 200 # Lineweight (in 1/100 mm)
|
|
}
|
|
|
|
for rack_name, rack_geom in plines.rack_geometry.items():
|
|
classifier = rack_name[0]
|
|
|
|
if classifier in ("t", "v", "d"):
|
|
continue
|
|
|
|
coords = [(pt.x, pt.y, pt.z) for pt in rack_geom.coordinates]
|
|
if not coords:
|
|
continue
|
|
|
|
polyline = msp.add_polyline3d(coords, dxfattribs=dxfattribs_rack)
|
|
|
|
if classifier == "c":
|
|
continue
|
|
|
|
# Textplatzierung
|
|
x, y, z = coords[0] # Get the first coordinate for text placement
|
|
|
|
# Orientierung bestimmen (horizontal oder vertikal)
|
|
if len(coords) >= 2:
|
|
x2, y2, _ = coords[1]
|
|
dx = x2 - x
|
|
dy = y2 - y
|
|
|
|
is_vertical = abs(dy) > abs(dx)
|
|
else:
|
|
is_vertical = False # Standard: horizontal
|
|
|
|
# Text platzieren
|
|
text_entity = msp.add_text(
|
|
rack_name,
|
|
dxfattribs={
|
|
"layer": rack_layer,
|
|
"height": 75,
|
|
"color": 3,
|
|
"rotation": 90 if is_vertical else 0,
|
|
}
|
|
)
|
|
|
|
# Offset definieren
|
|
offset_x = -50 if is_vertical else 50
|
|
offset_y = 100 if is_vertical else 50
|
|
|
|
text_entity.set_placement((x + offset_x, y + offset_y))
|
|
|
|
|
|
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):
|
|
"""
|
|
Zeichnet die Sensoren als Textobjekte an den Endpunkten der Kabel und gruppiert sie ggf. gestaffelt.
|
|
|
|
Args:
|
|
plines (Polylines): Die Polylinien- und Geometriedaten.
|
|
doc (ezdxf.document.Drawing): Das zu bearbeitende DXF-Dokument.
|
|
"""
|
|
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, pt2 = pl.coords[-2], pl.coords[-1]
|
|
|
|
# Platzierungsinfo über Hilfsfunktion
|
|
placement_info = _calculate_text_placement(pt1, pt2, 'sensor', i, len(pls))
|
|
|
|
text = msp.add_text(sensor_name, dxfattribs=dxfattribs_sensors)
|
|
text.dxf.halign = placement_info["halign"]
|
|
text.dxf.valign = placement_info["valign"]
|
|
text.set_placement(placement_info["placement"])
|
|
|
|
def draw_subdists(plines, doc):
|
|
"""
|
|
Zeichnet die Unterverteiler (Subdistributoren) als Textobjekte an den Startpunkten der Kabel.
|
|
|
|
Args:
|
|
plines (Polylines): Die Polylinien- und Geometriedaten.
|
|
doc (ezdxf.document.Drawing): Das zu bearbeitende DXF-Dokument.
|
|
"""
|
|
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[1] # 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[0]
|
|
placement_info = _calculate_text_placement(pt1, pt2, 'subdist')
|
|
|
|
text = msp.add_text(subdist_name, dxfattribs=dxfattribs_subdists)
|
|
text.dxf.halign = placement_info["halign"]
|
|
text.dxf.valign = placement_info["valign"]
|
|
text.set_placement(placement_info["placement"])
|
|
|
|
def _calculate_text_placement(pt1, pt2, text_type='sensor', item_index=0, total_items=1):
|
|
"""
|
|
Berechnet die optimale Position und Ausrichtung für Beschriftungen.
|
|
Gibt ein Dictionary mit Platzierungskoordinaten und Ausrichtungs-Flags zurück.
|
|
"""
|
|
dx = pt2.x - pt1.x
|
|
dy = pt2.y - pt1.y
|
|
|
|
# Lese Offsets aus der Konfiguration, mit Fallback-Werten
|
|
offset_h_dist = 50.0
|
|
offset_v_dist = 50.0
|
|
offset_y_center = -80.0
|
|
offset_y_stack = 110.0
|
|
|
|
offsetx, offsety = 0, 0
|
|
|
|
if abs(dx) > abs(dy): # Kabel verläuft hauptsächlich horizontal
|
|
valign = 1 # BOTTOM
|
|
if text_type == 'sensor':
|
|
center_offset = item_index - (total_items - 1) / 2
|
|
offsety = offset_y_center + center_offset * offset_y_stack
|
|
else: # subdist
|
|
offsety = offset_y_center
|
|
|
|
if dx > 0:
|
|
halign = 0 # LEFT
|
|
offsetx = offset_h_dist
|
|
else:
|
|
halign = 2 # RIGHT
|
|
offsetx = -offset_h_dist
|
|
|
|
else: # Kabel verläuft hauptsächlich vertikal
|
|
halign = 1 # CENTER
|
|
if dy > 0:
|
|
valign = 0 # BASELINE
|
|
offsety = offset_v_dist + item_index * offset_y_stack
|
|
else:
|
|
valign = 3 # TOP
|
|
offsety = -offset_v_dist - item_index * offset_y_stack
|
|
|
|
return {
|
|
"placement": (pt2.x + offsetx, pt2.y + offsety),
|
|
"halign": halign,
|
|
"valign": valign,
|
|
}
|
|
|
|
def model_from_json(json_file):
|
|
"""
|
|
Lädt die Polylinien- und Geometriedaten aus einer JSON-Datei und gibt Polylines-Objekt und ErrorCollector zurück.
|
|
|
|
Args:
|
|
json_file (str): Pfad zur JSON-Datei.
|
|
|
|
Returns:
|
|
tuple: (Polylines-Objekt, ErrorCollector-Objekt)
|
|
"""
|
|
with open(json_file, encoding='utf-8') as fh:
|
|
data = json.load(fh)
|
|
|
|
# ErrorCollector für Fehler und Warnungen initialisieren
|
|
error_collector = ErrorCollector()
|
|
|
|
# Fehler-Daten aus JSON extrahieren und in ErrorCollector speichern
|
|
error_fields = {
|
|
'errors_routing': [],
|
|
'errors_sensors': [],
|
|
'errors_dists': [],
|
|
'errors_dists_not_in_layout': [],
|
|
'errors_sensors_not_in_layout': [],
|
|
'errors_missing_attributes': {},
|
|
'errors_tunnels': []
|
|
}
|
|
|
|
# Fehler aus data extrahieren
|
|
for field in error_fields.keys():
|
|
if field in data:
|
|
error_fields[field] = data.pop(field)
|
|
|
|
# Fehler zum ErrorCollector hinzufügen
|
|
errors_dict = {}
|
|
warnings_dict = {}
|
|
|
|
# Kritische Fehler
|
|
if error_fields['errors_routing']:
|
|
errors_dict['routing'] = error_fields['errors_routing']
|
|
if error_fields['errors_sensors']:
|
|
errors_dict['sensors_connection'] = error_fields['errors_sensors']
|
|
if error_fields['errors_dists']:
|
|
errors_dict['dists_connection'] = error_fields['errors_dists']
|
|
if error_fields['errors_dists_not_in_layout']:
|
|
errors_dict['dists_not_in_layout'] = error_fields['errors_dists_not_in_layout']
|
|
if error_fields['errors_sensors_not_in_layout']:
|
|
errors_dict['sensors_not_in_layout'] = error_fields['errors_sensors_not_in_layout']
|
|
if error_fields['errors_tunnels']:
|
|
errors_dict['tunnels'] = error_fields['errors_tunnels']
|
|
|
|
# Warnungen
|
|
if error_fields['errors_missing_attributes']:
|
|
warnings_dict['missing_attributes'] = error_fields['errors_missing_attributes']
|
|
|
|
# Warnungen aus getpositions und routing übernehmen
|
|
if 'warnings' in data:
|
|
warnings_from_upstream = data.pop('warnings')
|
|
if isinstance(warnings_from_upstream, dict):
|
|
warnings_dict.update(warnings_from_upstream)
|
|
|
|
if errors_dict:
|
|
error_collector.add_errors(errors_dict)
|
|
if warnings_dict:
|
|
error_collector.add_warnings(warnings_dict)
|
|
|
|
# Polylines-Objekt aus den verbleibenden Daten erstellen
|
|
plines = from_dict(
|
|
data_class=Polylines,
|
|
data=data
|
|
)
|
|
return plines, error_collector
|
|
|
|
def parse_sensors_from_json(positions_json):
|
|
"""
|
|
Liest Sensordaten aus einer JSON-Datei und gibt ein Dictionary mit Sensorobjekten zurück.
|
|
|
|
Args:
|
|
positions_json (str): Pfad zur JSON-Datei mit Sensorpositionen.
|
|
|
|
Returns:
|
|
dict: Dictionary mit Sensorname als Key und Sensorobjekt als Value.
|
|
"""
|
|
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 get_all_artnrs(plines: Polylines, sens2cable: dict) -> set:
|
|
"""
|
|
Sammelt alle verwendeten Artikelnummern aus den Polylinien und der Sensor-Kabel-Zuordnung.
|
|
|
|
Args:
|
|
plines (Polylines): Die Polylinien- und Geometriedaten.
|
|
sens2cable (dict): Mapping von Sensor-IDs zu Kabel-Artikelnummern.
|
|
|
|
Returns:
|
|
set: Menge aller gefundenen Artikelnummern.
|
|
"""
|
|
all_artnrs = set()
|
|
for pl in plines.kabel:
|
|
if pl.s_artinr:
|
|
all_artnrs.add(pl.s_artinr)
|
|
|
|
for cablelist in sens2cable.values():
|
|
for artnr in cablelist:
|
|
if isinstance(artnr, str) and artnr.isdigit():
|
|
all_artnrs.add(artnr)
|
|
return all_artnrs
|
|
|
|
def mark_missings(all_artnrs):
|
|
"""
|
|
Markiert fehlende Artikelnummern im Bezeichner-Config, falls sie nicht vorhanden sind.
|
|
|
|
Args:
|
|
all_artnrs (set): Menge aller verwendeten Artikelnummern.
|
|
"""
|
|
if "Sivasnummern" not in bezeichner_cfg:
|
|
bezeichner_cfg["Sivasnummern"] = {}
|
|
if "Missing" not in bezeichner_cfg:
|
|
bezeichner_cfg["Missing"] = {}
|
|
|
|
for artnr in all_artnrs:
|
|
if artnr not in bezeichner_cfg["Sivasnummern"]:
|
|
bezeichner_cfg["Missing"][artnr] = ""
|
|
|
|
def write_excel_from_json(plines: Polylines, sens2cable: dict, outpath: str, error_collector: ErrorCollector = None, with_bom=True):
|
|
"""
|
|
Erstellt Excel-Reports (Kabelübersicht, Fehlerlisten, Stückliste) aus den Polylinien- und Zuordnungsdaten.
|
|
|
|
Args:
|
|
plines (Polylines): Die Polylinien- und Geometriedaten.
|
|
sens2cable (dict): Mapping von Sensor-IDs zu Kabel-Artikelnummern.
|
|
outpath (str): Pfad zur Ausgabedatei.
|
|
error_collector (ErrorCollector, optional): ErrorCollector mit Fehlern und Warnungen.
|
|
with_bom (bool, optional): Ob zusätzlich eine Stückliste (BOM) erzeugt werden soll. Default: True.
|
|
"""
|
|
# 1. Daten aggregieren und für die Reports vorbereiten
|
|
processed_data = _process_cable_data(plines, sens2cable)
|
|
|
|
# 2. Haupt-Excel-Datei (Kabellängen und Fehler) erstellen
|
|
wb_main = Workbook()
|
|
|
|
_create_cable_list_sheet(wb_main.active, plines, sens2cable, bezeichner_cfg)
|
|
_create_cable_summary_sheet(wb_main.create_sheet(), processed_data)
|
|
_create_rack_lengths_sheet(wb_main.create_sheet(), plines)
|
|
_create_error_sheets(wb_main, error_collector)
|
|
|
|
wb_main.save(outpath)
|
|
print("Cable-Summary exported to Excel-file")
|
|
|
|
# 3. Optionale Stücklisten-Datei (BOM) erstellen
|
|
if with_bom:
|
|
_create_bom_workbook(outpath, processed_data, bezeichner_cfg)
|
|
|
|
def _get_sivas_name(artnr, bezeichner_cfg):
|
|
"""Holt den SIVAS-Namen für eine gegebene Artikelnummer aus der Konfiguration."""
|
|
if not artnr:
|
|
return "Keine Artikelnummer vergeben. Layout prüfen."
|
|
|
|
name = bezeichner_cfg["Sivasnummern"].get(artnr, "")
|
|
if not name:
|
|
return f"Kein Eintrag zu Art.-Nr: {artnr} in bezeichner.cfg."
|
|
return name
|
|
|
|
def _process_cable_data(plines: Polylines, sens2cable: dict):
|
|
"""
|
|
Aggregiert und bereitet die Kabeldaten für die Excel-Ausgabe auf.
|
|
|
|
Args:
|
|
plines (Polylines): Die Polylinien- und Geometriedaten.
|
|
sens2cable (dict): Mapping von Sensor-IDs zu Kabel-Artikelnummern.
|
|
|
|
Returns:
|
|
dict: Verschiedene aggregierte Auswertungen für die Excel-Ausgabe.
|
|
"""
|
|
# Globale Zusammenfassungen
|
|
artnr_to_stückware_count = defaultdict(int)
|
|
artnr_to_meterware_length = defaultdict(float)
|
|
artnr_to_sensor_count = defaultdict(int)
|
|
|
|
# UV-spezifische (Unterverteiler) Zusammenfassungen
|
|
uv_to_artnr_to_stückware_count = defaultdict(lambda: defaultdict(int))
|
|
uv_to_artnr_to_meterware_length = defaultdict(lambda: defaultdict(float))
|
|
uv_to_artnr_to_sensor_count = defaultdict(lambda: defaultdict(int))
|
|
|
|
seen_sensors = set()
|
|
uv_sensor_mapping = defaultdict(dict)
|
|
|
|
for pl in plines.kabel:
|
|
length_m = round(pl.length / 1000, 1)
|
|
uv_name, sensor_name = pl.id.split("-")[0], pl.id.split("-")[-1]
|
|
|
|
# Die Unterscheidung zwischen Stück- und Meterware wird hier explizit gemacht
|
|
is_meterware = "MA" in pl.id
|
|
|
|
# Globaler Sensor-Count (einmal pro Sensor)
|
|
if sensor_name not in seen_sensors:
|
|
seen_sensors.add(sensor_name)
|
|
artnr_to_sensor_count[pl.s_artinr] += 1
|
|
|
|
# UV-spezifisches Sensor-Mapping für präzise Zählung pro UV
|
|
if sensor_name not in uv_sensor_mapping[uv_name]:
|
|
uv_sensor_mapping[uv_name][sensor_name] = pl.s_artinr
|
|
|
|
for artnr in sens2cable[pl.id]:
|
|
if is_meterware:
|
|
artnr_to_meterware_length[artnr] += math.ceil(length_m)
|
|
uv_to_artnr_to_meterware_length[uv_name][artnr] += math.ceil(length_m)
|
|
else: # Stückware
|
|
artnr_to_stückware_count[artnr] += 1
|
|
uv_to_artnr_to_stückware_count[uv_name][artnr] += 1
|
|
|
|
# UV-spezifische Sensorzählung basierend auf dem Mapping
|
|
for uv_name, sensor_dict in uv_sensor_mapping.items():
|
|
for artnr in sensor_dict.values():
|
|
uv_to_artnr_to_sensor_count[uv_name][artnr] += 1
|
|
|
|
return {
|
|
"artnr_to_stückware_count": artnr_to_stückware_count,
|
|
"artnr_to_meterware_length": artnr_to_meterware_length,
|
|
"artnr_to_sensor_count": artnr_to_sensor_count,
|
|
"uv_to_artnr_to_stückware_count": uv_to_artnr_to_stückware_count,
|
|
"uv_to_artnr_to_meterware_length": uv_to_artnr_to_meterware_length,
|
|
"uv_to_artnr_to_sensor_count": uv_to_artnr_to_sensor_count,
|
|
}
|
|
|
|
def _create_cable_list_sheet(ws, plines, sens2cable, bezeichner_cfg):
|
|
"""
|
|
Erstellt das Arbeitsblatt 'Length by ID' mit allen Kabeln und deren Längen.
|
|
|
|
Args:
|
|
ws (Worksheet): Das Excel-Arbeitsblatt.
|
|
plines (Polylines): Die Polylinien- und Geometriedaten.
|
|
sens2cable (dict): Mapping von Sensor-IDs zu Kabel-Artikelnummern.
|
|
bezeichner_cfg (ConfigParser): Konfiguration mit Artikelnummern-Namen.
|
|
"""
|
|
ws.title = "Length by ID"
|
|
ws.append(["Cable-ID", "True Length (m)", "Cable-ArtNr", "Cable-Name (short)"])
|
|
ws.column_dimensions['A'].width = 18
|
|
ws.column_dimensions['B'].width = 15
|
|
ws.column_dimensions['C'].width = 15
|
|
ws.column_dimensions['D'].width = 25
|
|
|
|
for pl in plines.kabel:
|
|
length = round(pl.length / 1000, 1)
|
|
for artnr in sens2cable[pl.id]:
|
|
cable_name = bezeichner_cfg["Sivasnummern"].get(artnr, "") if artnr.isdigit() else ""
|
|
ws.append([pl.id, length, artnr, cable_name])
|
|
|
|
def _create_cable_summary_sheet(ws, processed_data):
|
|
"""
|
|
Erstellt das Arbeitsblatt 'Cables SIVAS' mit einer Zusammenfassung der Kabel nach Artikelnummer.
|
|
|
|
Args:
|
|
ws (Worksheet): Das Excel-Arbeitsblatt.
|
|
processed_data (dict): Aggregierte Kabeldaten.
|
|
"""
|
|
ws.title = "Cables SIVAS"
|
|
ws.append(["Cable-ArtNr", "Amount (pcs)", "Cumm. Length (m)"])
|
|
ws.column_dimensions['A'].width = 20
|
|
ws.column_dimensions['B'].width = 12
|
|
ws.column_dimensions['C'].width = 15
|
|
|
|
count_summary = processed_data["artnr_to_stückware_count"]
|
|
length_summary = processed_data["artnr_to_meterware_length"]
|
|
all_artnrs = sorted(set(count_summary.keys()) | set(length_summary.keys()))
|
|
|
|
for artnr in all_artnrs:
|
|
ws.append([
|
|
artnr,
|
|
count_summary.get(artnr, ""),
|
|
length_summary.get(artnr, "")
|
|
])
|
|
|
|
def _create_rack_lengths_sheet(ws, plines):
|
|
"""
|
|
Erstellt das Arbeitsblatt 'Rack-Lengths' mit den Längen der Racks.
|
|
|
|
Args:
|
|
ws (Worksheet): Das Excel-Arbeitsblatt.
|
|
plines (Polylines): Die Polylinien- und Geometriedaten.
|
|
"""
|
|
ws.title = "Rack-Lengths"
|
|
ws.append(["Rack-ID", "Length (m)"])
|
|
ws.column_dimensions['A'].width = 18
|
|
ws.column_dimensions['B'].width = 15
|
|
|
|
for rackname, rack_geom in plines.rack_geometry.items():
|
|
classifier = rackname[0]
|
|
if classifier in ("t", "v", "c", "d"):
|
|
continue
|
|
else:
|
|
ws.append([rackname, rack_geom.length])
|
|
|
|
def _create_error_sheets(wb, error_collector: ErrorCollector = None):
|
|
"""
|
|
Erstellt die Arbeitsblätter für alle aufgetretenen Fehler (Equipment-Connection, Routing, Attribute).
|
|
|
|
Args:
|
|
wb (Workbook): Das Excel-Workbook.
|
|
error_collector (ErrorCollector, optional): ErrorCollector mit Fehlern und Warnungen.
|
|
"""
|
|
if not error_collector:
|
|
return
|
|
|
|
errors = error_collector.errors
|
|
warnings = error_collector.warnings
|
|
|
|
# Sheet: ERR-Equipment-Connection
|
|
errors_sensors = errors.get('sensors_connection', [])
|
|
errors_dists = errors.get('dists_connection', [])
|
|
|
|
if errors_sensors or errors_dists:
|
|
ws = wb.create_sheet("ERR-Equipment-Connection")
|
|
ws.append(["Type", "ID", "x", "y"])
|
|
ws.column_dimensions['A'].width = 20
|
|
for error in errors_sensors:
|
|
ws.append(["Sensor / Actuator", error['name'], int(error['coords']['x']), int(error['coords']['y'])])
|
|
for error in errors_dists:
|
|
ws.append(["Subdistributor", error['name'], int(error['coords']['x']), int(error['coords']['y'])])
|
|
|
|
# Sheet: ERR-Routing
|
|
errors_routing = errors.get('routing', [])
|
|
errors_dists_not_in_layout = errors.get('dists_not_in_layout', [])
|
|
|
|
if errors_routing:
|
|
ws = wb.create_sheet("ERR-Routing")
|
|
ws.append(["Subdistributor", "Sensor / Actuator", "Details"])
|
|
ws.column_dimensions['A'].width = 20
|
|
ws.column_dimensions['B'].width = 20
|
|
ws.column_dimensions['C'].width = 50
|
|
|
|
nicht_angebunden = {e['name'] for e in errors_sensors + errors_dists}
|
|
|
|
for error in errors_routing:
|
|
uv = error['unterverteiler']
|
|
uv_nicht_angebunden = uv in nicht_angebunden
|
|
|
|
if uv in errors_dists_not_in_layout:
|
|
ws.append([uv, "-", "Distributor not found in given layout."])
|
|
continue
|
|
|
|
for sensor in 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)"
|
|
ws.append([uv, sensor, grund])
|
|
|
|
# Sheet: ERR-Attributes
|
|
missing_attributes = warnings.get('missing_attributes', {})
|
|
|
|
if missing_attributes:
|
|
ws = wb.create_sheet("ERR-Attributes")
|
|
ws.append(["ID", "Error Detail"])
|
|
ws.column_dimensions['B'].width = 35
|
|
for sname, err_msg in missing_attributes.items():
|
|
ws.append([sname, err_msg])
|
|
|
|
# Sheet: WARNINGS (für allgemeine Warnungen wie z-Koordinaten-Abweichungen)
|
|
general_warnings = {k: v for k, v in warnings.items() if k != 'missing_attributes'}
|
|
|
|
if general_warnings:
|
|
ws = wb.create_sheet("WARNINGS")
|
|
ws.append(["Warning Type", "Details"])
|
|
ws.column_dimensions['A'].width = 30
|
|
ws.column_dimensions['B'].width = 80
|
|
|
|
for warning_type, warning_msg in general_warnings.items():
|
|
# Formatiere den Warning Type lesbarer
|
|
readable_type = warning_type.replace('_', ' ').title()
|
|
ws.append([readable_type, str(warning_msg)])
|
|
|
|
def _create_bom_workbook(outpath, processed_data, bezeichner_cfg):
|
|
"""
|
|
Erstellt eine separate Excel-Arbeitsmappe für die Stückliste (BOM) und speichert sie ab.
|
|
|
|
Args:
|
|
outpath (str): Basis-Pfad für die Ausgabedatei.
|
|
processed_data (dict): Aggregierte Kabel- und Sensor-Daten.
|
|
bezeichner_cfg (ConfigParser): Konfiguration mit Artikelnummern-Namen.
|
|
"""
|
|
wb = Workbook()
|
|
|
|
# Daten aus dem verarbeiteten Dictionary extrahieren
|
|
sensor_counts = processed_data["artnr_to_sensor_count"]
|
|
stückware_counts = processed_data["artnr_to_stückware_count"]
|
|
meterware_lengths = processed_data["artnr_to_meterware_length"]
|
|
|
|
uv_sensor_counts = processed_data["uv_to_artnr_to_sensor_count"]
|
|
uv_stückware_counts = processed_data["uv_to_artnr_to_stückware_count"]
|
|
uv_meterware_lengths = processed_data["uv_to_artnr_to_meterware_length"]
|
|
|
|
# Sheet 1: Globale BOM
|
|
ws1 = wb.active
|
|
ws1.title = "BOM"
|
|
ws1.append(["Art.-Number", "Amount (pcs)", "Length (m)", "Name (SIVAS)"])
|
|
ws1.column_dimensions['A'].width = 15
|
|
ws1.column_dimensions['B'].width = 12
|
|
ws1.column_dimensions['C'].width = 10
|
|
ws1.column_dimensions['D'].width = 50
|
|
|
|
# Sensoren auflisten
|
|
for artnr in sorted(sensor_counts.keys()):
|
|
ws1.append([artnr, sensor_counts[artnr], "", _get_sivas_name(artnr, bezeichner_cfg)])
|
|
|
|
# Kabel auflisten
|
|
all_cable_artnrs = sorted(set(stückware_counts.keys()) | set(meterware_lengths.keys()))
|
|
for artnr in all_cable_artnrs:
|
|
ws1.append([
|
|
artnr,
|
|
stückware_counts.get(artnr, ""),
|
|
meterware_lengths.get(artnr, ""),
|
|
_get_sivas_name(artnr, bezeichner_cfg)
|
|
])
|
|
|
|
# Sheet 2: BOM nach UV
|
|
ws2 = wb.create_sheet("BOM by UV")
|
|
ws2.append(["UV", "Art.-Number", "Amount (pcs)", "Length (m)", "Name (SIVAS)"])
|
|
ws2.column_dimensions['A'].width = 15
|
|
ws2.column_dimensions['B'].width = 15
|
|
ws2.column_dimensions['C'].width = 12
|
|
ws2.column_dimensions['D'].width = 10
|
|
ws2.column_dimensions['E'].width = 50
|
|
|
|
all_uvs = sorted(set(uv_sensor_counts.keys()) | set(uv_stückware_counts.keys()) | set(uv_meterware_lengths.keys()))
|
|
|
|
for uv in all_uvs:
|
|
all_artnrs_in_uv = sorted(
|
|
set(uv_sensor_counts.get(uv, {}).keys()) |
|
|
set(uv_stückware_counts.get(uv, {}).keys()) |
|
|
set(uv_meterware_lengths.get(uv, {}).keys())
|
|
)
|
|
for artnr in all_artnrs_in_uv:
|
|
name = _get_sivas_name(artnr, bezeichner_cfg)
|
|
sensor_amount = uv_sensor_counts.get(uv, {}).get(artnr, 0)
|
|
cable_count = uv_stückware_counts.get(uv, {}).get(artnr, 0)
|
|
cable_length = uv_meterware_lengths.get(uv, {}).get(artnr, 0)
|
|
|
|
if sensor_amount > 0:
|
|
ws2.append([uv, artnr, sensor_amount, "", name])
|
|
if cable_count > 0:
|
|
ws2.append([uv, artnr, cable_count, "", name])
|
|
if cable_length > 0:
|
|
ws2.append([uv, artnr, "", cable_length, name])
|
|
|
|
bom_path = outpath.replace("_cables.xlsx", "_BOM.xlsx")
|
|
wb.save(bom_path)
|
|
print(f"BOM exported to Excel-file")
|
|
|
|
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):
|
|
"""
|
|
Kopiert bestimmte Layer (nach Filter) von einer Quell-DXF in eine Ziel-DXF.
|
|
|
|
Args:
|
|
dxf_source (ezdxf.document.Drawing): Quell-DXF-Dokument.
|
|
dxf_target (ezdxf.document.Drawing): Ziel-DXF-Dokument.
|
|
"""
|
|
|
|
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.
|
|
|
|
Args:
|
|
section (str): Name der Kabel-Section in der Config.
|
|
length (float): Kabellänge in Metern.
|
|
|
|
Returns:
|
|
tuple: (True, artnr) bei Erfolg, (False, Fehlermeldung) bei Fehler.
|
|
"""
|
|
# Existiert über BMK vergebene Kabel-Kennzeichnung in kabel.cfg?
|
|
if section not in cable_cfg:
|
|
return False, f"Keine Kabelkennzeichnung '{section}' in kabel.cfg"
|
|
|
|
entries = cable_cfg[section]
|
|
length_keys = sorted([float(k) for k in entries.keys()])
|
|
|
|
for l in length_keys:
|
|
if length <= l:
|
|
return True, entries[str(l)]
|
|
|
|
return False, f"Kabel länger als max. Kabellänge in Sektion '{section}'"
|
|
|
|
def map_sensor_to_cable_cfg(plines):
|
|
"""
|
|
Ordnet jedem Sensor/Kabel die passende Kabel-Artikelnummer gemäß BMK-Config und Kabel-Config zu.
|
|
|
|
Args:
|
|
plines (Polylines): Die Polylinien- und Geometriedaten.
|
|
|
|
Returns:
|
|
dict: Mapping von Kabel-IDs zu Artikelnummern.
|
|
"""
|
|
sens2cable = defaultdict(list)
|
|
|
|
mapping = config_BMK["Cable-Mapping"]
|
|
|
|
for pl in plines.kabel:
|
|
sensor_name = '-'.join(pl.id.split('-')[1:])
|
|
cable_length = round(pl.length/1000, 1)
|
|
sensor_artinr = pl.s_artinr
|
|
# SPS Prefex
|
|
name_prefix = sensor_name[:2]
|
|
|
|
# Suche nach Key in der BMK-Config
|
|
|
|
key_with_artnr = f"{name_prefix}-{sensor_artinr}" # Spezialfälle über "Key mit Artikelnummer" abgleichen
|
|
if key_with_artnr in mapping:
|
|
section_list = mapping[key_with_artnr]
|
|
elif name_prefix in mapping: # Standardzuweisung
|
|
section_list = mapping[name_prefix]
|
|
else:
|
|
sens2cable[pl.id].append("Kein Kabeltyp zugewiesen (BMK.cfg)")
|
|
section_list = list()
|
|
# TODO was soll hier passieren wenn die Config falsch ist?
|
|
|
|
# Liste aus evtl. mehreren Sektionen erzeugen
|
|
sections = [s.strip() for s in section_list.split(",")]
|
|
|
|
# Evtl. Kabelkürzung durchführen, falls 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:
|
|
sucess, result = get_cable_artnr(section, cable_length)
|
|
sens2cable[pl.id].append(result)
|
|
|
|
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('-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', metavar='original.dxf')
|
|
parser.add_argument('-l', '--local', action='store_true', help='using only local data for naming of article numbers. If not set: fetching names from SIVAS.')
|
|
|
|
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, error_collector = 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"))
|
|
|
|
bezeichner_cfg = configparser.ConfigParser()
|
|
bezeichner_cfg.optionxform = str #Keys case-sensitive
|
|
with open(os.path.join(config_dir, "bezeichner.cfg"), encoding="utf-8") as f:
|
|
bezeichner_cfg.read_file(f)
|
|
|
|
|
|
dxf_file = args.dxf
|
|
|
|
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.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:
|
|
# 1. Kabelzuordnung
|
|
sens2cable = map_sensor_to_cable_cfg(plines)
|
|
|
|
# 2. verwendete Artikelnummer (Sensoren + Kabel)
|
|
all_artnrs = get_all_artnrs(plines, sens2cable)
|
|
|
|
# 3. Fehlende Bezeichner in bezeichner.cfg markieren
|
|
mark_missings(all_artnrs)
|
|
|
|
# 4. Bezeichner ggf. aus SIVAS abrufen
|
|
if args.local == False:
|
|
config_path=os.path.join(config_dir, "bezeichner.cfg")
|
|
uc.update_bezeichner(bezeichner_cfg, config_path)
|
|
|
|
# 5. Excel schreiben
|
|
excel_path = os.path.join(work_dir, args.excel)
|
|
write_excel_from_json(plines, sens2cable, excel_path, error_collector)
|
|
|
|
# 6. Optionale Fehlerdatei schreiben
|
|
if error_collector.has_errors():
|
|
basename = Path(args.excel).stem
|
|
error_filename = f"{basename}_errors.json"
|
|
error_collector.write_errorfile(Path(work_dir), error_filename) |