Schoenenberger_Systeme_GmbH/dxfmakros
3
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Sivas Export Format geändert

Browse Source
This commit is contained in:
Michael Stangl
2026-06-29 17:08:37 +02:00
parent 3a3d22d103
commit 4863df7bac
4 changed files with 211 additions and 225 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/export_sivas.py
+191 -224
View File
@@ -1,10 +1,18 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
export_csv.py - Erzeugt CSV aus den von LISP gesammelten Block-Daten.
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_csv.py <export_raw.json> <data_dir> <output.csv>
python export_sivas.py <export_raw.json> <data_dir> <output.csv>
Argumente:
export_raw.json - JSON-Datei mit allen INSERT-Entities aus der Zeichnung
@@ -26,94 +34,49 @@ def build_lookup(boegen, weichen):
"""Erzeugt ein Dict: SivasNr (als String) -> (typ, eintrag)"""
lookup = {}
for b in boegen:
key = str(b["Sivasnr"])
lookup[key] = ("bogen", b)
lookup[str(b["Sivasnr"])] = ("bogen", b)
for w in weichen:
key = str(w["Sivasnr"])
lookup[key] = ("weiche", w)
lookup[str(w["Sivasnr"])] = ("weiche", w)
return lookup
def classify_weiche(eintrag):
"""Bestimmt den Weichen-Subtyp fuer die Omniflo Sum Zaehlung."""
"""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 == "Doppelweiche":
if wt in ("Doppelweiche", "Dreiwegeweiche"):
return "doppelweiche"
if wt == "Dreiwegeweiche":
return "dreiwegeweiche"
if wt == "Deltaweiche" or "DELTA" in profil.upper():
return "deltaweiche"
if wt == "Dreifachweiche":
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(block):
def get_hoehe_mm(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"
v = attribs.get("HOEHE") or str(int(block.get("z", 0) or 0)) or "2000"
return v
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 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_bogen_merkmale(block, eintrag):
"""Merkmale-JSON fuer einen Omniflo Bogen."""
return {
"Kurvenwinkel": eintrag.get("KurvenWinkel", 0),
"Radius": eintrag.get("Radius", 0),
"Höhe": get_hoehe(block),
"Drehung": get_drehung(block),
"SivasNummer": str(eintrag.get("Sivasnr", ""))
}
def build_weiche_merkmale(block, eintrag):
"""Merkmale-JSON fuer eine Omniflo Weiche."""
wt = eintrag.get("WeichenTyp", "Einzelweiche")
return {
"Weichentyp": wt,
"Richtung": str(eintrag.get("KurvenRichtung", "")),
"Weichenwinkel": eintrag.get("KurvenWinkel", 0),
"Höhe": get_hoehe(block),
"Drehung": get_drehung(block),
"Antrieb Kurve": eintrag.get("SivasnrTEF") is not None,
"SivasNummer": str(eintrag.get("Sivasnr", ""))
}
def build_gerade_merkmale(block):
"""Merkmale-JSON fuer eine Omniflo Gerade (AP110)."""
return {
"Anzahl der Separatoren": "0",
"Länge in Meter": "2",
"Winkel": "0",
"Anzahl der Scanner": 0,
"Höhe oben": "2000",
"Höhe unten": "2000",
"Drehung": block.get("rotation", 0.0),
"SivasNummer": ""
}
def build_kreisel_merkmale(block):
"""Merkmale-JSON fuer einen ILS Kreisel (liest Attribute aus Block)."""
def build_kreisel_details(block):
"""Merkmale-Dict fuer einen ILS Kreisel."""
attribs = block.get("attribs", {})
abstand_mm = attribs.get("ABSTAND", "2300")
try:
@@ -131,15 +94,14 @@ def build_kreisel_merkmale(block):
"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,
"Hoehe in m": hoehe_m,
"Drehrichtung": attribs.get("DREHRICHTUNG", "UZS"),
"Drehung": block.get("rotation", 0.0),
"Name": attribs.get("NAME", ""),
}
# --- Bekannte Blocknamen fuer nicht-Omniflo Elemente ---
## Kreisel-Bloecke werden ueber Praefix "KR_" erkannt (siehe process_blocks)
# --- Bekannte Bloecke ignorieren ---
SKIP_BLOCKS = {"K1", "K2", "K3", "K4", "KS_EIN", "KS_AUS"}
VARIO_BLOCKS = {
"AUS_Element_links", "EIN_Element_links",
@@ -147,19 +109,23 @@ VARIO_BLOCKS = {
"Vario_Spannstation_SP_500mm", "Vario_Motorstation_SP_500mm",
}
# Koordinatensystem-Bloecke (ignorieren)
SKIP_BLOCKS = {"K1", "K2", "K3", "K4", "KS_EIN", "KS_AUS"}
def process_blocks(blocks, lookup):
"""
Verarbeitet alle Bloecke und erzeugt:
- items: Liste der CSV-Einzelzeilen
- counters: Zaehler fuer Summierungszeilen
Verarbeitet alle Bloecke.
Rueckgabe:
items - Liste der CSV-Zeilen als Dicts
counters - Zaehler fuer ILS-Automation-Summierungszeile
"""
items = []
# Gruppen: sivasnr_str -> {anzahl, hoehe_mm}
bogen_groups = {} # sivasnr -> dict
weiche_groups = {} # sivasnr -> dict
# Gerade: artinr_str -> {anzahl, laenge_mm_total, hoehe_mm}
gerade_groups = {}
kreisel_list = [] # Liste von blocks
counters = {
# Omniflo Sum
"anzahl_boegen": 0,
"anzahl_weichenkoerper": 0,
"anzahl_einzelweichen": 0,
@@ -168,7 +134,6 @@ def process_blocks(blocks, lookup):
"anzahl_sternweichen": 0,
"gesamtlaenge_ap110": 0.0,
"gesamtlaenge_ap60": 0.0,
# ILS Automation
"anzahl_kreisel": 0,
"anzahl_kreisel_pinband": 0,
"sep_kreisel_pin": 0,
@@ -191,54 +156,32 @@ def process_blocks(blocks, lookup):
"anzahl_abwurfstationen": 0,
}
elem_nr = 0
bogen_count = 0
weiche_count = {}
gerade_count = 0
kreisel_count = 0
for block in blocks:
bname = block.get("block_name", "")
# Koordinatensystem-Bloecke ueberspringen
if bname in SKIP_BLOCKS:
continue
# Omniflo Bogen oder Weiche?
# --- Omniflo Bogen oder Weiche (erkannt via Katalog) ---
if bname in lookup:
typ, eintrag = lookup[bname]
elem_nr += 1
shape_id = block.get("attribs", {}).get("ID", "0000")
sivasnr = str(eintrag.get("Sivasnr", bname))
hoehe = get_hoehe_mm(block)
if typ == "bogen":
bogen_count += 1
items.append({
"nr": elem_nr,
"teileart": "Omniflo Kurve",
"teileid": shape_id,
"bezeichnung": f"OFBogen :{bogen_count}",
"anzahl": 1,
"merkmale": build_bogen_merkmale(block, eintrag)
})
if sivasnr not in bogen_groups:
bogen_groups[sivasnr] = {"anzahl": 0, "hoehe_mm": hoehe}
bogen_groups[sivasnr]["anzahl"] += 1
counters["anzahl_boegen"] += 1
elif typ == "weiche":
if sivasnr not in weiche_groups:
weiche_groups[sivasnr] = {"anzahl": 0, "hoehe_mm": hoehe}
weiche_groups[sivasnr]["anzahl"] += 1
subtyp = classify_weiche(eintrag)
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]}",
"anzahl": 1,
"merkmale": build_weiche_merkmale(block, eintrag)
})
if subtyp == "einzelweiche":
counters["anzahl_einzelweichen"] += 1
elif subtyp == "doppelweiche" or subtyp == "dreiwegeweiche":
elif subtyp in ("doppelweiche", "dreiwegeweiche"):
counters["anzahl_doppel_dreiwegeweichen"] += 1
elif subtyp == "deltaweiche":
counters["anzahl_deltaweichen"] += 1
@@ -246,36 +189,33 @@ def process_blocks(blocks, lookup):
counters["anzahl_sternweichen"] += 1
elif subtyp == "weichenkoerper":
counters["anzahl_weichenkoerper"] += 1
continue
# Omniflo Gerade (AP110)
# --- Omniflo Gerade (AP110) ---
if "AP110" in bname.upper() or "AP_110" in bname.upper():
elem_nr += 1
gerade_count += 1
items.append({
"nr": elem_nr,
"teileart": "Omniflo Gerade",
"teileid": block.get("attribs", {}).get("ID", "0000"),
"bezeichnung": f"OFGerade :{gerade_count}",
"anzahl": 1,
"merkmale": build_gerade_merkmale(block)
})
counters["gesamtlaenge_ap110"] += 2000.0
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
}
gerade_groups[key]["anzahl"] += 1
gerade_groups[key]["laenge_mm"] += laenge
counters["gesamtlaenge_ap110"] += laenge
continue
# ILS Kreisel (Compound-Bloecke mit Praefix KR_)
if bname.startswith("KR_"):
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}",
"anzahl": 1,
"merkmale": build_kreisel_merkmale(block)
})
# --- 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")
@@ -293,98 +233,136 @@ def process_blocks(blocks, lookup):
counters["anzahl_streckengruppen"] = 1
continue
# VarioFoerderer-Bloecke
is_vario = False
# --- VarioFoerderer ---
for vb in VARIO_BLOCKS:
if vb in bname:
is_vario = True
counters["anzahl_variofoerderer"] += 1
break
if is_vario:
counters["anzahl_variofoerderer"] += 1
continue
# --- 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": {},
})
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": {},
})
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": {},
})
nr += 1
for block in kreisel_list:
items.append({
"nr": nr,
"teileart": "ILS 2.0 Kreisel",
"sivasnr": "6200",
"sivasnr_quoted": True,
"dispgruppe": 20,
"anzahl": 1,
"laenge_mm": "",
"hoehe_m": "",
"details": build_kreisel_details(block),
})
nr += 1
# ILS Automation Summierungszeile (nur wenn ILS-Elemente vorhanden)
if counters["anzahl_kreisel"] > 0 or counters["anzahl_variofoerderer"] > 0:
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"],
}
items.append({
"nr": nr,
"teileart": "ILS 2.0 Automation",
"sivasnr": "6270",
"sivasnr_quoted": True,
"dispgruppe": 20,
"anzahl": 1,
"laenge_mm": "",
"hoehe_m": "",
"details": automation_details,
})
return items, counters
def build_ils_automation_row(elem_nr, counters):
"""Erzeugt die ILS 2.0 Automation Summierungszeile."""
merkmale = {
"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"],
"Längen aller Kreisel": counters["laengen_kreisel"],
"Anzahl der Weichen": counters["anzahl_weichen_ils"],
"Anzahl der Eckräder": counters["anzahl_eckraeder"],
"Anzahl der einzelnen Gefällestrecken": counters["anzahl_gefaellestrecken"],
"Anzahl der einzelnen Varioförderer": counters["anzahl_variofoerderer"],
"Anzahl der Streckengruppen": counters["anzahl_streckengruppen"],
"Längen aller Gefällestrecken": counters["laengen_gefaellestrecken"],
"Anzahl der Motoren Varioförderer": counters["anzahl_motoren_vario"],
"Längen aller Varioförderer": counters["laengen_variofoerderer"],
"Anzahl ETFs": counters["anzahl_etf"],
"Längen 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"],
}
return {
"nr": elem_nr,
"teileart": "ILS 2.0 Automation",
"teileid": "autogenerated_ils_automation_json",
"bezeichnung": "Automation ILS structgen",
"anzahl": 1,
"merkmale": merkmale,
}
def build_omniflo_sum_row(elem_nr, counters):
"""Erzeugt die Omniflo Sum Summierungszeile."""
merkmale = {
"Anzahl Bögen": counters["anzahl_boegen"],
"Anzahl Weichengrundkörper": counters["anzahl_weichenkoerper"],
"Anzahl Einwegweichen": counters["anzahl_einzelweichen"],
"Anzahl Deltaweichen": counters["anzahl_deltaweichen"],
"Anzahl Doppelweichen und Dreiwegeweichen": counters["anzahl_doppel_dreiwegeweichen"],
"Anzahl Sternweichen": counters["anzahl_sternweichen"],
"Gesamtlänge AP110": counters["gesamtlaenge_ap110"],
"Gesamtlänge AP60": counters["gesamtlaenge_ap60"],
}
return {
"nr": elem_nr,
"teileart": "Omniflo Sum",
"teileid": "autogenerated_of_json",
"bezeichnung": "Omniflo sum",
"anzahl": 1,
"merkmale": merkmale,
}
def format_csv_line(item):
"""Formatiert eine CSV-Zeile im geforderten Format."""
merkmale_json = json.dumps(item["merkmale"], ensure_ascii=False)
return '{nr};"{teileart}";"{teileid}";"{bezeichnung}";{anzahl};{merkmale}'.format(
nr=item["nr"],
teileart=item["teileart"],
teileid=item["teileid"],
bezeichnung=item["bezeichnung"],
anzahl=item["anzahl"],
merkmale=merkmale_json,
"""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_csv.py <export_raw.json> <data_dir> <output.csv>")
print("Aufruf: python export_sivas.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]
# Daten laden
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")
@@ -394,29 +372,18 @@ def main():
lookup = build_lookup(boegen, weichen)
print(f"[export_csv] {len(blocks)} Bloecke geladen, "
print(f"[export_sivas] {len(blocks)} Bloecke geladen, "
f"{len(boegen)} Boegen, {len(weichen)} Weichen im Katalog.")
# Bloecke verarbeiten
items, counters = process_blocks(blocks, lookup)
items, _ = process_blocks(blocks, lookup)
# Summierungszeilen anhaengen
next_nr = (items[-1]["nr"] + 1) if items else 1
ils_row = build_ils_automation_row(next_nr, counters)
items.append(ils_row)
of_row = build_omniflo_sum_row(next_nr + 1, counters)
items.append(of_row)
# CSV schreiben
header = "Elementnummer;TeileArt;TeileId;Bezeichnung;Anzahl;Merkmale"
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_csv] CSV geschrieben: {output_csv} ({len(items)} Zeilen)")
print(f"[export_sivas] CSV geschrieben: {output_csv} ({len(items)} Zeilen)")
if __name__ == "__main__":