391 lines
13 KiB
Python
391 lines
13 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 Polylines:
|
||
kabel: List[Polyline]
|
||
|
||
|
||
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 dxf ..")
|
||
|
||
doc = ezdxf.new('R2018', setup=True)
|
||
create_cables(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 to original .dxf ..")
|
||
|
||
doc = ezdxf.readfile(originaldxf)
|
||
create_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 from original .dxf into new .dxf ..")
|
||
|
||
quelle = ezdxf.readfile(originaldxf)
|
||
ziel = ezdxf.new('R2018', setup=True)
|
||
|
||
copy_layers_into_dxf_by_filter(quelle, ziel)
|
||
create_cables(plines, quelle)
|
||
|
||
ziel.saveas(out_path)
|
||
print("done")
|
||
|
||
|
||
def create_cables(plines, doc):
|
||
msp = doc.modelspace()
|
||
# # Beispiel für Text und Polyline
|
||
# add_text(msp)
|
||
# Layer mit Zeitstempel erstellen (z. B. "Layer_2025-05-07_14-30")
|
||
timestamp = datetime.now().strftime("Kabel_%Y-%m-%d_%H-%M")
|
||
if timestamp not in doc.layers:
|
||
doc.layers.add(name=timestamp, color=7)
|
||
|
||
dxfattribs={"layer": timestamp}
|
||
for pl in plines.kabel:
|
||
add_polyline(msp, pl, dxfattribs)
|
||
|
||
|
||
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 ..")
|
||
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 = "Kabellängen"
|
||
ws1.append(["Kabel-ID", "Länge aus Json (m)", "Gerundete Länge (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("Längenübersicht")
|
||
ws2.append(["Gerundete Länge (m)", "Anzahl Kabel"])
|
||
|
||
for rlength in sorted(length_summary):
|
||
ws2.append([int(rlength), length_summary[rlength]])
|
||
|
||
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 collect_text_entities(msp):
|
||
texts = []
|
||
for entity in msp.query("TEXT MTEXT"):
|
||
try:
|
||
text = entity.dxf.text
|
||
pos = tuple(round(c, 4) for c in entity.dxf.insert)
|
||
layer = entity.dxf.layer
|
||
texts.append((text, pos, layer))
|
||
except Exception as e:
|
||
print(f"Fehler beim Lesen eines Textobjekts: {e}")
|
||
return texts
|
||
|
||
def compare_texts(source_texts, target_texts):
|
||
missing_in_target = [t for t in source_texts if t not in target_texts]
|
||
extra_in_target = [t for t in target_texts if t not in source_texts]
|
||
|
||
if not missing_in_target and not extra_in_target:
|
||
print("✅ Alle TEXT/MTEXT-Objekte wurden korrekt kopiert.")
|
||
else:
|
||
if missing_in_target:
|
||
print("\n❌ Fehlende Texte im Ziel:")
|
||
for text in missing_in_target:
|
||
print(f" - {text}")
|
||
if extra_in_target:
|
||
print("\n⚠️ Zusätzliche Texte im Ziel:")
|
||
for text in extra_in_target:
|
||
print(f" - {text}")
|
||
|
||
def collect_block_references(doc):
|
||
msp = doc.modelspace()
|
||
blocks = []
|
||
for insert in msp.query("INSERT"):
|
||
try:
|
||
name = insert.dxf.name
|
||
pos = tuple(round(c, 4) for c in insert.dxf.insert)
|
||
layer = insert.dxf.layer
|
||
blocks.append((name, pos, layer))
|
||
except Exception as e:
|
||
print(f"Fehler beim Lesen eines INSERT-Objekts: {e}")
|
||
return blocks
|
||
|
||
def serialize_entity(e):
|
||
"""Wandelt eine Entity in ein vergleichbares Tupel um"""
|
||
dxfattribs = {}
|
||
for key in e.dxf.__dir__():
|
||
try:
|
||
value = getattr(e.dxf, key)
|
||
if callable(value) or key.startswith("_"):
|
||
continue
|
||
if isinstance(value, (list, tuple)):
|
||
value = tuple(round(v, 4) for v in value)
|
||
elif isinstance(value, float):
|
||
value = round(value, 4)
|
||
dxfattribs[key] = value
|
||
except Exception:
|
||
continue
|
||
return (e.dxftype(), dxfattribs)
|
||
|
||
def compare_blocks(source_doc, target_doc, used_block_names):
|
||
for name in used_block_names:
|
||
if name not in source_doc.blocks or name not in target_doc.blocks:
|
||
print(f"❌ Block '{name}' fehlt in einer der Dateien.")
|
||
continue
|
||
|
||
source_block = source_doc.blocks.get(name)
|
||
target_block = target_doc.blocks.get(name)
|
||
|
||
source_entities = [serialize_entity(e) for e in source_block]
|
||
target_entities = [serialize_entity(e) for e in target_block]
|
||
|
||
missing_in_target = [e for e in source_entities if e not in target_entities]
|
||
extra_in_target = [e for e in target_entities if e not in source_entities]
|
||
|
||
if not missing_in_target and not extra_in_target:
|
||
print(f"✅ Block '{name}' ist identisch.")
|
||
else:
|
||
print(f"\n🔍 Block '{name}' hat Unterschiede:")
|
||
if missing_in_target:
|
||
print(f" ❌ Fehlt im Ziel:")
|
||
for e in missing_in_target:
|
||
print(f" - {e}")
|
||
if extra_in_target:
|
||
print(f" ⚠️ Zusätzlich im Ziel:")
|
||
for e in extra_in_target:
|
||
print(f" - {e}")
|
||
|
||
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)
|
||
|
||
'''
|
||
# 2. alle Blockdefinitionen kopieren die benötigt werden
|
||
used_blocks = {ins.dxf.name for ins in msp_source.query("INSERT")}
|
||
for name in used_blocks:
|
||
# skip vorhandene
|
||
if name in dxf_target.blocks:
|
||
continue
|
||
src_block = dxf_source.blocks.get(name)
|
||
# neues BLOCK-Objekt anlegen mit Base-Point
|
||
new_block = dxf_target.blocks.new(name=name, base_point=src_block.base_point)
|
||
# **ALLE** Entities im Block kopieren
|
||
for ent in src_block:
|
||
new_block.add_entity(ent.copy())
|
||
|
||
|
||
# 3. Jetzt die INSERTs in den Modelspace kopieren
|
||
for ins in msp_source.query("INSERT"):
|
||
new_ins = ins.copy()
|
||
msp_target.add_entity(new_ins)
|
||
|
||
'''
|
||
|
||
# 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())
|
||
|
||
texts_source = collect_text_entities(msp_source)
|
||
texts_target = collect_text_entities(msp_target)
|
||
|
||
compare_texts(texts_source, texts_target)
|
||
|
||
used_block_names = set(insert.dxf.name for insert in dxf_source.modelspace().query("INSERT"))
|
||
compare_blocks(dxf_source, dxf_target, used_block_names)
|
||
|
||
|
||
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)
|
||
|
||
|
||
|