mistangl/dxfmakros
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
f751a900766f0a02d4fa6cf8d594437e466663be
dxfmakros/lib/export_csv.py
T
Michael StanglandClaude Sonnet 5 f751a90076 export_csv: VF_n/GF_n-Einzelelemente (Boegen, Streckenelemente, Umlenk-/Motorstation) als Proxy-Zeilen exportieren 
Bisher tauchten nur der Wrapper selbst, verpackte Separatoren und AS/ES-
Kettenenden als eigene CSV-Zeilen auf - die uebrigen Kettenbausteine
(Boegen, gerade Foerderer-/Gefaellestrecken-Segmente, Umlenk-/Motorstation)
blieben nur in der Wrapper-Blockdefinition verpackt. csv:element-proxies-
erzeugen (analog zu csv:sep-proxies-erzeugen) fuegt fuer jeden Fund eine
temporaere Kopie ein, K1/K2 kommen aus dem EIGENEN KS_EIN/KS_AUS jedes
Elements (nicht dem Gesamt-AS/ES der Kette) und bleiben leer, wo keines
existiert (z.B. das nie tatsaechlich eingefuegte Vario_Kurve_*).

Zusaetzlicher Fix: csv:get-k-kos-strings ignorierte bisher die Skalierung
des Top-Level-Inserts - fuer Staustrecke_SP_1000_mm (das per Laenge nicht-
uniform skaliert wird) waere K2 damit falsch gewesen, unabhaengig von der
tatsaechlich gebauten Segmentlaenge.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-09-24 13:08:35 +02:00

1642 lines
74 KiB
Python
Raw Blame History

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
export_csv.py - Erzeugt eine einfache Item-Liste als CSV.
Aufruf:
python export_csv.py <export_raw.json> <data_dir> <output.csv>
CSV-Format (Header):
Elementnummer;TeileArt;TeileId;Bezeichnung;Planquadrat;Anzahl;Position;Boundingbox;
Insertpoint;K1;K2;K3;K4;Nachbarn;TrackIds;Fehler;Warnungen;Merkmale
Beispielzeile (Separator, TeileId 0011 an Kreisel 0010):
2;"ILS 2.0 Separator";"0011";"Separator :1";"B/3";1;
"10800.0, 5000.0, 0.0";"210.0, 150.0, 14.0";"...";"";"";"";"";
"0010, 0012";"";"";"";{"Zuordnung":"0010"}
Fuer die vollstaendige Herleitung jeder Spalte (Nachbarn/TrackIds/sepliste/
Fehler/Warnungen, inkl. eines durchgerechneten Beispiels Kreisel+VarioFoerderer
von export_raw.json bis zur fertigen CSV-Zeile) siehe doc/csv-export.md.
"""
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_element_box_mm,
load_separator_box_mm,
separator_bbox,
load_collision_debug_target,
compute_neighbor_ids,
compute_neighbor_errors,
compute_sensor_zuordnung,
compute_scanner_nearest_separator,
compute_kreisel_touch_switches,
compute_sep_kette,
compute_kreisel_umlauf,
SENSOR_TEILEARTEN,
WEICHE_TEILEART,
GF_VF_TEILEARTEN,
KREISEL_SPLIT_TEILEART,
KREISEL_UMLAUF_TEILEARTEN,
AUSSCHLEUS_TEILEART,
EINSCHLEUS_TEILEART,
)
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()
ELEMENT_BOX_MM = load_element_box_mm()
SEPARATOR_BOX_MM = load_separator_box_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 bezeichnung_oder_fallback(block, fallback):
"""CSV-Spalte "Bezeichnung" fuer VF_N/GF_N/Kreisel/Eckrad: das vom
Anwender in BricsCAD frei editierbare Attribut "Bezeichnung" (ATTDEF,
siehe vf-bname/gf-bname bzw. KreiselInsert.lsp "name"-Dialogfeld), falls
vorhanden und nicht leer - sonst der generierte Standardname
("VarioFoerderer :3", "Kreisel :2" usw.).
Das Attribut wird beim Bau zunaechst mit dem Blocknamen (VF_n/GF_n) bzw.
"Kreisel<n>" vorbelegt, kann danach aber umbenannt werden - damit laesst
sich ein Element in der Zeichnung und im CSV-Export leichter
wiedererkennen als ueber die reine Lauf-Nummerierung."""
bez = (block.get("attribs", {}) or {}).get("Bezeichnung", "").strip()
return bez if bez else fallback
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.
Die CSV-Spalte "TrackIds" (item["trackids"], Reihenfolge der Separator-
TeileIds entlang der Strecke) steht hier NICHT drin - sie kommt erst nach
der sepliste-Aufloesung dazu (siehe write_separatorliste_merkmale in
process_blocks), weil an dieser Stelle die x/y-Koordinaten aus der
sepliste-XDATA noch nicht auf reale Separator-TeileIds aufgeloest sind.
"""
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).
"zuordnung_fix" (csv:block-to-json) hat Vorrang vor dem ZUORDNUNG-Attribut:
es steht an den temporaeren Kopien der in VF_n/GF_n/KREISEL_n VERPACKTEN
Separator_SP-Symbole und traegt die ID des Wrapper-Blocks, in dem das
Symbol steckt - eine bekannte, keine geratene Zuordnung.
"strecke_typ" (csv:block-to-json/csv:wrapper-strecke-typ) traegt bei
denselben verpackten Kopien den Streckentyp (VF/GF) des Wrappers, in dem
der Separator steckt - eine Strecke wechselt sich zwingend Foerderer/
Gefaellestrecke ab. Fehlt bei eigenstaendigen Separator_SP/Scanner-
Bloecken (keinem Wrapper zugeordnet) und bleibt dann leer."""
attribs = block.get("attribs", {})
return {
"Zuordnung": block.get("zuordnung_fix") or attribs.get("ZUORDNUNG", ""),
"ARTINR": attribs.get("ARTINR", ""),
"Höhe": get_hoehe(block),
"Drehung": get_drehung(block),
"Typ": block.get("strecke_typ", ""),
}
def build_element_merkmale(block):
"""Merkmale-Dict fuer eine ausgepackte VF_n/GF_n-Sub-Komponente (Bogen/
Streckenelement/Umlenkstation/Motorstation) - siehe csv:element-proxies-
erzeugen (export.lsp) und pattern_nested_bogen/_streckenelement/
_umlenkstation/_motorstation in cfg/export.cfg.
ARTINR/LAYER kommen direkt aus den ATTDEFs der ausgepackten Original-
Blockdefinition (ssg-attrib-read auf der temporaeren Kopie) - dieselbe
Quelle wie bei Separator/Scanner (build_sensor_merkmale), aber OHNE
"Zuordnung"-Feld: diese Elemente sind keine einem Carrier zugeordneten
Sensoren, sondern die Geometrie des Carriers selbst."""
attribs = block.get("attribs", {})
return {
"ARTINR": attribs.get("ARTINR", ""),
"Höhe": get_hoehe(block),
"Drehung": get_drehung(block),
}
def build_btmt_merkmale(block):
"""Merkmale-Dict fuer eine eigenstaendige BTMT-Beladung/SC_Entladung-Station
(siehe pattern_btmt_beladung/pattern_btmt_entladung in cfg/export.cfg).
Anders als build_sensor_merkmale ohne "Zuordnung": eine BTMT-Station ist
keinem Kreisel/GF/VF-Carrier zugeordnet, das ZUORDNUNG-Attribut existiert
hier nicht.
Drehung wird auf ganze Grad gerundet (0, 90, 180, ...) - anders als bei
Separator/Scanner/Bogen/Weiche, wo Kommastellen Bedeutung haben."""
attribs = block.get("attribs", {})
return {
"ARTINR": attribs.get("ARTINR", ""),
"Höhe": get_hoehe(block),
"Drehung": round(get_drehung(block)),
}
def build_geruest_merkmale(block):
"""Merkmale-Dict fuer einen GR_-Geruest-Baugruppenblock (siehe
pattern_geruest in cfg/export.cfg, GeruestInsert.lsp *geruest-attrib-defs*).
ARTINR/LAENGE/BEZEICHNUNG wandern in feste CSV-Spalten (teileid/
bezeichnung ueben in process_blocks unten), TEILEART/FARBE/LAYER bleiben
als Rest-Attribute in der Merkmale-JSON-Spalte - so wie bei Separator/
Scanner/BTMT bereits ueblich."""
attribs = block.get("attribs", {})
return {
"TeileArt": attribs.get("TEILEART", ""),
"Farbe": attribs.get("FARBE", ""),
"Laenge": attribs.get("LAENGE", ""),
"Layer": attribs.get("LAYER", ""),
}
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 _fallback_separator_bbox(cols, block, separator_box_mm=None):
"""Ergaenzt eine fehlende Bounding-Box (cols["_bbox"] is None) durch die
feste Separator-Symbol-Box aus cfg [Boundingbox] separator_box_* an der
realen INSERT-Position/-Drehung des Blocks. Wird fuer frei platzierte
Separator_SP-INSERTs gebraucht, deren vla-getboundingbox beim Export
scheiterte (z.B. eingefrorener Layer) - ohne "_bbox" fielen sie sonst aus
der Nachbarschafts-/Zuordnungserkennung. Hat der Block schon eine bbox,
bleibt cols unveraendert.
Rueckgabe: (ggf. ergaenztes) cols-dict."""
if cols.get("_bbox"):
return cols
if separator_box_mm is None:
separator_box_mm = SEPARATOR_BOX_MM
bbox = separator_bbox(block.get("x", 0.0) or 0.0, block.get("y", 0.0) or 0.0,
block.get("rotation", 0.0) or 0.0,
separator_box_mm=separator_box_mm,
z=block.get("z", 0.0) or 0.0)
cols = dict(cols)
cols["position"] = f'{bbox["cx"]:.2f}, {bbox["cy"]:.2f}, {bbox["cz"]:.2f}'
cols["boundingbox"] = f'{bbox["dx"]:.2f}, {bbox["dy"]:.2f}, {bbox["dz"]:.2f}'
cols["_bbox"] = bbox
return cols
def planquadrat_column(block, planquadrat_cfg):
"""Berechnet das Planquadrat aus der Bounding-Box des Blocks, falls
vorhanden (siehe export_planquadrat.py), sonst aus dem Einfuegepunkt.
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).
Ueberdeckt die Bounding-Box des Elements mehr als eine Zelle (z.B. ein
Kreisel oder eine lange VarioFoerderer-Kette), liefert compute_planquadrat
eine SPANNE "start-ende" (z.B. "A/1-J/1") statt nur der Mittelpunkt-Zelle.
"""
if planquadrat_cfg is None:
return ""
bbox = block.get("bbox") or {}
x = bbox.get("cx", block.get("x"))
y = bbox.get("cy", block.get("y"))
if x is None or y is None:
return ""
dx = bbox.get("dx", 0.0)
dy = bbox.get("dy", 0.0)
return compute_planquadrat(float(x), float(y), planquadrat_cfg, dx_mm=float(dx), dy_mm=float(dy))
# ---------------------------------------------------------------------------
# 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 max_numeric_teileid(items):
"""Hoechste rein-numerische TeileId in items (z.B. "0008" -> 8). Nicht-
numerische IDs (SEP-0001, autogenerated_of_json, ...) werden ignoriert.
0, wenn keine numerische ID existiert. Basis fuer die fortlaufende
Nummerierung der am Kreisel-Beruehrpunkt erzeugten ILS-Weichen (siehe
'Zaehle die Id aus der letzten enthaltenen ID hoch')."""
max_id = 0
for item in items:
tid = str(item.get("teileid", "")).strip()
if tid.isdigit():
max_id = max(max_id, int(tid))
return max_id
def build_kreisel_weiche_items(items, planquadrat_cfg, start_elem_nr, switches, dbg=None):
"""Haengt fuer jeden Kreisel-Beruehrpunkt eine "ILS Weiche"-Zeile an items
an (siehe export_neighbors.compute_kreisel_touch_switches).
switches = Rueckgabe von compute_kreisel_touch_switches, bereits VOR dem
Aufruf berechnet (der Aufrufer speist dieselbe Liste auch in
compute_kreisel_umlauf ein, damit die Weiche im Kreisel-Umlauf auftaucht -
siehe process_blocks). Jeder switches-Eintrag traegt seinen stabilen
"key" (sw["key"]) - hier wird KEIN neuer Schluessel erzeugt.
TeileId zaehlt aus der hoechsten bereits enthaltenen numerischen ID hoch
(max_numeric_teileid), 4-stellig aufgefuellt wie die Kreisel-IDs. Anzahl
ist wie bei allen Zeilen 1; Nachbarn = die beiden Kreisel-IDs (durch den
Kreisel-Umlauf ggf. bereits um die eigentlichen Umlauf-Nachbarn ergaenzt/
ersetzt, siehe process_blocks); K1-K4 und Insertpoint bleiben leer;
Position = Beruehrkoordinaten; Boundingbox = die feste, an der
Kreiselachse ausgerichtete Element-Box (element_box_mm, cfg/export.cfg
[Boundingbox]).
Rueckgabe: (fortgeschriebene Elementnummer, dict switches-"key" -> erzeugtes
Item) - die Key-Zuordnung braucht process_blocks, um nach dem Kreisel-
Umlauf die tatsaechliche "Nachbarn"-Spalte der Weiche zu setzen.
"""
naechste_id = max_numeric_teileid(items)
elem_nr = start_elem_nr
weiche_by_key = {}
for n, sw in enumerate(switches, start=1):
elem_nr += 1
naechste_id += 1
teileid = f"{naechste_id:04d}"
planquadrat = (compute_planquadrat(sw["x"], sw["y"], planquadrat_cfg)
if planquadrat_cfg is not None else "")
item = {
"nr": elem_nr,
"teileart": WEICHE_TEILEART,
"teileid": teileid,
"bezeichnung": f"ILS Weiche :{n}",
"planquadrat": planquadrat,
"position": f'{sw["x"]:.2f}, {sw["y"]:.2f}, {sw["z"]:.2f}',
"boundingbox": f'{sw["dx"]:.2f}, {sw["dy"]:.2f}, {sw["dz"]:.2f}',
"insertpoint": "",
"k1": "", "k2": "", "k3": "", "k4": "",
"nachbarn": ", ".join(sw["nachbarn"]),
"fehler": "",
"merkmale": {
"Erzeugt aus": "Kreisel-Beruehrpunkt",
"Kreisel A": sw["nachbarn"][0],
"Kreisel B": sw["nachbarn"][1],
"Achse von Kreisel": sw["achse_von"],
"Beruehrung": sw["beruehrung"],
},
}
items.append(item)
weiche_by_key[sw["key"]] = item
return elem_nr, weiche_by_key
def _strecke_typ_feld(teileart):
"""Merkmals-Feldname fuer die Ketten-Zugehoerigkeit einer AS/ES-/Separator-
Zeile, je Kettentyp unterschiedlich (User-Vorgabe): VF_n ("ILS 2.0
Strecke") -> "Förderer", GF_n ("ILS 2.0 Gefaellestrecke") -> "Strecke"
(wie bisher)."""
return "Strecke" if teileart == "ILS 2.0 Gefaellestrecke" else "Förderer"
def map_sepliste_kette_items(items, sep_kette, dbg=None):
"""Ordnet jeden sepliste-Eintrag (AS/SEP/ES, siehe compute_sep_kette)
einem bereits klassifizierten Item zu und liefert sie unter ihrem
sep_kette-Schluessel zurueck - fuer die Nachbar-Aufloesung in
resolve_umlauf_und_sepliste_nachbarn sowie die TrackIds-Spalte der
besitzenden Strecke (siehe write_separatorliste_merkmale).
ERZEUGT KEINE neuen Zeilen: AS/ES liegen bereits als eigene Items vor
(direkt aus dem dauerhaften ZZ_ASES_MARKER klassifiziert, siehe
process_blocks), SEP als die beim Export ausgepackte temporaere
Separator_SP-Kopie (csv:sep-proxies-erzeugen, export.lsp). Beide Wege
wuerden sonst dieselben physischen Bauteile DOPPELT zaehlen.
Zuordnung: EXAKTER ID-Abgleich. Seit csv:sepliste-ids-setzen/
csv:asmark-ids-setzen (Lisp/export.lsp) traegt praktisch jeder sepliste-
Eintrag bereits die TeileId des ihn vertretenden Items als Feld "id" -
das Item mit GENAU dieser teileid IST der gesuchte Partner. Kein Rate-/
Naeheverfahren mehr noetig, keine der frueheren Ambiguitaeten (ein
einzelner fehlerhafter sepliste-Koordinatenwert konnte vorher per
Naeherungssuche einen FALSCHEN Nachbarn waehlen und einen anderen
Separator ganz ohne Zuordnung lassen - siehe Memory
sepliste-lfdnr2-koordinaten-bug.md).
Fallback (kein "id"-Feld - Altbestand-Export ohne Marker/Kopie-ID, oder
deren Zuweisung ist in Lisp fehlgeschlagen): NUR fuer SEP-Eintraege
weiterhin gegen die per Zuordnung vorgefilterten Separator-Items, je
Strecke der erste noch nicht verbrauchte Kandidat (Baureihenfolge in
items). AS/ES ohne "id" bleiben unaufgeloest - fuer sie existiert seit
dem Wegfall der Kreisel<->Strecke-Kollisionserzeugung (frueher
compute_strecke_kreisel_schleus/build_strecke_schleus_items) ohnehin
kein Item mehr, das sie vertreten koennte.
Der Fallback matchte frueher per naechstem sepliste-Koordinatenpunkt
(x/y) - die sepliste-XDATA traegt seit 2026-09-25 kein x/y mehr, weil
diese Koordinaten der rohe, teils gespiegelte/verschobene Gruppencode-10-
Punkt des jeweiligen Bau-Primitivs waren (siehe export.lsp-Kommentar bei
sepliste-teile) und keine verlaessliche Zuordnungsgrundlage. Der
ID-Abgleich oben deckt praktisch jeden aktuellen Export ab; dieser
Fallback ist reine Altbestands-Kompatibilitaet.
Rueckgabe: sep_kette_by_key (neues dict).
"""
sep_kette_by_key = {}
by_teileid = {it.get("teileid", ""): it for it in items if it.get("teileid", "")}
# Kandidatenpool nur noch fuer den Fallback (SEP ohne "id"); verbrauchte
# Kandidaten werden entfernt, damit zwei Sepliste-Eintraege nicht auf
# dasselbe Item fallen.
sep_items_by_zuordnung = {}
for it in items:
if it.get("teileart", "") != "ILS 2.0 Separator":
continue
zuordnung = it.get("merkmale", {}).get("Zuordnung", "")
if zuordnung:
sep_items_by_zuordnung.setdefault(zuordnung, []).append(it)
if dbg:
dbg("=== map_sepliste_kette_items: sepliste-Eintrag -> Item ===")
per_id = 0
per_fallback = 0
ohne_treffer = 0
for key, eintrag in sep_kette.items():
marker_id = eintrag.get("id", "")
if marker_id:
item = by_teileid.get(marker_id)
if item is not None:
sep_kette_by_key[key] = item
per_id += 1
if dbg:
dbg(f" {key} (typ={eintrag.get('typ')!r}) -> Item per ID "
f"{marker_id!r} (TeileId {item.get('teileid', '')!r})")
continue
if dbg:
dbg(f" {key}: id={marker_id!r} gesetzt, aber KEIN Item mit "
f"dieser TeileId gefunden")
# Fallback nur fuer SEP ohne (aufloesbares) "id"-Feld (Altbestand).
if eintrag.get("typ") != "SEP":
ohne_treffer += 1
continue
kandidaten = sep_items_by_zuordnung.get(eintrag["strecke_id"], [])
if not kandidaten:
ohne_treffer += 1
if dbg:
dbg(f" {key}: kein Separator-Item mit Zuordnung "
f"{eintrag['strecke_id']!r} gefunden - uebersprungen")
continue
best = kandidaten.pop(0)
sep_kette_by_key[key] = best
per_fallback += 1
if dbg:
dbg(f" {key} -> Fallback (erster unverbrauchter Kandidat) auf "
f"Separator-Item TeileId {best.get('teileid', '')!r}")
if dbg:
dbg(f" Zuordnung: {per_id} per ID, {per_fallback} per Naehe-Fallback, "
f"{ohne_treffer} ohne Treffer")
return sep_kette_by_key
def resolve_umlauf_und_sepliste_nachbarn(items, kreisel_umlauf, sep_kette,
sep_kette_by_key, weiche_by_key, dbg=None):
"""Loest die "Nachbarn"-Spalte aller Zeilen auf, die build_kreisel_weiche_items
zunaechst leer gelassen bzw. map_sepliste_kette_items nur registriert hat,
weil ihr(e) Nachbar(n) beim eigenen Bau/der Registrierung noch nicht als
TeileId feststand(en) (Weiche vs. Separator im Kreisel-Umlauf, oder eine
sepliste-Zeile, deren Nachfolger erst danach registriert wird).
Muss NACH build_kreisel_weiche_items UND map_sepliste_kette_items laufen
(deren TeileIds stehen erst dann fest). items wird IN-PLACE veraendert
(item["nachbarn"]).
kreisel_umlauf = Rueckgabe von export_neighbors.compute_kreisel_umlauf.
sep_kette = Rueckgabe von export_neighbors.compute_sep_kette.
sep_kette_by_key = Rueckgabe von map_sepliste_kette_items (dict
sep_kette-Schluessel -> Item).
weiche_by_key = dict switches-"key" -> erzeugtes Item (von
build_kreisel_weiche_items).
"""
# Gemeinsamer Lookup fuer BEIDE Schluesselarten, die in kreisel_umlauf und
# sep_kette als punkt_key/vorgaenger/nachfolger auftreten koennen: freie
# Punkte (Separator/BTMT Beladung/BTMT Entladung, TeileId - siehe
# KREISEL_UMLAUF_TEILEARTEN), Kreisel-Kreisel-Weichen
# (weiche_by_key) UND sepliste-Zeilen (sep_kette_by_key, Schluessel
# "<strecke_id>#<lfdnr>"). Ein Umlauf-Punkt kann naemlich ein AS/ES-Ende
# sein (sep_kette-Schluessel) und ein sep_kette-Vorgaenger/Nachfolger kann
# das freie Kreiselende gar nicht sein (das loest ausschliesslich
# compute_kreisel_umlauf auf) - erst der gemeinsame Lookup deckt alle
# Kombinationen ab.
freipunkt_by_id = {it.get("teileid", ""): it for it in items
if it.get("teileart", "") in KREISEL_UMLAUF_TEILEARTEN}
alle_schluessel = {**sep_kette_by_key, **weiche_by_key, **freipunkt_by_id}
if dbg:
dbg("=== resolve_umlauf_und_sepliste_nachbarn ===")
dbg(f" freipunkt_by_id: {sorted(freipunkt_by_id.keys())}")
dbg(f" sep_kette_by_key: "
f"{[(k, v.get('teileid', '')) for k, v in sep_kette_by_key.items()]}")
dbg(f" weiche_by_key: "
f"{[(k, v.get('teileid', '')) for k, v in weiche_by_key.items()]}")
dbg("")
dbg(" -- Pass 1: aus kreisel_umlauf --")
# Freie Separatoren, Weichen UND die AS/ES-Enden andockender Ketten
# bekommen ihre "Nachbarn"-Spalte aus dem Kreisel-Umlauf.
if kreisel_umlauf:
for kreisel_tid, punkte in kreisel_umlauf.items():
for punkt_key, vorgaenger, nachfolger in punkte:
item = alle_schluessel.get(punkt_key)
if item is None:
if dbg:
dbg(f" Kreisel {kreisel_tid} Punkt {punkt_key!r}: "
f"KEIN Item in alle_schluessel gefunden - uebersprungen")
continue
nachbarn = []
for nachbar_key in (vorgaenger, nachfolger):
if not nachbar_key:
continue
nachbar_item = alle_schluessel.get(nachbar_key)
if nachbar_item is not None:
nachbarn.append(nachbar_item.get("teileid", ""))
elif dbg:
dbg(f" Kreisel {kreisel_tid} Punkt {punkt_key!r}: "
f"Nachbar-Key {nachbar_key!r} nicht in alle_schluessel aufloesbar")
teileid = item.get("teileid", "")
if nachbarn:
item["nachbarn"] = ", ".join(nachbarn)
if dbg:
dbg(f" Kreisel {kreisel_tid} Punkt {punkt_key!r} "
f"(TeileId {teileid!r}): nachbarn={nachbarn}")
elif dbg:
dbg(f" Kreisel {kreisel_tid} Punkt {punkt_key!r} "
f"(TeileId {teileid!r}): KEINE Nachbarn aufgeloest "
f"(vorgaenger={vorgaenger!r}, nachfolger={nachfolger!r})")
if dbg:
dbg("")
dbg(" -- Pass 2: ketteninterne sepliste-Nachbarn --")
# Ketteninterne sepliste-Nachbarn (SEP <-> SEP, SEP <-> AS/ES) - das
# jeweilige Kettenende (vorgaenger/nachfolger ist None, z.B. das AS-Ende
# ohne SEP davor) bleibt hier unveraendert, es wird stattdessen oben vom
# Kreisel-Umlauf gesetzt (compute_sep_kette dokumentiert diese Aufteilung).
for key, eintrag in sep_kette.items():
item = sep_kette_by_key.get(key)
if item is None:
if dbg:
dbg(f" sepliste-Key {key!r} (typ={eintrag.get('typ')!r}): "
f"KEIN Item in sep_kette_by_key - uebersprungen "
f"(siehe map_sepliste_kette_items)")
continue
nachbarn = []
for nachbar_key in (eintrag.get("vorgaenger"), eintrag.get("nachfolger")):
if not nachbar_key:
continue
nachbar_item = alle_schluessel.get(nachbar_key)
if nachbar_item is not None:
nachbarn.append(nachbar_item.get("teileid", ""))
teileid = item.get("teileid", "")
if nachbarn:
item["nachbarn"] = ", ".join(nachbarn)
if dbg:
dbg(f" sepliste-Key {key!r} (TeileId {teileid!r}): "
f"vorgaenger={eintrag.get('vorgaenger')!r} "
f"nachfolger={eintrag.get('nachfolger')!r} -> "
f"nachbarn={nachbarn} item.nachbarn={item.get('nachbarn', '')!r}")
def write_separatorliste_merkmale(items, sep_kette_by_key, dbg=None):
"""Traegt in jede Strecke (VF_n/GF_n mit "sepliste") die eigene CSV-Spalte
"TrackIds" (item["trackids"]) ein: die TeileIds von AS-Ende, Separatoren
und ES-Ende in Baureihenfolge, kommagetrennt (analog zur "Nachbarn"-
Spalte, ebenfalls ein Top-Level-Item-Feld statt eines Merkmals), z.B.
"0051, 0052, 0053, 0054, 0055" (AS ... SEP ... ES).
Quelle ist die rohe "sepliste" des Items (AS/SEP/ES mit x/y, bereits in
Baureihenfolge - siehe compute_sep_kette) statt sep_kette selbst: dessen
dict-Reihenfolge ist nicht garantiert, die sepliste-Liste dagegen schon.
AS/ES zaehlen HIER mit (anders als beim "Separatorlist"-Merkmal frueher) -
sie sind eigene Items ("ILS Ausschleuselement"/"ILS Einschleuselement",
direkt aus dem dauerhaften ZZ_ASES_MARKER klassifiziert, siehe
process_blocks), analog zu den AS/ES-Enden, die beim Kreisel bereits ueber
compute_kreisel_umlauf in dessen TrackIds auftauchen - ohne sie fehlten
hier die Kettenenden, obwohl sie beim Kreisel schon drin sind.
Sowohl der Separator als auch AS/ES wurden bereits von
map_sepliste_kette_items ueber die sepliste-ID (Feld "id") auf ihr reales
Item (TeileId) aufgeloest, unter demselben "<strecke_id>#<lfdnr>"-
Schluessel in sep_kette_by_key eingetragen; hier wird dieser Schluessel
nur noch je Strecke eingesammelt. Muss darum NACH map_sepliste_kette_items
laufen.
"""
for item in items:
if item.get("teileart", "") not in GF_VF_TEILEARTEN:
continue
sepliste = item.get("sepliste")
if not sepliste:
continue
sid = item.get("teileid", "")
track_ids = []
for entry in sepliste:
key = f"{sid}#{entry.get('lfdnr')}"
sep_item = sep_kette_by_key.get(key)
if sep_item is None:
if dbg:
dbg(f" TrackIds {key}: kein aufgeloestes Item - uebersprungen")
continue
track_ids.append(sep_item.get("teileid", ""))
if track_ids:
item["trackids"] = ", ".join(track_ids)
def write_kreisel_separatorliste_merkmale(items, kreisel_umlauf, sep_kette_by_key,
weiche_by_key, dbg=None):
"""Traegt in jeden Kreisel ("ILS 2.0 Kreisel") die eigene CSV-Spalte
"TrackIds" (item["trackids"], gleiches Feld wie bei VF_n/GF_n, siehe
write_separatorliste_merkmale) ein: die TeileIds aller auf ihm liegenden
freien Punkte (Separator, BTMT Beladung/Entladung - siehe
KREISEL_UMLAUF_TEILEARTEN), andockenden AS/ES-Streckenenden und Kreisel-
Kreisel-Weichen, in Umlaufreihenfolge ab Winkel 0 Grad (siehe
compute_kreisel_umlauf - die Liste beginnt dort bereits bei 0 Grad in
Drehrichtung DREHRICHTUNG), kommagetrennt.
Quelle ist kreisel_umlauf (export_neighbors.compute_kreisel_umlauf), das
fuer jeden Kreisel bereits alle Punkte (freie Punkte, AS/ES-Enden,
Kreisel-Kreisel-Weichen) nach Winkel um die Kreiselachse sortiert hat -
hier werden nur noch deren rohe Schluessel (TeileId direkt, sep_kette-
Schluessel oder Weichen-"key") auf die TeileId des jeweils zugehoerigen
Items aufgeloest. Derselbe Lookup wie in
resolve_umlauf_und_sepliste_nachbarn, daher muss diese Funktion NACH
map_sepliste_kette_items/build_kreisel_weiche_items laufen (deren
TeileIds stehen erst dann fest).
"""
freipunkt_by_id = {it.get("teileid", ""): it for it in items
if it.get("teileart", "") in KREISEL_UMLAUF_TEILEARTEN}
alle_schluessel = {**sep_kette_by_key, **weiche_by_key, **freipunkt_by_id}
kreisel_by_id = {it.get("teileid", ""): it for it in items
if it.get("teileart", "") == "ILS 2.0 Kreisel"}
for tid, punkte in (kreisel_umlauf or {}).items():
kreisel_item = kreisel_by_id.get(tid)
if kreisel_item is None or not punkte:
continue
teileids = []
for punkt_key, _vorgaenger, _nachfolger in punkte:
item = alle_schluessel.get(punkt_key)
if item is None:
if dbg:
dbg(f" TrackIds Kreisel {tid}: kein Item fuer "
f"Umlauf-Punkt {punkt_key!r} - uebersprungen")
continue
teileids.append(item.get("teileid", ""))
if teileids:
kreisel_item["trackids"] = ", ".join(teileids)
def pruefe_ases_in_beiden_containern(items, dbg=None):
"""Prueft, ob jede AS-/ES-Zeile (AUSSCHLEUS_TEILEART/EINSCHLEUS_TEILEART)
in BEIDEN Containern auftaucht, zu denen sie gehoert:
- in den TrackIds ihrer EIGENEN Strecke (VF_n/GF_n, ueber das
ZUORDNUNG-Attribut des Markers - siehe Item-Feld "_zuordnung")
- in den TrackIds MINDESTENS eines Kreisels (der beruehrte Carrier,
geometrisch ueber compute_kreisel_umlauf/KREISEL_UMLAUF_TEILEARTEN
ermittelt - nur "ILS 2.0 Kreisel", Eckrad nimmt am Umlauf nicht teil,
siehe compute_kreisel_umlauf/KREISEL_SPLIT_TEILEART)
Muss NACH write_separatorliste_merkmale UND
write_kreisel_separatorliste_merkmale laufen (beide TrackIds-Spalten
muessen bereits stehen).
Ein AS/ES, das in einem der beiden Container fehlt, zeigt einen echten
Fehler in der Zuordnungskette an (z.B. ZUORDNUNG-Attribut leer/falsch,
Kollisionstoleranz zu knapp, Kreisel zu weit weg, fehlende Bounding-Box am
Marker) - kein Normalfall. Traegt in diesem Fall eine Warnung in
item["warnung"] ein (erscheint in der CSV-Spalte "Warnungen") und gibt sie
zusaetzlich auf der Konsole/im Debuglog aus.
Rueckgabe: Liste der betroffenen TeileIds (fuer Tests/Diagnose), leer bei
vollstaendiger Zuordnung.
"""
strecke_by_id = {it.get("teileid", ""): it for it in items
if it.get("teileart", "") in GF_VF_TEILEARTEN}
kreisel_trackid_sets = [
{t.strip() for t in (it.get("trackids", "") or "").split(",") if t.strip()}
for it in items if it.get("teileart", "") == KREISEL_SPLIT_TEILEART
]
auffaellig = []
for it in items:
if it.get("teileart", "") not in (AUSSCHLEUS_TEILEART, EINSCHLEUS_TEILEART):
continue
teileid = it.get("teileid", "")
fehlend = []
strecke = strecke_by_id.get(it.get("_zuordnung", ""))
if strecke is None:
fehlend.append("eigene Strecke (ZUORDNUNG unaufloesbar)")
else:
strecke_ids = {t.strip() for t in (strecke.get("trackids", "") or "").split(",") if t.strip()}
if teileid not in strecke_ids:
fehlend.append("eigene Strecke")
if not any(teileid in ids for ids in kreisel_trackid_sets):
fehlend.append("Kreisel")
if fehlend:
auffaellig.append(teileid)
warnung = f"nicht in TrackIds von: {', '.join(fehlend)}"
bisherige = it.get("warnung", "")
it["warnung"] = f"{bisherige}; {warnung}" if bisherige else warnung
meldung = f"[export_csv] WARNUNG: {it.get('teileart', '')} {teileid}: {warnung}"
print(meldung)
if dbg:
dbg(f" {meldung}")
return auffaellig
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
strecke_bogen_count = 0
streckenelement_count = 0
umlenkstation_count = 0
motorstation_count = 0
separator_count = 0
scanner_count = 0
btmt_beladung_count = 0
btmt_entladung_count = 0
eingang_count = 0
ausgang_count = 0
geruest_count = 0
ausschleus_count = 0
einschleus_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
# ZZ_ASES_MARKER (siehe cfg/export.cfg pattern_ases_marker sowie
# csv:asmark-aktualisieren in Lisp/export.lsp): der persistente
# Marker, den Lisp an jedes AS-/ES-Kettenende setzt. Er IST die
# AS-/ES-Zeile - TeileId, Position und Boundingbox kommen direkt vom
# Marker-Block selbst (echtes INSERT, echte vla-getboundingbox-Box),
# kein synthetischer Bau-Schritt mehr noetig (fruehere Version:
# compute_strecke_kreisel_schleus/build_strecke_schleus_items suchten
# den anschliessenden Kreisel per Kollisionstest und erfanden dabei
# eine eigene, bei jedem Export neue TeileId). "Kreisel"-Merkmal und
# ID-basierte sep_kette-Zuordnung tragen unten zwei kurze Nachlaeufe
# in process_blocks nach, sobald alle Items klassifiziert sind (die
# eigene Kette bzw. der beruehrte Kreisel koennen an DIESER Stelle
# noch unklassifiziert sein).
if matches_any(bname, BLOCKPATTERNS.get("pattern_ases_marker", [])):
attribs = block.get("attribs", {})
typ = attribs.get("TYP", "")
teileart = {"AS": AUSSCHLEUS_TEILEART, "ES": EINSCHLEUS_TEILEART}.get(typ)
if teileart is None:
# Sollte nicht vorkommen (TYP wird von csv:asmark-aktualisieren
# immer gesetzt) - defensiv ueberspringen statt abzustuerzen.
print(f"[export_csv] WARNUNG: {bname} ohne gueltiges TYP-Attribut "
f"(TYP={typ!r}) - uebersprungen")
continue
elem_nr += 1
if typ == "AS":
ausschleus_count += 1
zaehler_n = ausschleus_count
else:
einschleus_count += 1
zaehler_n = einschleus_count
items.append({
"nr": elem_nr,
"teileart": teileart,
"teileid": attribs.get("ID") or format_sensor_id("ASES", elem_nr),
"bezeichnung": f"{teileart} :{zaehler_n}",
"planquadrat": planquadrat_column(block, planquadrat_cfg),
"merkmale": {
"Streckenende": "Anfang" if typ == "AS" else "Ende",
"Element": attribs.get("ELEMENT", ""),
"ARTINR": attribs.get("ARTINR", ""),
},
# Private Felder (nicht in der CSV) fuer die beiden Nachlaeufe
# weiter unten in process_blocks: "_zuordnung" = ID des
# eigenen VF_n/GF_n-Wrappers (fuer das "Förderer"/"Strecke"-
# Merkmal), "_lfdnr" wird hier nicht gebraucht (die sep_kette-
# Zuordnung laeuft exakt ueber die TeileId, siehe
# map_sepliste_kette_items).
"_zuordnung": attribs.get("ZUORDNUNG", ""),
**bbox_columns(block),
})
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": bezeichnung_oder_fallback(block, f"VarioFoerderer :{vf_count}"),
"planquadrat": planquadrat_column(block, planquadrat_cfg),
"merkmale": build_variofoerderer_merkmale(block),
# Sepliste-XDATA (AS/SEP/ES-Reihenfolge) durchreichen - Quelle
# fuer compute_sep_kette/map_sepliste_kette_items (Ketten-
# interne Separatoren + AS/ES-Nachbarn). Ohne diese Weitergabe
# liefe die gesamte sepliste-basierte Nachbarverkettung ins
# Leere (das Feld stand im JSON, wurde aber nicht ans Item
# uebernommen).
"sepliste": block.get("sepliste"),
**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": bezeichnung_oder_fallback(block, f"Gefaellestrecke :{gf_count}"),
"planquadrat": planquadrat_column(block, planquadrat_cfg),
"merkmale": build_variofoerderer_merkmale(block),
# Sepliste-XDATA durchreichen (siehe VF_N oben).
"sepliste": block.get("sepliste"),
**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": bezeichnung_oder_fallback(block, 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": bezeichnung_oder_fallback(block, 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
# Ausgepackte VF_n/GF_n-Sub-Komponenten (Boegen, gerade Foerderer-/
# Gefaellestrecken-Segmente, Umlenk-/Motorstation) - temporaere Kopien,
# siehe csv:element-proxies-erzeugen (export.lsp). MUESSEN vor
# pattern_strecke_module geprueft werden: dessen Muster ("Vario*,
# Staustrecke*, ...") wuerde diese Bloecke sonst zuerst abfangen und
# als generische "ILS 2.0 Strecke - Modul"-Zeile ohne K1/K2 ausgeben.
# k1/k2 brauchen hier keine Sonderbehandlung - bbox_columns() liest
# sie bereits generisch aus dem "k1"/"k2"-Feld des JSON-Rows
# (csv:get-k-kos-strings), genau wie bei jedem anderen Block.
if matches_any(bname, BLOCKPATTERNS.get("pattern_nested_bogen", [])):
elem_nr += 1
strecke_bogen_count += 1
items.append({
"nr": elem_nr,
"teileart": "ILS 2.0 Bogen",
"teileid": block.get("attribs", {}).get("ID") or format_sensor_id("BOG", strecke_bogen_count),
"bezeichnung": f"Bogen :{strecke_bogen_count}",
"planquadrat": planquadrat_column(block, planquadrat_cfg),
"merkmale": build_element_merkmale(block),
**bbox_columns(block),
})
continue
if matches_any(bname, BLOCKPATTERNS.get("pattern_nested_streckenelement", [])):
elem_nr += 1
streckenelement_count += 1
items.append({
"nr": elem_nr,
"teileart": "ILS 2.0 Streckenelement",
"teileid": block.get("attribs", {}).get("ID") or format_sensor_id("STR", streckenelement_count),
"bezeichnung": f"Streckenelement :{streckenelement_count}",
"planquadrat": planquadrat_column(block, planquadrat_cfg),
"merkmale": build_element_merkmale(block),
**bbox_columns(block),
})
continue
if matches_any(bname, BLOCKPATTERNS.get("pattern_nested_umlenkstation", [])):
elem_nr += 1
umlenkstation_count += 1
items.append({
"nr": elem_nr,
"teileart": "ILS 2.0 Umlenkstation",
"teileid": block.get("attribs", {}).get("ID") or format_sensor_id("UML", umlenkstation_count),
"bezeichnung": f"Umlenkstation :{umlenkstation_count}",
"planquadrat": planquadrat_column(block, planquadrat_cfg),
"merkmale": build_element_merkmale(block),
**bbox_columns(block),
})
continue
if matches_any(bname, BLOCKPATTERNS.get("pattern_nested_motorstation", [])):
elem_nr += 1
motorstation_count += 1
items.append({
"nr": elem_nr,
"teileart": "ILS 2.0 Motorstation",
"teileid": block.get("attribs", {}).get("ID") or format_sensor_id("MOT", motorstation_count),
"bezeichnung": f"Motorstation :{motorstation_count}",
"planquadrat": planquadrat_column(block, planquadrat_cfg),
"merkmale": build_element_merkmale(block),
**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
cols = bbox_columns(block)
# Fehlende Bounding-Box (vla-getboundingbox scheiterte, z.B.
# eingefrorener Layer) durch die feste Separator-Symbol-Box aus
# cfg [Boundingbox] separator_box_* ersetzen - sonst nimmt der
# Separator an der Nachbarschafts-/Umlauf-/Zuordnungserkennung
# nicht teil (die brauchen "_bbox"). Position/Drehung des realen
# INSERTs sind ja bekannt.
cols = _fallback_separator_bbox(cols, block)
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),
# Privates Feld (nicht in der CSV): markiert eine BEKANNTE
# Zuordnung, die compute_sensor_zuordnung unten nicht
# ueberschreiben darf - siehe build_sensor_merkmale.
"_zuordnung_fix": bool(block.get("zuordnung_fix")),
# Roher INSERT-Punkt (nicht der bbox-Mittelpunkt) - dient
# map_sepliste_kette_items als Koordinaten-FALLBACK, falls der
# sepliste-Eintrag (noch) kein "id"-Feld traegt (Altbestand;
# der Normalfall loest exakt per ID auf, siehe dort).
"_x": block.get("x", 0.0), "_y": block.get("y", 0.0),
**cols,
})
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),
# "warnung" kommt von cs-zuordnung-lauf (Lisp/count_sep_scan.lsp),
# das vor dem Export strittige Distanz-Zuordnungen (Scanner
# ausserhalb jeder Kreisel-/Strecken-Boundingbox, Abstand zu zwei
# Kandidaten fast gleich) in *cs-scanner-warnung-by-handle*
# sammelt - csv:block-to-json (export.lsp) haengt den Text als
# "warnung"-Feld an den betroffenen Scanner-Block im JSON an.
"warnung": block.get("warnung", ""),
**bbox_columns(block),
})
continue
# BTMT-Beladestation - eigene TeileArt-Zeile, sonst analog zu Separator/
# Scanner oben. Anders als dort NICHT von export_sivas.py ausgeschlossen
# (siehe SEP_SCANNER_PATTERNS-Kommentar) - eine BTMT-Station wird von
# keinem anderen Element mitgezaehlt.
if matches_any(bname, BLOCKPATTERNS.get("pattern_btmt_beladung", [])):
elem_nr += 1
btmt_beladung_count += 1
items.append({
"nr": elem_nr,
"teileart": "ILS 2.0 BTMT Beladung",
"teileid": block.get("attribs", {}).get("ID") or format_sensor_id("BEL", btmt_beladung_count),
"bezeichnung": f"BTMT Beladung :{btmt_beladung_count}",
"planquadrat": planquadrat_column(block, planquadrat_cfg),
"merkmale": build_btmt_merkmale(block),
**bbox_columns(block),
})
continue
# BTMT-Entladestation (Abwurfstation) - analog zur Beladestation oben.
if matches_any(bname, BLOCKPATTERNS.get("pattern_btmt_entladung", [])):
elem_nr += 1
btmt_entladung_count += 1
items.append({
"nr": elem_nr,
"teileart": "ILS 2.0 BTMT Entladung",
"teileid": block.get("attribs", {}).get("ID") or format_sensor_id("ENT", btmt_entladung_count),
"bezeichnung": f"BTMT Entladung :{btmt_entladung_count}",
"planquadrat": planquadrat_column(block, planquadrat_cfg),
"merkmale": build_btmt_merkmale(block),
**bbox_columns(block),
})
continue
# Eingangssymbol (Materialfluss-Eingang der Anlage) - eigene
# TeileArt-Zeile, analog zu BTMT. Wie BTMT von keinem Carrier-Attribut
# mitgezaehlt; anders als BTMT NUR im CSV-Export (export_sivas.py
# ueberspringt Eingang/Ausgang). Die TeileId kommt aus dem ID-Attribut
# (ssg-id-check-all fuellt es vor jedem Export, ID-ATTDEF seit dem
# allocations.json-Nachtrag) - dieselbe TeileId, die die
# "TrackIds"-Spalte an Kreisel/Strecke referenziert (KREISEL_UMLAUF_
# TEILEARTEN in export_neighbors.py). Fallback-Nummer wie bei BTMT.
if matches_any(bname, BLOCKPATTERNS.get("pattern_eingang", [])):
elem_nr += 1
eingang_count += 1
items.append({
"nr": elem_nr,
"teileart": "Eingang",
"teileid": block.get("attribs", {}).get("ID") or format_sensor_id("EIN", eingang_count),
"bezeichnung": f"Eingang :{eingang_count}",
"planquadrat": planquadrat_column(block, planquadrat_cfg),
"merkmale": build_btmt_merkmale(block),
**bbox_columns(block),
})
continue
# Ausgangssymbol (Materialfluss-Ausgang der Anlage) - analog zum
# Eingangssymbol oben.
if matches_any(bname, BLOCKPATTERNS.get("pattern_ausgang", [])):
elem_nr += 1
ausgang_count += 1
items.append({
"nr": elem_nr,
"teileart": "Ausgang",
"teileid": block.get("attribs", {}).get("ID") or format_sensor_id("AUS", ausgang_count),
"bezeichnung": f"Ausgang :{ausgang_count}",
"planquadrat": planquadrat_column(block, planquadrat_cfg),
"merkmale": build_btmt_merkmale(block),
**bbox_columns(block),
})
continue
# Geruest-Baugruppe (GR_<id>_<timestamp>) - eigene TeileArt-Zeile,
# analog zu BTMT: kein ANZAHL_*-Attribut an einem Carrier zaehlt sie
# mit. TeileId ist das echte ID-Attribut (ssg-id-check-all fuellt es
# vor jedem Export, siehe *geruest-attrib-defs*); ARTINR (Sivas-IDs,
# ggf. mehrere kommagetrennt) landet als eigene Spalte in den
# Merkmalen statt in "teileart" - anders als bei Kreisel/BTMT ist
# "TEILEART" hier ein FREIES Attribut (Rohrabhaenger/Saeule/...), keine
# feste Konstante.
if matches_any(bname, BLOCKPATTERNS.get("pattern_geruest", [])):
elem_nr += 1
geruest_count += 1
attribs = block.get("attribs", {})
bezeichnung = attribs.get("BEZEICHNUNG", "").strip()
items.append({
"nr": elem_nr,
"teileart": "Geruest",
"teileid": attribs.get("ID") or format_sensor_id("GR", geruest_count),
"bezeichnung": bezeichnung if bezeichnung else f"Geruest :{geruest_count}",
"planquadrat": planquadrat_column(block, planquadrat_cfg),
"merkmale": {"ARTINR": attribs.get("ARTINR", ""), **build_geruest_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,
element_box_mm=ELEMENT_BOX_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", "") not in SENSOR_TEILEARTEN or not zuordnung:
continue
# Bekannte Zuordnung (verpackter Separator, Carrier = Wrapper-Block,
# siehe "_zuordnung_fix"/build_sensor_merkmale) nicht ueberschreiben:
# die BBox-Suche hier priorisiert GF > Foerderer > Kreiselhaelfte und
# wuerde einen in einem VF-Block verpackten Separator an eine
# Gefaellestrecke haengen, in deren Box er zufaellig hineinragt.
if item.get("_zuordnung_fix"):
continue
# Kein Treffer ("nicht zugeordnet") ersetzt einen vorhandenen
# ZUORDNUNG-Attributwert nicht - sonst geht die in der Zeichnung
# gesetzte Zuordnung verloren (siehe Docstring von
# compute_sensor_zuordnung: ersetzen nur bei echtem Treffer).
if zuordnung == "nicht zugeordnet" and item["merkmale"].get("Zuordnung"):
continue
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
# Ketteninterne Separator-/AS-/ES-Reihenfolge aus der neuen "sepliste"-
# XDATA (siehe ssg-sepliste-xdata-schreiben) - leeres dict, wenn keine
# Strecke eine sepliste traegt (reiner Altbestand ohne neue XDATA).
sep_kette = compute_sep_kette(items, dbg=dbg)
# Kreisel-Kreisel-Weichen VOR dem Umlauf berechnen (reine Geometrie,
# braucht keine TeileIds) - dieselbe Liste geht sowohl in
# compute_kreisel_umlauf (damit die Weiche als Verzweigungspunkt in
# BEIDEN beruehrenden Kreisel-Umlaeufen auftaucht) als auch in
# build_kreisel_weiche_items (das die tatsaechlichen Zeilen erzeugt).
touch_switches = compute_kreisel_touch_switches(
items, NEIGHBOR_TOLERANCE_MM, durchmesser_mm=KREISEL_DURCHMESSER_MM,
element_box_mm=ELEMENT_BOX_MM, dbg=dbg)
# Umlauf-Reihenfolge an jedem Kreisel: freie Separatoren, BTMT-Stationen
# UND die AS/ES-Enden andockender Ketten (alle zusammen KREISEL_UMLAUF_
# TEILEARTEN, einheitlich per eigener Bounding-Box zugeordnet - ein AS/ES
# bringt seine Box vom ZZ_ASES_MARKER direkt mit) sowie die Kreisel-
# Kreisel-Weichen (touch_switches), sortiert nach Winkel um die
# Kreiselachse (siehe compute_kreisel_umlauf - DREHRICHTUNG-Attribut
# bestimmt den Umlaufsinn).
kreisel_umlauf = compute_kreisel_umlauf(
items, NEIGHBOR_TOLERANCE_MM,
durchmesser_mm=KREISEL_DURCHMESSER_MM, touch_switches=touch_switches, dbg=dbg)
# Neue Zeilen, deren "Nachbarn"-Spalte aus einem sep_kette-Schluessel
# aufgeloest werden muss, erzeugen: nur noch die Kreisel-Kreisel-Weichen
# (AS/ES sind bereits als eigene Items klassifiziert, siehe process_blocks
# oben). Ein sep_kette-Schluessel kann auf eine Zeile zeigen, die ERST
# DANACH registriert wird, darum erst NACH allen Registrierungen aufloesen
# (siehe unten) statt inline - frueherer Bug schrieb sonst den rohen
# Schluessel ("0026#3") statt der TeileId.
elem_nr, weiche_by_key = build_kreisel_weiche_items(
items, planquadrat_cfg, elem_nr, touch_switches, dbg=dbg)
# "Förderer"/"Strecke"-Merkmal an jedem AS/ES nachtragen: die eigene Kette
# (ZUORDNUNG-Attribut des Markers, siehe "_zuordnung") kann bei der
# Klassifikation des Markers selbst noch unklassifiziert gewesen sein -
# der FeldNAME haengt von deren TeileArt ab (VF -> "Förderer",
# GF -> "Strecke", siehe _strecke_typ_feld), der WERT ist einfach die
# ZUORDNUNG selbst (keine Geometrie mehr noetig).
strecke_by_id = {it.get("teileid", ""): it for it in items
if it.get("teileart", "") in GF_VF_TEILEARTEN}
for it in items:
if it.get("teileart", "") not in (AUSSCHLEUS_TEILEART, EINSCHLEUS_TEILEART):
continue
strecke = strecke_by_id.get(it.get("_zuordnung", ""))
if strecke is not None:
it["merkmale"][_strecke_typ_feld(strecke.get("teileart", ""))] = \
strecke.get("teileid", "")
# sep_kette-Schluessel (AS/SEP/ES) auf ihre bereits vorhandenen Items
# aufloesen - exakt per ID, siehe map_sepliste_kette_items.
sep_kette_by_key = map_sepliste_kette_items(items, sep_kette, dbg=dbg)
# Spalte "TrackIds" je Strecke (Reihenfolge der SEP-TeileIds aus der
# sepliste, x/y bereits oben auf das reale Separator-Item aufgeloest).
write_separatorliste_merkmale(items, sep_kette_by_key, dbg=dbg)
resolve_umlauf_und_sepliste_nachbarn(
items, kreisel_umlauf, sep_kette, sep_kette_by_key, weiche_by_key, dbg=dbg)
# Spalte "TrackIds" je Kreisel: dieselben Umlauf-Punkte (freie
# Separatoren, AS/ES-Enden, Kreisel-Kreisel-Weichen), die oben schon die
# "Nachbarn"-Spalte gespeist haben, jetzt als eine einzige, bei 0 Grad
# beginnende Reihenfolge auf den Kreisel selbst geschrieben (analog zu
# write_separatorliste_merkmale bei VF_n/GF_n).
write_kreisel_separatorliste_merkmale(
items, kreisel_umlauf, sep_kette_by_key, weiche_by_key, dbg=dbg)
# Konsistenzpruefung: jedes AS/ES muss jetzt sowohl in den TrackIds seiner
# eigenen Strecke als auch in denen eines Kreisels auftauchen (siehe
# Docstring) - Abweichungen deuten auf einen echten Fehler in der
# Zuordnungskette hin und werden ueber "Warnungen" sichtbar gemacht.
pruefe_ases_in_beiden_containern(items, dbg=dbg)
# Nach dem Sammelheimen aller Zeilen: sortiert nach TeileId ausgeben, damit
# Elementnummer (erste Spalte) und TeileId nach Moeglichkeit uebereinstimmen
# (siehe format_csv_line/Elementnummer-Neuvergabe unten). Numerische IDs
# zuerst aufsteigend; nicht-numerische IDs (z.B. "autogenerated_of_json"
# der Omniflo-Sum-Zeile oder SEP-/SCN-Fallback-IDs ohne ID-Attribut)
# behalten stabil ihre bisherige Reihenfolge und stehen danach. Wurde kein
# Element geloescht, ergibt sich dadurch Elementnummer == TeileId.
def teileid_sort_key(indexed_item):
index, item = indexed_item
tid = str(item.get("teileid", "")).strip()
return (0, int(tid)) if tid.isdigit() else (1, index)
items = [item for _, item in sorted(enumerate(items), key=teileid_sort_key)]
for i, item in enumerate(items, start=1):
item["nr"] = i
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.get("trackids", ""))}"'
f';"{csv_quote(item["fehler"])}"'
f';"{csv_quote(item.get("warnung", ""))}"'
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;TrackIds;Fehler;Warnungen;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