#!/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). Nach der Kollisions-/Nachbarschaftserkennung wird ausserdem an jedem Beruehrpunkt zweier Kreisel eine "ILS Weiche"-Zeile angehaengt (siehe build_kreisel_weiche_items / export_neighbors.compute_kreisel_touch_switches): Anzahl 1, Position = Beruehrkoordinaten, Nachbarn = die beiden Kreisel-IDs, K1-K4/Insertpoint leer, Boundingbox = feste Element-Box (element_box_mm, cfg/export.cfg [Boundingbox]), mit der Laengsseite senkrecht zur Achse des beruehrenden Kreisels; die TeileId zaehlt aus der hoechsten bereits vergebenen numerischen ID hoch. Anschliessend (TeileId-Zaehlung laeuft ueber die Weichen hinweg weiter) wird an jedem Ende einer Gefaellestrecke/eines Foerderers, an dem ein Kreisel anschliesst, ein Aus- (Anfang) bzw. Einschleuselement (Ende) an der Kollisions- position Kreisel<->Strecke angehaengt (siehe build_strecke_schleus_items / export_neighbors.compute_strecke_kreisel_schleus) - Format wie die Weiche, Nachbarn = Kreisel-ID + Strecken-ID. Zugeordnet wird je Streckenende der naechstgelegene ihrer BBox-Kreisel-Nachbarn. Warnungen = Spalte fuer Auffaelligkeiten. Aktuell: eine Strecke sollte genau zwei Kreisel (Anfang + Ende) beruehren - beruehrt ihre (in x/y gepruefte) Bounding-Box mehr, steht hier ein Hinweis und es wird zusaetzlich eine Konsolen-WARNUNG ausgegeben (mit exakten 3D-BBoxen sollte das nicht auftreten). Scanner-Zeilen bekommen dieselbe Spalte fuer strittige Distanz-Zuordnungen: liegt ein Scanner ausserhalb jeder Kreisel-/Streckengruppen-/Gefaellestrecke- Boundingbox und ist der Abstand zum naechsten und zweitnaechsten Carrier fast gleich gross, schreibt Lisp/count_sep_scan.lsp (cs-zuordnung-lauf, vor dem Export automatisch aufgerufen) einen Warnungstext mit den IDs beider Kandidaten in das "warnung"-Feld des Scanner-Blocks im Roh-JSON (siehe csv:block-to-json in export.lsp) - hier unveraendert uebernommen. Aufruf: python export_csv.py CSV-Format: Elementnummer;TeileArt;TeileId;Bezeichnung;Planquadrat;Anzahl;Position;Boundingbox; Insertpoint;K1;K2;K3;K4;Nachbarn;Fehler;Warnungen;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. Reihenfolge/Elementnummer: die Zeilen werden am Ende aufsteigend nach TeileId sortiert (numerische IDs) und die Elementnummer (erste Spalte) danach neu von 1 an vergeben (siehe teileid_sort_key in process_blocks) - wurde kein Element geloescht, stimmen Elementnummer und TeileId dadurch ueberein, geloeschte Elemente hinterlassen Luecken in der TeileId-Folge. Zeilen ohne numerische TeileId (Omniflo-Sum-Zeile, SEP-/SCN-Fallback-IDs ohne ID-Attribut) behalten stabil ihre bisherige Reihenfolge und stehen danach. """ 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_collision_debug_target, compute_neighbor_ids, compute_neighbor_errors, compute_sensor_zuordnung, compute_scanner_nearest_separator, compute_kreisel_touch_switches, compute_strecke_kreisel_schleus, SENSOR_TEILEARTEN, WEICHE_TEILEART, GF_VF_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() ELEMENT_BOX_MM = load_element_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 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). "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.""" 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), } 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 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 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, dbg=None): """Haengt fuer jeden Kreisel-Beruehrpunkt eine "ILS Weiche"-Zeile an items an (siehe export_neighbors.compute_kreisel_touch_switches). 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; 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 (start_elem_nr + Anzahl Weichen). """ switches = compute_kreisel_touch_switches( items, NEIGHBOR_TOLERANCE_MM, durchmesser_mm=KREISEL_DURCHMESSER_MM, element_box_mm=ELEMENT_BOX_MM, dbg=dbg) naechste_id = max_numeric_teileid(items) elem_nr = start_elem_nr 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 "") items.append({ "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"], }, }) return elem_nr def build_strecke_schleus_items(items, planquadrat_cfg, start_elem_nr, dbg=None): """Haengt fuer jedes Strecken-Ende mit anschliessendem Kreisel eine "ILS Ausschleuselement"- (Anfang) bzw. "ILS Einschleuselement"-Zeile (Ende) an items an (siehe export_neighbors.compute_strecke_kreisel_schleus). Muss NACH build_kreisel_weiche_items laufen, damit die TeileId-Zaehlung (max_numeric_teileid) auch die bereits erzeugten Weichen einschliesst. Position = Kollisionsposition Kreisel<->Strecke; Nachbarn = Kreisel-ID + Strecken-ID; Anzahl 1; K1-K4/Insertpoint leer; Boundingbox = die statische, an der Kreiselachse ausgerichtete Default-Box. Rueckgabe: fortgeschriebene Elementnummer. """ schleus, warnings = compute_strecke_kreisel_schleus( items, durchmesser_mm=KREISEL_DURCHMESSER_MM, element_box_mm=ELEMENT_BOX_MM, dbg=dbg) # Warnungen (>2 Kreisel-BBox-Beruehrungen je Strecke) auf die Strecken-Zeile # schreiben (Spalte "Warnungen") und in die Konsole ausgeben. if warnings: strecke_by_id = {it.get("teileid", ""): it for it in items if it.get("teileart", "") in GF_VF_TEILEARTEN} for sid, text in warnings.items(): if sid in strecke_by_id: strecke_by_id[sid]["warnung"] = text print(f"[export_csv] WARNUNG: Strecke {sid} {text}") naechste_id = max_numeric_teileid(items) elem_nr = start_elem_nr zaehler = {} for sp in schleus: elem_nr += 1 naechste_id += 1 teileid = f"{naechste_id:04d}" teileart = sp["teileart"] zaehler[teileart] = zaehler.get(teileart, 0) + 1 planquadrat = (compute_planquadrat(sp["x"], sp["y"], planquadrat_cfg) if planquadrat_cfg is not None else "") items.append({ "nr": elem_nr, "teileart": teileart, "teileid": teileid, "bezeichnung": f"{teileart} :{zaehler[teileart]}", "planquadrat": planquadrat, "position": f'{sp["x"]:.2f}, {sp["y"]:.2f}, {sp["z"]:.2f}', "boundingbox": f'{sp["dx"]:.2f}, {sp["dy"]:.2f}, {sp["dz"]:.2f}', "insertpoint": "", "k1": "", "k2": "", "k3": "", "k4": "", "nachbarn": ", ".join([sp["kreisel_id"], sp["strecke_id"]]), "fehler": "", "merkmale": { "Erzeugt aus": "Kreisel-Strecke-Kollision", "Kreisel": sp["kreisel_id"], "Strecke": sp["strecke_id"], "Streckenende": sp["ende_label"], }, }) return elem_nr 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 btmt_beladung_count = 0 btmt_entladung_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), # x/y (Block-Einfuegepunkt = Kettenanfang) fuer die Anfang/Ende- # Bestimmung der Ein-/Ausschleuselemente (compute_strecke_kreisel_schleus) "x": block.get("x", 0.0), "y": block.get("y", 0.0), **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), # x/y (Block-Einfuegepunkt = Kettenanfang) fuer die Anfang/Ende- # Bestimmung der Ein-/Ausschleuselemente (compute_strecke_kreisel_schleus) "x": block.get("x", 0.0), "y": block.get("y", 0.0), **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), # 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")), **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), # "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 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 # Nach der Kollisions-/Nachbarschaftserkennung: an jedem Kreisel- # Beruehrpunkt eine ILS-Weiche anhaengen (eigene Zeilen am Listenende - # sie nehmen an der Nachbarschaftserkennung selbst nicht teil, ihre # Nachbarn sind fest die beiden beruehrenden Kreisel). Danach an jedem # Strecken-Ende mit anschliessendem Kreisel ein Aus-/Einschleuselement - # die TeileId-Zaehlung laeuft ueber beide hinweg weiter (Weichen zuerst). elem_nr = build_kreisel_weiche_items(items, planquadrat_cfg, elem_nr, dbg=dbg) build_strecke_schleus_items(items, planquadrat_cfg, elem_nr, 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["fehler"])}"' f';"{csv_quote(item.get("warnung", ""))}"' 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;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()