484 lines
19 KiB
Python
484 lines
19 KiB
Python
"""
|
||
placeblocks.py
|
||
Erzeugt DXF-Elemente aus einer RuleDesigner-CSV.
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import csv
|
||
import json
|
||
import re
|
||
import argparse
|
||
import configparser
|
||
import ezdxf
|
||
from pathlib import Path
|
||
import math
|
||
from utils import check_environment_var, setup_logger
|
||
|
||
# --------------------------------------------------------- CFG-Leser für shapes.cfg
|
||
def get_shape_cfg(teileart, cfg_path, logger=None):
|
||
parser = configparser.ConfigParser()
|
||
try:
|
||
with open(cfg_path, encoding='utf-8') as f:
|
||
parser.read_file(f)
|
||
except Exception as e:
|
||
msg = f"Fehler beim Lesen der Config-Datei {cfg_path}: {e}"
|
||
if logger:
|
||
logger.error(msg)
|
||
else:
|
||
print(msg)
|
||
return []
|
||
section = teileart
|
||
if section not in parser:
|
||
return []
|
||
# Blöcke
|
||
items = parser.get(section, "items", fallback="").replace('"', '').split(",")
|
||
blocks = [item.strip() for item in items if item.strip()]
|
||
symbols = []
|
||
for i, name in enumerate(blocks):
|
||
# Offset
|
||
offset_key = f"offset_symb{i+1}"
|
||
offset_str = parser.get(section, offset_key, fallback="0,0")
|
||
try:
|
||
ox, oy = [float(x) for x in offset_str.split(",")]
|
||
except Exception:
|
||
ox, oy = 0.0, 0.0
|
||
# Rotation
|
||
rot_key = f"rot_symb{i+1}"
|
||
rot_str = parser.get(section, rot_key, fallback="0.0")
|
||
try:
|
||
rot = float(rot_str)
|
||
except Exception:
|
||
rot = 0.0
|
||
symbols.append({
|
||
"name": name,
|
||
"offset": (ox, oy),
|
||
"rotation": rot
|
||
})
|
||
return symbols
|
||
|
||
# --------------------------------------------------------- Konstante Parameter
|
||
ATTR_TAG = "TeileId" # Attributtag im Block
|
||
RADIUS = 400 # Radius der Kreiselkreise (mm)
|
||
|
||
# --------------------------------------------------------- Hilfsfunktionen
|
||
def extract_coords(planquadrat: str) -> tuple[float, float]:
|
||
"""Extrahiert X/Y Koordinaten aus PlanquadratString."""
|
||
m = re.search(r"X:(\d+[\.,]?\d*)\s+Y:(\d+[\.,]?\d*)", planquadrat)
|
||
if not m:
|
||
raise ValueError(f"Koordinaten nicht gefunden in: '{planquadrat}'")
|
||
x, y = m.groups()
|
||
return float(x.replace(",", ".")), float(y.replace(",", "."))
|
||
|
||
def parse_merkmale(merkmale_str: str) -> dict:
|
||
"""Parst Merkmale-JSON-String in dict; bei Fehler → leeres Dict."""
|
||
try:
|
||
return json.loads(merkmale_str)
|
||
except json.JSONDecodeError:
|
||
return {}
|
||
|
||
def import_block(block_name: str, from_doc, to_doc) -> None:
|
||
"""Importiert Blockdefinition block_name von from_doc nach to_doc."""
|
||
if block_name in to_doc.blocks:
|
||
return
|
||
if block_name not in from_doc.blocks:
|
||
raise ValueError(f"Block '{block_name}' nicht in Bibliothek gefunden.")
|
||
src = from_doc.blocks[block_name]
|
||
tgt = to_doc.blocks.new(name=block_name)
|
||
for ent in src:
|
||
tgt.add_entity(ent.copy())
|
||
|
||
def berechne_hoehe(csv_path, logger=None):
|
||
y_werte = []
|
||
try:
|
||
with csv_path.open(newline="", encoding="utf-8") as fh:
|
||
reader = csv.DictReader(fh, delimiter=';')
|
||
for row in reader:
|
||
planquadrat = row.get("Planquadrat", "")
|
||
try:
|
||
_, y = extract_coords(planquadrat)
|
||
y_werte.append(y)
|
||
except Exception as e:
|
||
if logger:
|
||
logger.warning(f"Fehler beim Extrahieren der Koordinate aus '{planquadrat}': {e}")
|
||
except Exception as e:
|
||
if logger:
|
||
logger.error(f"Fehler beim Lesen der CSV-Datei {csv_path}: {e}")
|
||
else:
|
||
print(f"Fehler beim Lesen der CSV-Datei {csv_path}: {e}")
|
||
return 0
|
||
if not y_werte:
|
||
msg = "Keine Y-Koordinaten in der CSV gefunden!"
|
||
if logger:
|
||
logger.error(msg)
|
||
else:
|
||
print(msg)
|
||
raise ValueError(msg)
|
||
return max(y_werte)
|
||
|
||
def transform_coords(x: float, y: float, height: float) -> tuple[float, float]:
|
||
"""Transformiert Bildschirmkoordinaten (0,0 oben links) ins DXF-KoSy (0,0 unten links)."""
|
||
return x, height - y
|
||
|
||
def handle_ils_2_0_kreisel(msp, teileid, merkmale, x, y, doc, lib_doc, verbose, symbols):
|
||
abstand_m = merkmale.get(
|
||
"Abstand (Kreiselachse A - Kreiselachse) in Meter", "20"
|
||
).replace(",", ".")
|
||
try:
|
||
abstand = float(abstand_m) * 1000 # Meter → mm
|
||
except ValueError:
|
||
abstand = 10000 # Fallback 10 m
|
||
|
||
# Drehung (Winkel in Grad, Standard 0) aus Merkmale
|
||
try:
|
||
winkel = float(merkmale.get("Drehung", 0))
|
||
except (ValueError, TypeError):
|
||
winkel = 0.0
|
||
winkel_rad = math.radians(winkel)
|
||
|
||
# Die Koordinaten (x, y) sind die Mitte zwischen den beiden Blöcken (bereits transformiert)
|
||
halbabstand = abstand / 2
|
||
dx = halbabstand * math.cos(winkel_rad)
|
||
dy = halbabstand * math.sin(winkel_rad)
|
||
pos1 = (x - dx, y - dy)
|
||
pos2 = (x + dx, y + dy)
|
||
positions = [pos1, pos2]
|
||
for i, sym in enumerate(symbols):
|
||
blockname = sym["name"]
|
||
offset = sym["offset"]
|
||
rotation = sym["rotation"]
|
||
if i < len(positions):
|
||
pos = (positions[i][0] + offset[0], positions[i][1] + offset[1])
|
||
import_block(blockname, lib_doc, doc)
|
||
bref = msp.add_blockref(blockname, pos, dxfattribs={"rotation": rotation})
|
||
bref.add_auto_attribs({ATTR_TAG: teileid})
|
||
if verbose:
|
||
print(f"[INFO] Block '{blockname}' (Kreisel) → {teileid} "
|
||
f"({pos[0]:.1f}, {pos[1]:.1f}), rot={rotation}")
|
||
# Linien zeichnen
|
||
draw_kreisel_lines(msp, pos1, pos2)
|
||
draw_kreisel_drehrichtung_markierung(msp, pos1, pos2, merkmale, lib_doc, doc, verbose)
|
||
|
||
def draw_kreisel_lines(msp, pos1, pos2):
|
||
"""Zeichnet tangentiale Linien zwischen zwei Kreiselblöcken, unabhängig vom Winkel."""
|
||
x1, y1 = pos1
|
||
x2, y2 = pos2
|
||
# Verbindungsvektor
|
||
dx = x2 - x1
|
||
dy = y2 - y1
|
||
# Länge
|
||
length = math.hypot(dx, dy)
|
||
if length == 0:
|
||
return # keine Linie bei identischen Punkten
|
||
# Normalenvektor (senkrecht, normiert, Länge = RADIUS)
|
||
nx = -dy / length * RADIUS
|
||
ny = dx / length * RADIUS
|
||
# Tangentialpunkte
|
||
p1a = (x1 + nx, y1 + ny)
|
||
p1b = (x1 - nx, y1 - ny)
|
||
p2a = (x2 + nx, y2 + ny)
|
||
p2b = (x2 - nx, y2 - ny)
|
||
# Linien zeichnen
|
||
msp.add_line(p1a, p2a)
|
||
msp.add_line(p1b, p2b)
|
||
|
||
def draw_kreisel_drehrichtung_markierung(msp, pos1, pos2, merkmale, lib_doc, doc, verbose):
|
||
drehrichtung = merkmale.get("Drehrichtung", "").upper()
|
||
if drehrichtung not in ("UZS", "GUZS"):
|
||
return
|
||
x1, y1 = pos1
|
||
x2, y2 = pos2
|
||
dx = x2 - x1
|
||
dy = y2 - y1
|
||
length = math.hypot(dx, dy)
|
||
if length == 0:
|
||
return
|
||
# Normalenvektor (senkrecht, normiert, Länge = RADIUS)
|
||
nx = -dy / length * RADIUS
|
||
ny = dx / length * RADIUS
|
||
# Obere Linie
|
||
p1_oben = (x1 + nx, y1 + ny)
|
||
p2_oben = (x2 + nx, y2 + ny)
|
||
# Untere Linie
|
||
p1_unten = (x1 - nx, y1 - ny)
|
||
p2_unten = (x2 - nx, y2 - ny)
|
||
# S-LP auf oberer Linie (Drehrichtung wie angegeben)
|
||
for i in range(1, 4):
|
||
t = i / 4 # 1/4, 2/4, 3/4
|
||
px = p1_oben[0] + t * (p2_oben[0] - p1_oben[0])
|
||
py = p1_oben[1] + t * (p2_oben[1] - p1_oben[1])
|
||
rotation = math.degrees(math.atan2(p2_oben[1] - p1_oben[1], p2_oben[0] - p1_oben[0]))
|
||
if drehrichtung == "GUZS":
|
||
rotation += 180
|
||
import_block("S-LP", lib_doc, doc)
|
||
bref = msp.add_blockref("S-LP", (px, py), dxfattribs={"rotation": rotation})
|
||
if verbose:
|
||
print(f"[INFO] Drehrichtung '{drehrichtung}': S-LP oben bei ({px:.1f}, {py:.1f}), rot={rotation:.1f}")
|
||
# S-LP auf unterer Linie (Drehrichtung invertiert)
|
||
for i in range(1, 4):
|
||
t = i / 4
|
||
px = p1_unten[0] + t * (p2_unten[0] - p1_unten[0])
|
||
py = p1_unten[1] + t * (p2_unten[1] - p1_unten[1])
|
||
rotation = math.degrees(math.atan2(p2_unten[1] - p1_unten[1], p2_unten[0] - p1_unten[0]))
|
||
if drehrichtung == "UZS":
|
||
rotation += 180
|
||
import_block("S-LP", lib_doc, doc)
|
||
bref = msp.add_blockref("S-LP", (px, py), dxfattribs={"rotation": rotation})
|
||
if verbose:
|
||
print(f"[INFO] Drehrichtung '{drehrichtung}': S-LP unten bei ({px:.1f}, {py:.1f}), rot={rotation:.1f}")
|
||
|
||
def handle_standard(msp, blocknames, teileid, x, y, lib_doc, doc, verbose):
|
||
for blockname in blocknames:
|
||
import_block(blockname, lib_doc, doc)
|
||
bref = msp.add_blockref(blockname, (x, y))
|
||
bref.add_auto_attribs({ATTR_TAG: teileid})
|
||
if verbose:
|
||
print(f"[INFO] Block '{blockname}' (Standard) → {teileid} "
|
||
f"({x:.1f}, {y:.1f})")
|
||
|
||
def handle_ils_2_0_gefaellestrecke(msp, teileid, merkmale, x, y, doc, lib_doc, verbose, symbols):
|
||
# blocks: [block1, block2], offsets: [(ox1, oy1), (ox2, oy2)]
|
||
# Länge der Strecke (in Meter, Standard 10)
|
||
laenge_m = merkmale.get("Länge in Meter", "10").replace(",", ".")
|
||
try:
|
||
laenge = float(laenge_m) * 1000 # Meter → mm
|
||
except ValueError:
|
||
laenge = 10000 # Fallback 10 m
|
||
|
||
# Drehung (Winkel in Grad, Standard 0)
|
||
try:
|
||
winkel = float(merkmale.get("Drehung", 0))
|
||
except (ValueError, TypeError):
|
||
winkel = 0.0
|
||
winkel_rad = math.radians(winkel)
|
||
|
||
# Die Koordinaten (x, y) sind die Mitte der Strecke
|
||
halbe_laenge = laenge / 2
|
||
dx = halbe_laenge * math.cos(winkel_rad)
|
||
dy = halbe_laenge * math.sin(winkel_rad)
|
||
start = (x - dx, y - dy)
|
||
ende = (x + dx, y + dy)
|
||
msp.add_line(start, ende)
|
||
if verbose:
|
||
print(f"[INFO] Gefällestrecke → {teileid} Linie von ({start[0]:.1f}, {start[1]:.1f}) nach ({ende[0]:.1f}, {ende[1]:.1f})")
|
||
|
||
# Blöcke am Anfang und Ende der Strecke aus der CFG platzieren
|
||
if len(symbols) >= 2 and lib_doc is not None:
|
||
for i, sym in enumerate(symbols[:2]):
|
||
blockname = sym["name"]
|
||
offset = sym["offset"]
|
||
rotation = sym["rotation"]
|
||
pos = (start[0] + offset[0], start[1] + offset[1]) if i == 0 else (ende[0] + offset[0], ende[1] + offset[1])
|
||
import_block(blockname, lib_doc, doc)
|
||
bref = msp.add_blockref(blockname, pos, dxfattribs={"rotation": rotation})
|
||
bref.add_auto_attribs({ATTR_TAG: teileid})
|
||
if verbose:
|
||
print(f"[INFO] Block '{blockname}' an {'Startpunkt' if i==0 else 'Endpunkt'} {pos} für {teileid}, rot={rotation}")
|
||
elif lib_doc is None:
|
||
print("[WARN] lib_doc nicht verfügbar, Blöcke werden nicht eingefügt.")
|
||
|
||
def handle_omniflo(msp, teileid, merkmale, x, y, doc, lib_doc, verbose, symbols):
|
||
"""
|
||
Für Omniflo Gerade: zeichnet eine Linie (Mitte = Koordinate, Länge und Winkel aus Merkmale).
|
||
Für alle anderen Omniflo-Typen: Block mit SivasNummer an den Koordinaten.
|
||
"""
|
||
# Prüfen, ob es sich um eine Gerade handelt
|
||
if merkmale.get("Länge in Meter") is not None and merkmale.get("Winkel") is not None:
|
||
try:
|
||
laenge = float(merkmale.get("Länge in Meter", "0").replace(",", ".")) * 1000 # Meter → mm
|
||
except Exception:
|
||
laenge = 0
|
||
try:
|
||
winkel = float(merkmale.get("Drehung", 0))
|
||
except Exception:
|
||
winkel = 0.0
|
||
winkel_rad = math.radians(winkel)
|
||
halbe_laenge = laenge / 2
|
||
dx = halbe_laenge * math.cos(winkel_rad)
|
||
dy = halbe_laenge * math.sin(winkel_rad)
|
||
start = (x - dx, y - dy)
|
||
ende = (x + dx, y + dy)
|
||
msp.add_line(start, ende)
|
||
if verbose:
|
||
print(f"[INFO] Omniflo Gerade → {teileid} Linie von ({start[0]:.1f}, {start[1]:.1f}) nach ({ende[0]:.1f}, {ende[1]:.1f})")
|
||
return
|
||
# Sonst wie gehabt: Block mit SivasNummer
|
||
if not lib_doc:
|
||
print("[WARN] lib_doc nicht verfügbar, Block wird nicht eingefügt.")
|
||
return
|
||
blockname = merkmale.get("SivasNummer")
|
||
if not blockname:
|
||
print(f"[WARN] Keine SivasNummer für {teileid}, überspringe.")
|
||
return
|
||
if blockname not in lib_doc.blocks:
|
||
print(f"[WARN] Omniflo-Block '{blockname}' nicht in Bibliothek {lib_doc.filename}. Überspringe {teileid}.")
|
||
return
|
||
import_block(blockname, lib_doc, doc)
|
||
bref = msp.add_blockref(blockname, (x, y))
|
||
bref.add_auto_attribs({ATTR_TAG: teileid})
|
||
if verbose:
|
||
print(f"[INFO] Block '{blockname}' (Omniflo) → {teileid} ({x:.1f}, {y:.1f})")
|
||
|
||
def normalize_func_name(name):
|
||
return (
|
||
name.replace('ä', 'ae')
|
||
.replace('ö', 'oe')
|
||
.replace('ü', 'ue')
|
||
.replace('ß', 'ss')
|
||
.replace(' ', '_')
|
||
.replace('.', '_')
|
||
.lower()
|
||
)
|
||
|
||
def get_libfile_cfg(teileart, cfg_path):
|
||
"""Liest den Bibliotheksdateinamen für eine TeileArt aus der allgemein.cfg."""
|
||
parser = configparser.ConfigParser()
|
||
with open(cfg_path, encoding='utf-8') as f:
|
||
parser.read_file(f)
|
||
# Teileart kann z.B. "ILS 2.0 Kreisel" sein, wir nehmen den ersten Teil vor erstem Leerzeichen oder Punkt
|
||
# oder suchen iterativ nach Sektionen, die im Teileart-Namen vorkommen
|
||
for section in parser.sections():
|
||
if section in teileart:
|
||
return parser.get(section, "libfile", fallback=None)
|
||
return None
|
||
|
||
# --------------------------------------------------------- Hauptfunktion
|
||
def main(csv_path: Path, lib_path: Path, cfg_path: Path,
|
||
output_path: Path, verbose=False, logger=None):
|
||
|
||
# Bibliothek nur laden, wenn Datei existiert
|
||
lib_doc = None
|
||
if lib_path.exists():
|
||
try:
|
||
lib_doc = ezdxf.readfile(lib_path)
|
||
if verbose:
|
||
logger.info(f"[INFO] Bibliothek geladen: {lib_path}") if logger else print(f"[INFO] Bibliothek geladen: {lib_path}")
|
||
except Exception as e:
|
||
msg = f"Fehler beim Lesen der Bibliothek '{lib_path}': {e}"
|
||
if logger:
|
||
logger.error(msg)
|
||
else:
|
||
print(msg)
|
||
sys.exit(1)
|
||
else:
|
||
msg = f"[INFO] Keine Bibliothek gefunden unter {lib_path}."
|
||
if logger:
|
||
logger.warning(msg)
|
||
else:
|
||
print(msg)
|
||
sys.exit(1)
|
||
|
||
# Neue Zielzeichnung (DXF R2018)
|
||
doc = ezdxf.new(dxfversion="R2018")
|
||
msp = doc.modelspace()
|
||
|
||
# Höhe bestimmen für Koordinaten-Transformation
|
||
try:
|
||
height = berechne_hoehe(csv_path, logger=logger)
|
||
except Exception as e:
|
||
msg = f"Fehler bei der Höhenberechnung: {e}"
|
||
if logger:
|
||
logger.error(msg)
|
||
else:
|
||
print(msg)
|
||
sys.exit(1)
|
||
|
||
# Verarbeitung der Blöcke
|
||
with csv_path.open(newline="", encoding="utf-8") as fh:
|
||
reader = csv.DictReader(fh, delimiter=';')
|
||
for row in reader:
|
||
teileart = row["TeileArt"].strip()
|
||
teileid = row["TeileId"].strip()
|
||
planquadrat = row["Planquadrat"]
|
||
merkmale = parse_merkmale(row.get("Merkmale", ""))
|
||
|
||
try:
|
||
x_screen, y_screen = extract_coords(planquadrat)
|
||
x, y = transform_coords(x_screen, y_screen, height)
|
||
except Exception as e:
|
||
msg = f"[WARN] {teileid}: {e}"
|
||
if logger:
|
||
logger.warning(msg)
|
||
else:
|
||
print(msg)
|
||
continue
|
||
|
||
# Bibliotheksdatei bestimmen
|
||
libfile = get_libfile_cfg(teileart, allgemein_cfg_path)
|
||
if libfile:
|
||
lib_path = blocklib_dir / libfile
|
||
else:
|
||
lib_path = default_lib_path
|
||
|
||
# Bibliothek laden (mit Cache)
|
||
lib_doc = None
|
||
if lib_path in lib_docs:
|
||
lib_doc = lib_docs[lib_path]
|
||
elif lib_path.exists():
|
||
try:
|
||
lib_doc = ezdxf.readfile(lib_path)
|
||
lib_docs[lib_path] = lib_doc
|
||
if verbose:
|
||
print(f"[INFO] Bibliothek geladen: {lib_path}")
|
||
except Exception as e:
|
||
print(f"[WARN] Fehler beim Lesen der Bibliothek '{lib_path}': {e}")
|
||
else:
|
||
print(f"[INFO] Keine Bibliothek gefunden unter {lib_path}. Komplexe Formen werden übersprungen.")
|
||
|
||
# Funktions-Dispatch: handle_<teileart> (mit _ statt Leerzeichen und Punkten, alles klein)
|
||
func_name = f'handle_{normalize_func_name(teileart)}'
|
||
handler = globals().get(func_name)
|
||
symbols = get_shape_cfg(teileart, shapes_cfg_path, logger=logger)
|
||
# Mapping für Omniflo-Typen
|
||
if func_name.startswith('handle_omniflo'):
|
||
handler = globals().get('handle_omniflo')
|
||
if handler:
|
||
handler(msp, teileid, merkmale, x, y, doc, lib_doc, verbose, symbols)
|
||
else:
|
||
msg = f"[WARN] Keine Routine für TeileArt '{teileart}'. Überspringe '{teileid}'."
|
||
if logger:
|
||
logger.warning(msg)
|
||
else:
|
||
print(msg)
|
||
continue
|
||
|
||
# DXF speichern
|
||
doc.saveas(output_path)
|
||
if logger:
|
||
logger.info(f"[DONE] DXF gespeichert unter: {output_path}")
|
||
else:
|
||
print(f"[DONE] DXF gespeichert unter: {output_path}")
|
||
|
||
if __name__ == "__main__":
|
||
parser = argparse.ArgumentParser(
|
||
description="Plaziert Anlagenkomponenten aus RuleDesigner CSV.")
|
||
parser.add_argument("-f", "--file", required=True, help="CSV-Datei (Name oder Pfad)", metavar="input.csv")
|
||
parser.add_argument("-c", "--config", help="CFG mit einfachen Formen", metavar="shapes.cfg")
|
||
parser.add_argument("-l", "--lib", help="DXF-Bibliothek mit Blöcken", metavar="bibliothek.dxf")
|
||
parser.add_argument("-o", "--output", help="Ziel-DXF (Standard: PROJECT_WORK/anlage.dxf)", metavar="anlage.dxf")
|
||
parser.add_argument("-v", "--verbose", action="store_true", help="mehr Ausgaben anzeigen")
|
||
args = parser.parse_args()
|
||
|
||
# Verzeichnisse aus Umgebungsvariablen
|
||
log_dir = check_environment_var("PROJECT_LOG")
|
||
data_dir = check_environment_var("PROJECT_DATA")
|
||
work_dir = check_environment_var("PROJECT_WORK")
|
||
config_dir = check_environment_var("PROJECT_CFG")
|
||
|
||
logger = setup_logger(log_dir, name='plant2dxf')
|
||
logger.info("=== plant2dxf Verarbeitung gestartet ===")
|
||
|
||
# CSV‑Pfad: nur Dateiname → im WORK‑Ordner suchen
|
||
if os.sep not in args.file and "/" not in args.file:
|
||
csv_path = work_dir / args.file
|
||
else:
|
||
csv_path = Path(args.file)
|
||
|
||
cfg_path = Path(args.config) if args.config else config_dir / "shapes.cfg"
|
||
allgemein_cfg_path = config_dir / "allgemein.cfg"
|
||
default_lib_path = Path(args.lib) if args.lib else data_dir / "blocks.dxf"
|
||
output_path = Path(args.output) if args.output else (work_dir / f"{csv_path.stem}.dxf")
|
||
|
||
main(csv_path, default_lib_path, cfg_path, output_path, verbose=args.verbose, logger=logger)
|
||
logger.info("=== plant2dxf Verarbeitung abgeschlossen ===")
|