#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ export_sivas.py - Erzeugt Sivas-Export-CSV mit gruppierten Zeilen und Summierungszeilen. Neues CSV-Format: Nr;TeileArt;SivasNummer;Dispgruppe;Anzahl;Laenge[mm];Hoehe[m];details - Omniflo Bogen/Weiche: gruppiert nach SivasNummer, Anzahl summiert - Omniflo Gerade: gruppiert nach ARTINR-Attribut, Laenge[mm] summiert - ILS 2.0 Kreisel: je Kreisel eine Zeile, details = JSON mit Parametern - ILS 2.0 Automation: Summierungszeile (nur wenn ILS-Elemente vorhanden) Aufruf: python export_sivas.py Argumente: export_raw.json - JSON-Datei mit allen INSERT-Entities aus der Zeichnung data_dir - Pfad zum data/-Verzeichnis (mit json/ Unterordner) output.csv - Ziel-CSV-Datei """ import json import sys import os def load_json(path): with open(path, "r", encoding="utf-8") 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 classify_weiche(eintrag): """Bestimmt den Weichen-Subtyp fuer die ILS-Automation-Zaehlung.""" wt = eintrag.get("WeichenTyp", "") profil = eintrag.get("ProfilTyp", "") if "WEICHENKOERPER" in profil.upper(): return "weichenkoerper" if wt == "Einzelweiche": return "einzelweiche" if wt in ("Doppelweiche", "Dreiwegeweiche"): return "doppelweiche" if wt in ("Deltaweiche", "Dreifachweiche") or "DELTA" in profil.upper(): return "deltaweiche" if wt == "Sternweiche" or "STAR" in profil.upper(): return "sternweiche" return "einzelweiche" def get_hoehe_mm(block): """Liest HOEHE aus Block-Attributen. Fallback: Z-Koordinate.""" attribs = block.get("attribs", {}) v = attribs.get("HOEHE") or str(int(block.get("z", 0) or 0)) or "2000" return v def get_id(block): """Liest das eindeutige ID-Attribut eines Blocks (von ssg-id-check-all vergeben).""" return block.get("attribs", {}).get("ID", "") def format_hoehe_m(hoehe_mm_str): """Konvertiert mm-String nach m mit deutschem Dezimaltrennzeichen.""" try: m = float(hoehe_mm_str) / 1000.0 if m == int(m): return str(int(m)) return f"{m:.1f}".replace(".", ",") except (ValueError, TypeError): return "" def build_kreisel_details(block): """Merkmale-Dict fuer einen ILS Kreisel.""" attribs = block.get("attribs", {}) abstand_mm = attribs.get("ABSTAND", "2300") try: abstand_m = str(round(float(abstand_mm) / 1000.0, 2)) except (ValueError, TypeError): abstand_m = "2.3" hoehe_mm = attribs.get("HOEHE", "0") try: hoehe_m = str(round(float(hoehe_mm) / 1000.0, 2)) except (ValueError, TypeError): hoehe_m = "0" return { "Abstand (Kreiselachse A - Kreiselachse) in Meter": abstand_m, "Anzahl der Separatoren": attribs.get("N_SEPARATOREN", "2"), "Kreiselart": attribs.get("KREISELART", "STANDARD"), "Anzahl der Scanner": attribs.get("N_SCANNER", "0"), "Anzahl der Rampen": attribs.get("N_RAMPEN", "0"), "Hoehe in m": hoehe_m, "Drehrichtung": attribs.get("DREHRICHTUNG", "UZS"), "Drehung": block.get("rotation", 0.0), } # --- Bekannte Bloecke ignorieren --- SKIP_BLOCKS = {"K1", "K2", "K3", "K4", "KS_EIN", "KS_AUS"} # Sivas-Nummern und Bezeichnungen je ILS-Elementtyp ILS_SIVASNR = { "kreisel": ("6200", "ILS 2.0 Kreisel"), "eckrad": ("6210", "ILS 2.0 Eckrad"), "zusatzmodul":("6215", "ILS 2.0 Zusatzmodul"), "ils_weiche": ("6216", "ILS 2.0 Weichen"), "strecke": ("6220", "ILS 2.0 Strecke - Module"), "automation": ("6269", "ILS 2.0 Automation und Pneumatik - Gesamtanlage"), } # Strecken-Module: Blocknamen-Fragmente fuer Vario/Staustrecke-Elemente STRECKE_BLOCKS = { "AUS_Element_links", "EIN_Element_links", "Staustrecke_SP_1000_mm", "Staustrecke_Separator_SP_300_mm", "Vario_Spannstation_SP_500mm", "Vario_Motorstation_SP_500mm", } def process_blocks(blocks, lookup): """ Verarbeitet alle Bloecke. Rueckgabe: items - Liste der CSV-Zeilen als Dicts counters - Zaehler fuer ILS-Automation-Summierungszeile """ # Gruppen: sivasnr_str -> {anzahl, hoehe_mm, ids} bogen_groups = {} # sivasnr -> dict weiche_groups = {} # sivasnr -> dict # Gerade: artinr_str -> {anzahl, laenge_mm_total, hoehe_mm, ids} gerade_groups = {} kreisel_list = [] # Liste von blocks eckrad_ids = [] # IDs aller Eckrad-Bloecke strecke_ids = [] # IDs aller Strecke-Module (VF_*, GF_*, Vario/Staustrecke) counters = { "anzahl_boegen": 0, "anzahl_weichenkoerper": 0, "anzahl_einzelweichen": 0, "anzahl_deltaweichen": 0, "anzahl_doppel_dreiwegeweichen": 0, "anzahl_sternweichen": 0, "gesamtlaenge_ap110": 0.0, "gesamtlaenge_ap60": 0.0, "anzahl_kreisel": 0, "anzahl_kreisel_pinband": 0, "sep_kreisel_pin": 0, "sep_kreisel_ohne_pin": 0, "anzahl_verbindungskreisel": 0, "laengen_kreisel": 0.0, "anzahl_weichen_ils": 0, "anzahl_eckraeder": 0, "anzahl_gefaellestrecken": 0, "anzahl_variofoerderer": 0, "anzahl_streckengruppen": 0, "laengen_gefaellestrecken": 0.0, "anzahl_motoren_vario": 0, "laengen_variofoerderer": 0.0, "anzahl_etf": 0, "laengen_etf": 0.0, "anzahl_separatoren": 0, "anzahl_scanner": 0, "anzahl_btbeladung": 0, "anzahl_abwurfstationen": 0, } for block in blocks: bname = block.get("block_name", "") if bname in SKIP_BLOCKS: continue # --- Omniflo Bogen oder Weiche (erkannt via Katalog) --- if bname in lookup: typ, eintrag = lookup[bname] sivasnr = str(eintrag.get("Sivasnr", bname)) hoehe = get_hoehe_mm(block) if typ == "bogen": if sivasnr not in bogen_groups: bogen_groups[sivasnr] = {"anzahl": 0, "hoehe_mm": hoehe, "ids": []} bogen_groups[sivasnr]["anzahl"] += 1 bogen_groups[sivasnr]["ids"].append(get_id(block)) counters["anzahl_boegen"] += 1 elif typ == "weiche": if sivasnr not in weiche_groups: weiche_groups[sivasnr] = {"anzahl": 0, "hoehe_mm": hoehe, "ids": []} weiche_groups[sivasnr]["anzahl"] += 1 weiche_groups[sivasnr]["ids"].append(get_id(block)) subtyp = classify_weiche(eintrag) if subtyp == "einzelweiche": counters["anzahl_einzelweichen"] += 1 elif subtyp in ("doppelweiche", "dreiwegeweiche"): counters["anzahl_doppel_dreiwegeweichen"] += 1 elif subtyp == "deltaweiche": counters["anzahl_deltaweichen"] += 1 elif subtyp == "sternweiche": counters["anzahl_sternweichen"] += 1 elif subtyp == "weichenkoerper": counters["anzahl_weichenkoerper"] += 1 continue # --- Omniflo Gerade (AP110) --- if "AP110" in bname.upper() or "AP_110" in bname.upper(): attribs = block.get("attribs", {}) artinr = attribs.get("ARTINR", "") key = artinr if artinr else "AP110" hoehe = get_hoehe_mm(block) laenge = 2000.0 laenge_attr = attribs.get("LAENGE") or attribs.get("A") if laenge_attr: try: laenge = float(laenge_attr) except (ValueError, TypeError): pass if key not in gerade_groups: gerade_groups[key] = { "anzahl": 0, "laenge_mm": 0.0, "hoehe_mm": hoehe, "artinr": artinr, "ids": [], } gerade_groups[key]["anzahl"] += 1 gerade_groups[key]["laenge_mm"] += laenge gerade_groups[key]["ids"].append(get_id(block)) counters["gesamtlaenge_ap110"] += laenge continue # --- ILS Kreisel --- if bname.startswith("KR_") or bname.startswith("KREISEL_"): kreisel_list.append(block) attribs = block.get("attribs", {}) counters["anzahl_kreisel"] += 1 n_sep = int(attribs.get("N_SEPARATOREN", "2") or "2") kreiselart = attribs.get("KREISELART", "STANDARD") if kreiselart == "PIN": counters["anzahl_kreisel_pinband"] += 1 counters["sep_kreisel_pin"] += n_sep else: counters["sep_kreisel_ohne_pin"] += n_sep abstand_mm = float(attribs.get("ABSTAND", "2300") or "2300") counters["laengen_kreisel"] += abstand_mm / 1000.0 counters["anzahl_separatoren"] += n_sep counters["anzahl_scanner"] += int(attribs.get("N_SCANNER", "0") or "0") if counters["anzahl_streckengruppen"] == 0: counters["anzahl_streckengruppen"] = 1 continue # --- ILS Eckrad --- if bname.startswith("ECKRAD_"): counters["anzahl_eckraeder"] += 1 eckrad_ids.append(get_id(block)) continue # --- VarioFoerderer Compound-Block (VF_N) --- if bname.startswith("VF_"): attribs = block.get("attribs", {}) counters["anzahl_variofoerderer"] += 1 strecke_ids.append(get_id(block)) try: counters["laengen_variofoerderer"] += float(attribs.get("L_VF_m", "0") or "0") except (ValueError, TypeError): pass try: counters["anzahl_separatoren"] += int(attribs.get("ANZAHL_SEPARATOR", "0") or "0") except (ValueError, TypeError): pass try: counters["anzahl_scanner"] += int(attribs.get("ANZAHL_SCANNER", "0") or "0") except (ValueError, TypeError): pass if counters["anzahl_streckengruppen"] == 0: counters["anzahl_streckengruppen"] = 1 continue # --- Gefaellestrecke Compound-Block (GF_N) --- if bname.startswith("GF_"): attribs = block.get("attribs", {}) counters["anzahl_variofoerderer"] += 1 counters["anzahl_gefaellestrecken"] += 1 strecke_ids.append(get_id(block)) try: counters["laengen_gefaellestrecken"] += float(attribs.get("L_GF_m", "0") or "0") except (ValueError, TypeError): pass try: counters["anzahl_separatoren"] += int(attribs.get("ANZAHL_SEPARATOR", "0") or "0") except (ValueError, TypeError): pass try: counters["anzahl_scanner"] += int(attribs.get("ANZAHL_SCANNER", "0") or "0") except (ValueError, TypeError): pass if counters["anzahl_streckengruppen"] == 0: counters["anzahl_streckengruppen"] = 1 continue # --- ILS Strecke-Module (Vario/Staustrecke, einzelne Komponenten in Modelspace) --- for sb in STRECKE_BLOCKS: if sb in bname: counters["anzahl_variofoerderer"] += 1 strecke_ids.append(get_id(block)) break # --- Zeilen aufbauen --- items = [] nr = 1 for sivasnr, g in bogen_groups.items(): items.append({ "nr": nr, "teileart": "Omniflo Bogen", "sivasnr": sivasnr, "sivasnr_quoted": False, "dispgruppe": 10, "anzahl": g["anzahl"], "laenge_mm": "", "hoehe_m": format_hoehe_m(g["hoehe_mm"]), "details": {"IDs": g["ids"]}, }) nr += 1 for sivasnr, g in weiche_groups.items(): items.append({ "nr": nr, "teileart": "Omniflo Weiche", "sivasnr": sivasnr, "sivasnr_quoted": False, "dispgruppe": 10, "anzahl": g["anzahl"], "laenge_mm": "", "hoehe_m": format_hoehe_m(g["hoehe_mm"]), "details": {"IDs": g["ids"]}, }) nr += 1 for key, g in gerade_groups.items(): items.append({ "nr": nr, "teileart": "Omniflo Gerade", "sivasnr": g["artinr"], "sivasnr_quoted": True, "dispgruppe": 10, "anzahl": g["anzahl"], "laenge_mm": int(g["laenge_mm"]), "hoehe_m": format_hoehe_m(g["hoehe_mm"]), "details": {"IDs": g["ids"]}, }) nr += 1 for block in kreisel_list: sivasnr, teileart = ILS_SIVASNR["kreisel"] details = build_kreisel_details(block) details["ID"] = get_id(block) items.append({ "nr": nr, "teileart": teileart, "sivasnr": sivasnr, "sivasnr_quoted": True, "dispgruppe": 20, "anzahl": 1, "laenge_mm": "", "hoehe_m": "", "details": details, }) nr += 1 if counters["anzahl_eckraeder"] > 0: sivasnr, teileart = ILS_SIVASNR["eckrad"] items.append({ "nr": nr, "teileart": teileart, "sivasnr": sivasnr, "sivasnr_quoted": True, "dispgruppe": 20, "anzahl": counters["anzahl_eckraeder"], "laenge_mm": "", "hoehe_m": "", "details": {"IDs": eckrad_ids}, }) nr += 1 if counters["anzahl_variofoerderer"] > 0: sivasnr, teileart = ILS_SIVASNR["strecke"] items.append({ "nr": nr, "teileart": teileart, "sivasnr": sivasnr, "sivasnr_quoted": True, "dispgruppe": 20, "anzahl": counters["anzahl_variofoerderer"], "laenge_mm": "", "hoehe_m": "", "details": {"IDs": strecke_ids}, }) nr += 1 # ILS Automation Summierungszeile (nur wenn ILS-Elemente vorhanden) has_ils = (counters["anzahl_kreisel"] > 0 or counters["anzahl_eckraeder"] > 0 or counters["anzahl_variofoerderer"] > 0) if has_ils: automation_details = { "Anzahl Kreisel": counters["anzahl_kreisel"], "Anzahl Kreisel mit Pinband": counters["anzahl_kreisel_pinband"], "Anzahl aller Separatoren in Kreisel mit PIN": counters["sep_kreisel_pin"], "Anzahl aller Separatoren in Kreisel ohne PIN": counters["sep_kreisel_ohne_pin"], "Anzahl Verbindungskreisel": counters["anzahl_verbindungskreisel"], "Laengen aller Kreisel": counters["laengen_kreisel"], "Anzahl der Weichen": counters["anzahl_weichen_ils"], "Anzahl der Eckraeder": counters["anzahl_eckraeder"], "Anzahl der einzelnen Gefaellestrecken": counters["anzahl_gefaellestrecken"], "Anzahl der einzelnen Variofoerderer": counters["anzahl_variofoerderer"], "Anzahl der Streckengruppen": counters["anzahl_streckengruppen"], "Laengen aller Gefaellestrecken": counters["laengen_gefaellestrecken"], "Anzahl der Motoren Variofoerderer": counters["anzahl_motoren_vario"], "Laengen aller Variofoerderer": counters["laengen_variofoerderer"], "Anzahl ETFs": counters["anzahl_etf"], "Laengen aller ETFs": counters["laengen_etf"], "Anzahl aller Separatoren": counters["anzahl_separatoren"], "Anzahl aller Scanner": counters["anzahl_scanner"], "Anzahl aller BTBeladung": counters["anzahl_btbeladung"], "Anzahl aller Abwurfstationen": counters["anzahl_abwurfstationen"], } auto_sivasnr, auto_teileart = ILS_SIVASNR["automation"] items.append({ "nr": nr, "teileart": auto_teileart, "sivasnr": auto_sivasnr, "sivasnr_quoted": True, "dispgruppe": 20, "anzahl": 1, "laenge_mm": "", "hoehe_m": "", "details": automation_details, }) return items, counters def format_csv_line(item): """Formatiert eine CSV-Zeile im neuen Sivas-Format.""" details_json = json.dumps(item["details"], ensure_ascii=False) sivasnr = f'"{item["sivasnr"]}"' if item["sivasnr_quoted"] else item["sivasnr"] return ( f'{item["nr"]}' f';"{item["teileart"]}"' f';{sivasnr}' f';{item["dispgruppe"]}' f';{item["anzahl"]}' f';{item["laenge_mm"]}' f';{item["hoehe_m"]}' f';{details_json}' ) def main(): if len(sys.argv) < 4: print("Aufruf: python export_sivas.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_sivas] {len(blocks)} Bloecke geladen, " f"{len(boegen)} Boegen, {len(weichen)} Weichen im Katalog.") items, _ = process_blocks(blocks, lookup) header = "Nr;TeileArt;SivasNummer;Dispgruppe;Anzahl;Laenge[mm];Hoehe[m];details" 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_sivas] CSV geschrieben: {output_csv} ({len(items)} Zeilen)") if __name__ == "__main__": main()