Files
dxfmakros/lib/export_csv.py
T

446 lines
15 KiB
Python

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
export_csv.py - Erzeugt eine einfache Item-Liste als CSV.
Fuer Omniflo-Elemente (Bogen/Weiche/Gerade) wird am Ende zusaetzlich eine
"Omniflo Sum"-Zeile mit Anzahl Boegen/Weichentypen und Gesamtlaenge AP110/AP60
angehaengt (siehe build_omni_sum_merkmale).
Aufruf:
python export_csv.py <export_raw.json> <data_dir> <output.csv>
CSV-Format:
Elementnummer;TeileArt;TeileId;NachbarIds;Bezeichnung;Planquadrat;rotation;Merkmale
"""
import json
import sys
import os
from export_blockpatterns import load_patterns, matches_any
BLOCKPATTERNS = load_patterns()
# ---------------------------------------------------------------------------
# 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 not None:
try:
return float(drehung_raw)
except (ValueError, TypeError):
pass
return 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):
attribs = block.get("attribs", {})
return {
"Typ": attribs.get("TYP", ""),
"Seite": attribs.get("SEITE", ""),
"Winkel": attribs.get("WINKEL", ""),
"DeltaH_mm": attribs.get("DELTA_H", ""),
"DeltaL_mm": attribs.get("DELTA_L", ""),
"L_VF_m": attribs.get("L_VF_m", ""),
"L_GF_m": attribs.get("L_GF_m", ""),
"Montagehoehe_m": attribs.get("MONTAGEHOEHE_m", ""),
"Hoehe_Von_mm": attribs.get("HOEHE_VON", ""),
"Hoehe_Bis_mm": attribs.get("HOEHE_BIS", ""),
"Antriebfahrtrichtung": attribs.get("ANTRIEBFAHRTRICHTUNG", ""),
"Anzahl_Separator": attribs.get("ANZAHL_SEPARATOR", ""),
"Anzahl_Scanner": attribs.get("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")
try:
return float(laenge)
except (TypeError, ValueError):
return 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": ""
}
# ---------------------------------------------------------------------------
# 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_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"),
"Höhe in m": hoehe_m,
"Drehrichtung": attribs.get("DREHRICHTUNG", "UZS"),
"Drehung": block.get("rotation", 0.0),
"Name": attribs.get("NAME", ""),
}
# ---------------------------------------------------------------------------
# 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):
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
# 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}",
"merkmale": build_bogen_merkmale(block, eintrag),
})
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]}",
"merkmale": build_weiche_merkmale(block, eintrag),
})
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}",
"merkmale": build_gerade_merkmale(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": "VarioFoerderer",
"teileid": block.get("attribs", {}).get("ID", "0000"),
"bezeichnung": f"VarioFoerderer :{vf_count}",
"merkmale": build_variofoerderer_merkmale(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": "Gefaellestrecke",
"teileid": block.get("attribs", {}).get("ID", "0000"),
"bezeichnung": f"Gefaellestrecke :{gf_count}",
"merkmale": build_variofoerderer_merkmale(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}",
"merkmale": build_kreisel_merkmale(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}",
"merkmale": build_kreisel_merkmale(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}",
"merkmale": {"Blockname": bname},
})
continue
if has_omniflo:
elem_nr += 1
items.append({
"nr": elem_nr,
"teileart": "Omniflo Sum",
"teileid": "autogenerated_of_json",
"bezeichnung": "Omniflo sum",
"merkmale": build_omni_sum_merkmale(
bogen_count, cnt_wk, cnt_einzel, cnt_delta, cnt_doppel, cnt_stern,
len_ap110_mm, len_ap60_mm),
})
return items
# ---------------------------------------------------------------------------
# CSV-Formatierung
# ---------------------------------------------------------------------------
def format_csv_line(item):
merkmale_json = json.dumps(item["merkmale"], ensure_ascii=False)
return (
f'{item["nr"]}'
f';"{item["teileart"]}"'
f';"{item["teileid"]}"'
f';"{item["bezeichnung"]}"'
f';1'
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.")
items = process_blocks(blocks, lookup)
header = "Elementnummer;TeileArt;TeileId;Bezeichnung;Anzahl;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()