#!/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 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 "-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 ") 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()