Schoenenberger_Systeme_GmbH/dxfmakros
3
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity
Files
046d34c4aae44dffa484561811e50335629f6361
dxfmakros/lib/export_csv.py
T
m.stangl 9c8f84799c [CHANGE] Omniflo-Nachbarschaft: STRtree-Broad-Phase + KOS-Verfeinerung (10mm); Gerade-K1/K2 im CSV-Export 
Die Omniflo-Nachbarschaftserkennung (export_neighbors.py, nur EXPORTCSV) wird
zweistufig:
- Broad-Phase ueber shapely-STRtree auf den Bounding-Boxen (Fallback: das
  bisherige Raster, falls shapely fehlt).
- Narrow-Phase (KOS-Verfeinerung): zwei moegliche Nachbarn gelten nur dann als
  benachbart, wenn ein Anschluss-Koordinatensystem K1-K4 des einen naeher als
  omniflo_ks_toleranz_mm (neu, Default 10mm) an einem K-Punkt des anderen
  liegt. Das verwirft bloss ueberlappende BBoxen ohne echten Anschluss.
  Elemente ohne K1-K4 fallen auf reine BBox-Ueberschneidung zurueck.

Geraden fuehren keine echten K-Bloecke in der Zeichnung; ihre K1 (Anfang) /
K2 (Ende) werden beim Export aus Einfuegepunkt + Laenge + Rotation
synthetisiert (csv:gerade-k-kos-strings in export.lsp) und stehen damit in den
CSV-Spalten K1/K2 - wie echte KOS - und nehmen an der Kollisionspruefung teil.

Verifiziert an results/test-omnicollision: gerichtete Nachbarschaften 20 -> 14.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-31 16:56:10 +02:00

750 lines
31 KiB
Python
Raw Blame History

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
export_csv.py - Erzeugt eine einfache Item-Liste als CSV.
Fuer Omniflo-Elemente (Bogen/Weiche/Gerade) wird am Ende zusaetzlich eine
"Omniflo Sum"-Zeile mit Anzahl Boegen/Weichentypen und Gesamtlaenge AP110/AP60
angehaengt (siehe build_omni_sum_merkmale).
Aufruf:
python export_csv.py <export_raw.json> <data_dir> <output.csv>
CSV-Format:
Elementnummer;TeileArt;TeileId;Bezeichnung;Planquadrat;Anzahl;Position;Boundingbox;
Insertpoint;K1;K2;K3;K4;Nachbarn;Fehler;Merkmale
Position und Boundingbox stammen aus der von export.lsp (csv:get-bbox, per
vla-getboundingbox) ermittelten Bounding-Box je Block:
Position = Mittenkoordinate x, y, z
Boundingbox = Ausdehnung (Laenge, Breite, Hoehe) in x, y, z
Insertpoint = KOS-String (24-Zeichen Base64, Position+Rotation als Quaternion,
siehe csv:kos-encode in export.lsp) des Block-Einfuegepunkts selbst.
K1-K4 = Positions-Strings der im Block enthaltenen Koordinatensysteme K1..K4
(12-Zeichen Base64, nur Position ohne Rotation, siehe csv:trans-encode;
nur bei Omniflo Bogen/Weiche/Gerade vorhanden, sonst leer).
Planquadrat wird aus der x/y-Koordinate des Blocks berechnet, siehe
export_planquadrat.py ([Planquadrate] in cfg/export.cfg).
Nachbarn = kommaseparierte TeileId-Liste ueberschneidender Elemente (Bounding-
Box-Ueberschneidung in der x/y-Ebene, reines Python ohne externe Abhaengigkeit,
siehe export_neighbors.py). Geprueft wird NICHT global ueber alle Elemente,
sondern in drei getrennten Gruppen: 1) Kreisel/Eckrad gegen Kreisel/Eckrad,
2) Kreisel/Eckrad gegen Gefaellestrecke/Foerderer/Strecke-Modul (diese drei
nicht gegeneinander), 3) Omniflo-Elemente gegen Omniflo-Elemente. Fuer die
Omniflo-Gruppe liefert eine shapely-STRtree-Broad-Phase zunaechst alle Paare
mit ueberschneidenden Bounding-Boxen; anschliessend gelten sie nur dann als
benachbart, wenn sich ihre Anschluss-Koordinatensysteme K1-K4 tatsaechlich
beruehren (ein K-Punkt naeher als omniflo_ks_toleranz_mm, Default 10mm) -
das trennt bloss ueberlappende BBoxen von echten Anschluessen (Geraden fuehren
dafuer beim Export synthetisierte K1/K2, siehe csv:gerade-k-kos-strings).
Toleranzen und (Fallback-)Rastergroesse konfigurierbar ueber [Nachbarschaft]
in cfg/export.cfg.
Fehler = "unverbunden" (keine Partner, aber mindestens einer noetig),
"nur ein Partner" (ein Partner, aber mindestens zwei noetig) oder leer
(Mindestanzahl erreicht bzw. TeileArt wird nicht geprueft). Kreisel/Eckrad/
Omniflo-Elemente brauchen mindestens einen Partner, Gefaellestrecke/
Foerderer/Strecke-Modul mindestens zwei (siehe export_neighbors.py
MIN_PARTNER). Ein Kreisel wird dabei NICHT als eine, sondern als zwei
BBoxen (links/rechts, geteilt entlang der Achse durch Antriebs-/
Spannstation, siehe export_neighbors.kreisel_half_bboxes) gegen
Gefaellestrecke/Foerderer getestet - deren "Nachbarn"-Spalte zeigt darum
"<KreiselTeileId>-L" bzw. "-R" statt nur der Kreisel-TeileId.
Separator/Scanner-Zeilen bekommen zusaetzlich ein berechnetes "Zuordnung"-
Merkmal (Gefaellestrecke > Foerderer > Kreiselhaelfte, per BBox-Ueber-
schneidung - siehe export_neighbors.compute_sensor_zuordnung), das die
ATTRIB-Zuordnung (falls das Attribut ZUORDNUNG in der Zeichnung von Hand
gesetzt wurde) ersetzt, sofern ein Treffer gefunden wird. Scanner-Zeilen
bekommen zusaetzlich ein "Naechster Separator"-Merkmal (raeumlich naechst-
gelegener Separator, euklidischer Abstand der BBox-Mittelpunkte - eigene,
von der Zuordnung unabhaengige Suche, siehe export_neighbors.compute_
scanner_nearest_separator).
Position, Boundingbox, Planquadrat, Nachbarn und Fehler sind nur fuer
EXPORTCSV vorhanden (nicht EXPORTSIVAS) - siehe csv:run-export. Insertpoint/
K1-K4 stehen zwar in export_raw.json fuer jeden Export zur Verfuegung (siehe
csv:block-to-json), werden aber ebenfalls nur von diesem Skript (EXPORTCSV)
als CSV-Spalten ausgegeben.
"""
import json
import sys
import os
from export_blockpatterns import load_patterns, matches_any, csv_quote, safe_float
from export_planquadrat import load_planquadrat_config, resolve_origins, compute_planquadrat
from export_neighbors import (
load_neighbor_tolerance_mm,
load_omniflo_cell_size_mm,
load_omniflo_ks_tolerance_mm,
load_kreisel_durchmesser_mm,
load_collision_debug_target,
compute_neighbor_ids,
compute_neighbor_errors,
compute_sensor_zuordnung,
compute_scanner_nearest_separator,
SENSOR_TEILEARTEN,
)
BLOCKPATTERNS = load_patterns()
PLANQUADRAT_CFG = load_planquadrat_config()
NEIGHBOR_TOLERANCE_MM = load_neighbor_tolerance_mm()
NEIGHBOR_OMNIFLO_CELL_SIZE_MM = load_omniflo_cell_size_mm()
NEIGHBOR_OMNIFLO_KS_TOLERANCE_MM = load_omniflo_ks_tolerance_mm()
KREISEL_DURCHMESSER_MM = load_kreisel_durchmesser_mm()
NEIGHBOR_DEBUG_TARGET = load_collision_debug_target()
# ---------------------------------------------------------------------------
# Gemeinsame Hilfsfunktionen (auch in export_sivas.py verwendet)
# ---------------------------------------------------------------------------
def load_json(path):
"""Laedt JSON, tolerant gegenueber BricsCAD-Ausgaben.
AutoLISP schreibt export_raw.json (write-line) im ANSI-Codepage des
Systems, nicht in UTF-8 - Sonderzeichen (z.B. Gradzeichen) koennen
daher nicht als UTF-8 dekodierbar sein. Fallback auf cp1252.
"""
try:
with open(path, "r", encoding="utf-8") as f:
return json.load(f)
except UnicodeDecodeError:
with open(path, "r", encoding="cp1252") as f:
return json.load(f)
def build_lookup(boegen, weichen):
"""Erzeugt ein Dict: SivasNr (als String) -> (typ, eintrag)"""
lookup = {}
for b in boegen:
lookup[str(b["Sivasnr"])] = ("bogen", b)
for w in weichen:
lookup[str(w["Sivasnr"])] = ("weiche", w)
return lookup
def format_sivasnr(sivasnr):
"""
Wandelt einen "+"-separierten Sivasnr-Wert (Rohformat aus den JSON-Katalogen,
dort weiterhin benoetigt fuer Dateinamen-Referenzen) in eine reine
kommaseparierte Liste fuer die Ausgabe (CSV/Block-Attribut) um.
"""
return ", ".join(p.strip() for p in str(sivasnr).split("+") if p.strip())
# ---------------------------------------------------------------------------
# Merkmale-Builder
# ---------------------------------------------------------------------------
def get_hoehe(block):
"""Liest HOEHE aus Block-Attributen. Fallback: Z-Koordinate."""
attribs = block.get("attribs", {})
return attribs.get("HOEHE") or str(int(block.get("z", 0) or 0)) or "2000"
def get_drehung(block):
"""Liest DREHUNG aus Block-Attributen. Fallback: CAD-Rotation."""
attribs = block.get("attribs", {})
drehung_raw = attribs.get("DREHUNG")
if drehung_raw is None:
return block.get("rotation", 0.0)
return safe_float(drehung_raw, block.get("rotation", 0.0))
def build_bogen_merkmale(block, eintrag):
return {
"Kurvenwinkel": float(eintrag.get("KurvenWinkel", 0)),
"Radius": float(eintrag.get("Radius", 0)),
"Höhe": get_hoehe(block),
"Drehung": get_drehung(block),
"SivasNummer": format_sivasnr(eintrag.get("Sivasnr", ""))
}
def build_weiche_merkmale(block, eintrag):
wt = eintrag.get("WeichenTyp", "Einzelweiche")
return {
"Weichentyp": wt,
"Richtung": str(eintrag.get("KurvenRichtung", "")),
"Weichenwinkel": float(eintrag.get("KurvenWinkel", 0)),
"Höhe": get_hoehe(block),
"Drehung": get_drehung(block),
"Antrieb Kurve": eintrag.get("SivasnrTEF") is not None,
"SivasNummer": format_sivasnr(eintrag.get("Sivasnr", ""))
}
def build_variofoerderer_merkmale(block):
"""Merkmale-Dict fuer einen VF_N- oder GF_N-Streckenblock.
Die Attributnamen folgen dem gemeinsamen Schema *strecke-attr-front* /
*strecke-attr-gruppe* in ssg_core.lsp. Hoehen und Delta-Werte tragen dort
das Einheiten-Suffix _mm; SEITE ist in SEITE_AS/SEITE_ES (Antriebs-/
Endstation) und WINKEL in GF_WINKEL/VF_WINKEL aufgeteilt. Die alten Namen
ohne Suffix werden per Fallback weiter gelesen, damit aeltere
export_raw.json-Dateien (z.B. tests/reference/export_raw.json) und noch
nicht neu aufgebaute Bloecke in der Zeichnung weiterhin Werte liefern.
"""
attribs = block.get("attribs", {})
def att(*keys):
"""Erster nicht-leerer Attributwert; aktuelles Schema zuerst, Alt-Name
als Fallback."""
for key in keys:
value = attribs.get(key)
if value not in (None, ""):
return value
return ""
return {
"Typ": att("TYP"),
"Seite_AS": att("SEITE_AS", "SEITE"),
"Seite_ES": att("SEITE_ES", "SEITE"),
"Winkel": att("GF_WINKEL", "WINKEL"),
"VF_Winkel": att("VF_WINKEL"),
"DeltaH_mm": att("DELTA_H_mm", "DELTA_H"),
"DeltaL_mm": att("DELTA_L_mm", "DELTA_L"),
"L_VF_m": att("L_VF_m"),
"L_GF_m": att("L_GF_m"),
"Montagehoehe_m": att("MONTAGEHOEHE_m"),
"Hoehe_Von_mm": att("HOEHE_VON_mm", "HOEHE_VON"),
"Hoehe_Bis_mm": att("HOEHE_BIS_mm", "HOEHE_BIS"),
"Antriebfahrtrichtung": att("ANTRIEBFAHRTRICHTUNG"),
"Anzahl_Separator": att("ANZAHL_SEPARATOR"),
"Anzahl_Scanner": att("ANZAHL_SCANNER"),
}
def get_laenge_mm(block):
"""Liest LAENGE (oder A) aus Block-Attributen. Fallback: 2000mm."""
attribs = block.get("attribs", {})
laenge = attribs.get("LAENGE") or attribs.get("A")
return safe_float(laenge, 2000.0)
def build_gerade_merkmale(block):
laenge_mm = get_laenge_mm(block)
hoehe = get_hoehe(block)
return {
"Anzahl der Separatoren": "0",
"Länge in Meter": f"{laenge_mm / 1000.0:.2f}",
"Winkel": "0",
"Anzahl der Scanner": 0,
"Höhe oben": hoehe,
"Höhe unten": hoehe,
"Drehung": get_drehung(block),
"SivasNummer": ""
}
def build_sensor_merkmale(block):
"""Merkmale-Dict fuer einen eigenstaendigen Separator_SP/S-LP- oder
Scanner-Block (siehe pattern_separator/pattern_scanner in cfg/export.cfg).
ZUORDNUNG/ARTINR/LAYER (+ neuerdings ID) sind die ATTDEFs (siehe
ils-insert-sensor/mubea:build-separator-one, ssg-id-collect-blocks in
ssg_id.lsp)."""
attribs = block.get("attribs", {})
return {
"Zuordnung": attribs.get("ZUORDNUNG", ""),
"ARTINR": attribs.get("ARTINR", ""),
"Höhe": get_hoehe(block),
"Drehung": get_drehung(block),
}
def format_sensor_id(prefix, n):
"""Fallback-TeileId fuer Separator/Scanner-Zeilen im CSV-Export, falls das
Block-Attribut "ID" (noch) leer ist.
Separator_SP/Scanner haben seit kurzem ein eigenes ID-ATTDEF (siehe
ssg-id-collect-blocks in ssg_id.lsp) - normalerweise wird also das echte,
ueber ssg-id-generate/ssg-id-check-all vergebene "ID"-Attribut als
TeileId verwendet, genau wie bei Kreisel/GF/VF. Nur wenn das (noch)
fehlt (z.B. bei aelteren, noch nicht neu exportierten Bloecken), erzeugt
diese Funktion eine rein fuer DIESEN Exportlauf gueltige, NICHT ueber
mehrere Exportlaeufe hinweg stabile Ersatznummer - der Praefix macht
sichtbar, dass es keine echte, persistierte Block-ID ist.
"""
return f"{prefix}-{n:04d}"
# ---------------------------------------------------------------------------
# Omniflo Sum-Zeile (Anzahl Boegen/Weichentypen, Gesamtlaenge AP110/AP60)
# ---------------------------------------------------------------------------
WEICHE_SUBTYP_MAP = {
"Einzelweiche": "einzelweiche",
"Doppelweiche": "doppelweiche",
"Dreiwegeweiche": "doppelweiche",
"Deltaweiche": "deltaweiche",
"Dreifachweiche": "deltaweiche",
"Sternweiche": "sternweiche",
}
def weichensubtyp(block, eintrag):
"""Bestimmt den Weichen-Subtyp fuer die Omniflo-Sum-Zaehlung."""
beschr = (block.get("attribs", {}).get("BESCHR") or "").upper()
if "WEICHENK" in beschr:
return "weichenkoerper"
subtyp = WEICHE_SUBTYP_MAP.get(eintrag.get("WeichenTyp", ""))
if subtyp:
return subtyp
if "DELTA" in beschr:
return "deltaweiche"
if "STAR" in beschr:
return "sternweiche"
return "einzelweiche"
def build_omni_sum_merkmale(cnt_boegen, cnt_wk, cnt_einzel, cnt_delta, cnt_doppel, cnt_stern,
len_ap110_mm, len_ap60_mm):
return {
"Anzahl Bögen": cnt_boegen,
"Anzahl Weichengrundkörper": cnt_wk,
"Anzahl Einwegweichen": cnt_einzel,
"Anzahl Deltaweichen": cnt_delta,
"Anzahl Doppelweichen und Dreiwegeweichen": cnt_doppel,
"Anzahl Sternweichen": cnt_stern,
"Gesamtlänge AP110": round(len_ap110_mm, 1),
"Gesamtlänge AP60": round(len_ap60_mm, 1),
}
def build_kreisel_merkmale(block):
attribs = block.get("attribs", {})
abstand_m = str(round(safe_float(attribs.get("ABSTAND", "2300"), 2300.0) / 1000.0, 2))
hoehe_m = str(round(safe_float(attribs.get("HOEHE", "0"), 0.0) / 1000.0, 2))
return {
"Abstand (Kreiselachse A - Kreiselachse) in Meter": abstand_m,
"Anzahl der Separatoren": attribs.get("ANZAHL_SEPARATOR", attribs.get("N_SEPARATOREN", "2")),
"Kreiselart": attribs.get("KREISELART", "STANDARD"),
"Anzahl der Scanner": attribs.get("ANZAHL_SCANNER", attribs.get("N_SCANNER", "0")),
"Anzahl der Rampen": attribs.get("ANZAHL_RAMPEN", attribs.get("N_RAMPEN", "0")),
"Höhe in m": hoehe_m,
"Drehrichtung": attribs.get("DREHRICHTUNG", "UZS"),
"Drehung": block.get("rotation", 0.0),
"Name": attribs.get("Bezeichnung", ""),
"Geruest fuer Einzelmodul": attribs.get("GERUEST_EINZELMODUL", "0") == "1",
}
# ---------------------------------------------------------------------------
# Bounding-Box-Spalten (Position/Boundingbox)
# ---------------------------------------------------------------------------
def bbox_columns(block):
"""Liest die von export.lsp (csv:get-bbox) ermittelte Bounding-Box, sowie
die KOS-Strings von Insertpoint und K1-K4 (csv:kos-encode/csv:block-to-json).
Rueckgabe: dict mit "position" (Mittenkoordinate x,y,z) und "boundingbox"
(Ausdehnung x,y,z) als Strings, oder leere Strings falls kein bbox-Feld
vorhanden ist (z.B. EXPORTSIVAS oder die synthetische Omniflo-Sum-Zeile).
"insertpoint" ist ein 24-Zeichen Base64-KOS-String (Position + Rotation
als Quaternion, csv:kos-encode); "k1".."k4" sind 12-Zeichen Positions-
Strings (nur Position, csv:trans-encode) - leer falls das jeweilige
Koordinatensystem am Block nicht existiert.
"_bbox" enthaelt zusaetzlich die rohen Zahlenwerte fuer die Nachbarschafts-
erkennung (siehe export_neighbors.py), wird nicht in die CSV geschrieben.
"""
kos_cols = {
"insertpoint": block.get("insertpoint", ""),
"k1": block.get("k1", ""),
"k2": block.get("k2", ""),
"k3": block.get("k3", ""),
"k4": block.get("k4", ""),
}
bbox = block.get("bbox")
if not bbox:
return {"position": "", "boundingbox": "", "_bbox": None, **kos_cols}
return {
"position": f'{bbox.get("cx", 0):.2f}, {bbox.get("cy", 0):.2f}, {bbox.get("cz", 0):.2f}',
"boundingbox": f'{bbox.get("dx", 0):.2f}, {bbox.get("dy", 0):.2f}, {bbox.get("dz", 0):.2f}',
"_bbox": bbox,
**kos_cols,
}
def planquadrat_column(block, planquadrat_cfg):
"""Berechnet das Planquadrat aus x/y des Blocks (siehe export_planquadrat.py).
planquadrat_cfg = Rueckgabe von resolve_origins() fuer den aktuellen
Exportlauf (x_origin_mm/y_origin_mm bereits aufgeloest).
Leerer String, wenn keine Planquadrat-Konfiguration geladen werden konnte
(fehlende/unvollstaendige [Planquadrate]-Sektion in cfg/export.cfg).
"""
if planquadrat_cfg is None:
return ""
x = block.get("x")
y = block.get("y")
if x is None or y is None:
return ""
return compute_planquadrat(float(x), float(y), planquadrat_cfg)
# ---------------------------------------------------------------------------
# Bekannte Blocknamen
# ---------------------------------------------------------------------------
SKIP_BLOCKS = set(BLOCKPATTERNS.get("pattern_ks_subblocks", ["K1", "K2", "K3", "K4", "KS_EIN", "KS_AUS"]))
# ---------------------------------------------------------------------------
# Bloecke verarbeiten (einfache Liste, keine Summierung)
# ---------------------------------------------------------------------------
def process_blocks(blocks, lookup, dbg=None):
# x_origin_mm/y_origin_mm einmalig fuer diesen Exportlauf aufloesen (siehe
# export_planquadrat.resolve_origins): automatisch aus den Bloecken, falls
# nicht explizit in [Planquadrate] konfiguriert.
planquadrat_cfg = resolve_origins(PLANQUADRAT_CFG, blocks)
items = []
elem_nr = 0
bogen_count = 0
weiche_count = {}
gerade_count = 0
kreisel_count = 0
eckrad_count = 0
vf_count = 0
gf_count = 0
strecke_modul_count = 0
separator_count = 0
scanner_count = 0
# Zaehlung fuer die Omniflo-Sum-Zeile (nur Bogen/Weiche/Gerade)
cnt_wk = cnt_einzel = cnt_delta = cnt_doppel = cnt_stern = 0
len_ap110_mm = 0.0
len_ap60_mm = 0.0
has_omniflo = False
for block in blocks:
bname = block.get("block_name", "")
if bname in SKIP_BLOCKS:
continue
# Omniflo Bogen oder Weiche
if bname in lookup:
typ, eintrag = lookup[bname]
elem_nr += 1
shape_id = block.get("attribs", {}).get("ID", "0000")
has_omniflo = True
if typ == "bogen":
bogen_count += 1
items.append({
"nr": elem_nr,
"teileart": "Omniflo Kurve",
"teileid": shape_id,
"bezeichnung": f"OFBogen :{bogen_count}",
"planquadrat": planquadrat_column(block, planquadrat_cfg),
"merkmale": build_bogen_merkmale(block, eintrag),
**bbox_columns(block),
})
elif typ == "weiche":
wt = eintrag.get("WeichenTyp", "Einzelweiche")
weiche_count[wt] = weiche_count.get(wt, 0) + 1
items.append({
"nr": elem_nr,
"teileart": "Omniflo Weiche",
"teileid": shape_id,
"bezeichnung": f"OFWeiche :{weiche_count[wt]}",
"planquadrat": planquadrat_column(block, planquadrat_cfg),
"merkmale": build_weiche_merkmale(block, eintrag),
**bbox_columns(block),
})
subtyp = weichensubtyp(block, eintrag)
if subtyp == "weichenkoerper":
cnt_wk += 1
elif subtyp == "deltaweiche":
cnt_delta += 1
elif subtyp == "sternweiche":
cnt_stern += 1
elif subtyp == "doppelweiche":
cnt_doppel += 1
else:
cnt_einzel += 1
continue
# Omniflo Gerade (AP110)
if matches_any(bname, BLOCKPATTERNS.get("pattern_gerade", [])):
elem_nr += 1
gerade_count += 1
has_omniflo = True
items.append({
"nr": elem_nr,
"teileart": "Omniflo Gerade",
"teileid": block.get("attribs", {}).get("ID", "0000"),
"bezeichnung": f"OFGerade :{gerade_count}",
"planquadrat": planquadrat_column(block, planquadrat_cfg),
"merkmale": build_gerade_merkmale(block),
**bbox_columns(block),
})
len_ap110_mm += get_laenge_mm(block)
continue
# VarioFoerderer Compound-Block (VF_N)
if matches_any(bname, BLOCKPATTERNS.get("pattern_variofoerderer", [])):
elem_nr += 1
vf_count += 1
items.append({
"nr": elem_nr,
"teileart": "ILS 2.0 Strecke",
"teileid": block.get("attribs", {}).get("ID", "0000"),
"bezeichnung": f"VarioFoerderer :{vf_count}",
"planquadrat": planquadrat_column(block, planquadrat_cfg),
"merkmale": build_variofoerderer_merkmale(block),
**bbox_columns(block),
})
continue
# Gefaellestrecke Compound-Block (GF_N)
if matches_any(bname, BLOCKPATTERNS.get("pattern_gefaellestrecke", [])):
elem_nr += 1
gf_count += 1
items.append({
"nr": elem_nr,
"teileart": "ILS 2.0 Gefaellestrecke",
"teileid": block.get("attribs", {}).get("ID", "0000"),
"bezeichnung": f"Gefaellestrecke :{gf_count}",
"planquadrat": planquadrat_column(block, planquadrat_cfg),
"merkmale": build_variofoerderer_merkmale(block),
**bbox_columns(block),
})
continue
# ILS Eckrad (vor Kreisel pruefen: pattern_kreisel schliesst
# ECKRAD_* mit ein, pattern_eckrad ist die praezisere Teilmenge)
if matches_any(bname, BLOCKPATTERNS.get("pattern_eckrad", [])):
elem_nr += 1
eckrad_count += 1
items.append({
"nr": elem_nr,
"teileart": "ILS 2.0 Eckrad",
"teileid": block.get("attribs", {}).get("ID", "0000"),
"bezeichnung": f"Eckrad :{eckrad_count}",
"planquadrat": planquadrat_column(block, planquadrat_cfg),
"merkmale": build_kreisel_merkmale(block),
**bbox_columns(block),
})
continue
# ILS Kreisel (Compound-Bloecke mit Praefix KR_ oder KREISEL_)
if matches_any(bname, BLOCKPATTERNS.get("pattern_kreisel", [])):
elem_nr += 1
kreisel_count += 1
items.append({
"nr": elem_nr,
"teileart": "ILS 2.0 Kreisel",
"teileid": block.get("attribs", {}).get("ID", "0000"),
"bezeichnung": f"Kreisel :{kreisel_count}",
"planquadrat": planquadrat_column(block, planquadrat_cfg),
"merkmale": build_kreisel_merkmale(block),
# x/y/rotation/attribs (roh, nicht nur die Merkmale) werden
# von export_neighbors.kreisel_half_bboxes gebraucht, um die
# Links/Rechts-Trennung ueber AN8/SP8/ABSTAND nachzurechnen -
# bbox_columns() liefert nur die fertige Gesamt-BBox, keine
# Rohdaten dafuer.
"x": block.get("x", 0.0),
"y": block.get("y", 0.0),
"rotation": block.get("rotation", 0.0),
"attribs": block.get("attribs", {}),
**bbox_columns(block),
})
continue
# ILS Strecke-Module (Vario/Staustrecke-Einzelkomponenten)
if matches_any(bname, BLOCKPATTERNS.get("pattern_strecke_module", [])):
elem_nr += 1
strecke_modul_count += 1
items.append({
"nr": elem_nr,
"teileart": "ILS 2.0 Strecke - Modul",
"teileid": block.get("attribs", {}).get("ID", "0000"),
"bezeichnung": f"Streckenmodul :{strecke_modul_count}",
"planquadrat": planquadrat_column(block, planquadrat_cfg),
"merkmale": {"Blockname": bname},
**bbox_columns(block),
})
continue
# Separator (Separator_SP/S-LP) - eigene TeileArt-Zeile. TeileId: das
# echte ID-Attribut, falls vorhanden (Separator_SP/Scanner haben seit
# kurzem ein eigenes ID-ATTDEF, siehe ssg-id-collect-blocks in
# ssg_id.lsp - ssg-id-check-all fuellt es vor jedem Export). Fallback
# format_sensor_id (laufende SEP-/SCN-Nummer NUR fuer diesen Export-
# lauf) fuer aeltere Bloecke, deren ID-Attribut noch leer ist (z.B.
# vor dem naechsten Insert/ssg-id-check-all-Durchlauf).
# EXPORTSIVAS schliesst denselben Blockpattern bewusst aus (siehe
# export_sivas.py) - dort zaehlt bereits ANZAHL_SEPARATOR am
# zugehoerigen Kreisel/GF/VF.
if matches_any(bname, BLOCKPATTERNS.get("pattern_separator", [])):
elem_nr += 1
separator_count += 1
items.append({
"nr": elem_nr,
"teileart": "ILS 2.0 Separator",
"teileid": block.get("attribs", {}).get("ID") or format_sensor_id("SEP", separator_count),
"bezeichnung": f"Separator :{separator_count}",
"planquadrat": planquadrat_column(block, planquadrat_cfg),
"merkmale": build_sensor_merkmale(block),
**bbox_columns(block),
})
continue
# Scanner - eigene TeileArt-Zeile, sonst analog zu Separator oben.
if matches_any(bname, BLOCKPATTERNS.get("pattern_scanner", [])):
elem_nr += 1
scanner_count += 1
items.append({
"nr": elem_nr,
"teileart": "ILS 2.0 Scanner",
"teileid": block.get("attribs", {}).get("ID") or format_sensor_id("SCN", scanner_count),
"bezeichnung": f"Scanner :{scanner_count}",
"planquadrat": planquadrat_column(block, planquadrat_cfg),
"merkmale": build_sensor_merkmale(block),
**bbox_columns(block),
})
continue
if has_omniflo:
elem_nr += 1
items.append({
"nr": elem_nr,
"teileart": "Omniflo Sum",
"teileid": "autogenerated_of_json",
"bezeichnung": "Omniflo sum",
"planquadrat": "",
"merkmale": build_omni_sum_merkmale(
bogen_count, cnt_wk, cnt_einzel, cnt_delta, cnt_doppel, cnt_stern,
len_ap110_mm, len_ap60_mm),
"position": "",
"boundingbox": "",
"insertpoint": "",
"k1": "",
"k2": "",
"k3": "",
"k4": "",
})
neighbor_ids = compute_neighbor_ids(
items, NEIGHBOR_TOLERANCE_MM, NEIGHBOR_OMNIFLO_CELL_SIZE_MM,
kreisel_durchmesser_mm=KREISEL_DURCHMESSER_MM,
omniflo_ks_tolerance_mm=NEIGHBOR_OMNIFLO_KS_TOLERANCE_MM, dbg=dbg)
neighbor_errors = compute_neighbor_errors(items, neighbor_ids)
for item, nachbarn, fehler in zip(items, neighbor_ids, neighbor_errors):
item["nachbarn"] = nachbarn
item["fehler"] = fehler
# Separator/Scanner: Zuordnung zu Gefaellestrecke/Foerderer/Kreiselhaelfte
# per BBox-Ueberschneidung berechnen (Prioritaet siehe compute_sensor_
# zuordnung) - ersetzt die von Hand gesetzte ZUORDNUNG-Attribut-Merkmale
# nur, wenn tatsaechlich ein Treffer gefunden wurde.
sensor_zuordnung = compute_sensor_zuordnung(
items, NEIGHBOR_TOLERANCE_MM, kreisel_durchmesser_mm=KREISEL_DURCHMESSER_MM, dbg=dbg)
for item, zuordnung in zip(items, sensor_zuordnung):
if item.get("teileart", "") in SENSOR_TEILEARTEN and zuordnung:
item["merkmale"]["Zuordnung"] = zuordnung
# NACH der Separator-Zuordnung: zu jedem Scanner den raeumlich naechst-
# gelegenen Separator suchen (eigene, unabhaengige Suche - siehe
# compute_scanner_nearest_separator).
nearest_separator = compute_scanner_nearest_separator(items, dbg=dbg)
for item, separator_id in zip(items, nearest_separator):
if item.get("teileart", "") == "ILS 2.0 Scanner" and separator_id:
item["merkmale"]["Naechster Separator"] = separator_id
return items
# ---------------------------------------------------------------------------
# CSV-Formatierung
# ---------------------------------------------------------------------------
def format_csv_line(item):
merkmale_json = json.dumps(item["merkmale"], ensure_ascii=False)
return (
f'{item["nr"]}'
f';"{csv_quote(item["teileart"])}"'
f';"{csv_quote(item["teileid"])}"'
f';"{csv_quote(item["bezeichnung"])}"'
f';"{csv_quote(item["planquadrat"])}"'
f';1'
f';"{csv_quote(item["position"])}"'
f';"{csv_quote(item["boundingbox"])}"'
f';"{csv_quote(item["insertpoint"])}"'
f';"{csv_quote(item["k1"])}"'
f';"{csv_quote(item["k2"])}"'
f';"{csv_quote(item["k3"])}"'
f';"{csv_quote(item["k4"])}"'
f';"{csv_quote(item["nachbarn"])}"'
f';"{csv_quote(item["fehler"])}"'
f';{merkmale_json}'
)
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
def main():
if len(sys.argv) < 4:
print("Aufruf: python export_csv.py <export_raw.json> <data_dir> <output.csv>")
sys.exit(1)
raw_json_path = sys.argv[1]
data_dir = sys.argv[2]
output_csv = sys.argv[3]
blocks = load_json(raw_json_path)
boegen_path = os.path.join(data_dir, "json", "omniflo_boegen.json")
weichen_path = os.path.join(data_dir, "json", "omniflo_weichen.json")
boegen = load_json(boegen_path) if os.path.exists(boegen_path) else []
weichen = load_json(weichen_path) if os.path.exists(weichen_path) else []
lookup = build_lookup(boegen, weichen)
print(f"[export_csv] {len(blocks)} Bloecke geladen, "
f"{len(boegen)} Boegen, {len(weichen)} Weichen im Katalog.")
# Optionales Kollisions-/Nachbarschafts-Debuglog (siehe [Nachbarschaft] ->
# debug_log in cfg/export.cfg). dbg ist ein Callable(str), das eine Zeile
# in die .dbg-Datei schreibt - oder None, wenn das Log deaktiviert ist.
dbg_file = None
dbg = None
if NEIGHBOR_DEBUG_TARGET:
try:
os.makedirs(os.path.dirname(NEIGHBOR_DEBUG_TARGET), exist_ok=True)
dbg_file = open(NEIGHBOR_DEBUG_TARGET, "w", encoding="utf-8")
dbg = lambda msg: dbg_file.write(msg + "\n")
dbg(f"# Kollisions-/Nachbarschafts-Debuglog (export_csv.py)")
dbg(f"# Quelle: {raw_json_path}")
dbg(f"# Bloecke im Roh-JSON: {len(blocks)}")
dbg("")
except OSError as exc:
print(f"[export_csv] WARNUNG: Debuglog nicht schreibbar "
f"({NEIGHBOR_DEBUG_TARGET}): {exc}")
dbg_file = None
dbg = None
items = process_blocks(blocks, lookup, dbg=dbg)
if dbg_file:
dbg_file.close()
print(f"[export_csv] Kollisions-Debuglog geschrieben: {NEIGHBOR_DEBUG_TARGET}")
header = ("Elementnummer;TeileArt;TeileId;Bezeichnung;Planquadrat;Anzahl;Position;Boundingbox;"
"Insertpoint;K1;K2;K3;K4;Nachbarn;Fehler;Merkmale")
with open(output_csv, "w", encoding="utf-8") as f:
f.write(header + "\n")
for item in items:
f.write(format_csv_line(item) + "\n")
print(f"[export_csv] CSV geschrieben: {output_csv} ({len(items)} Zeilen)")
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink