From ef3d8170feb746e3f7ae5931a9fd49fc22c73e05 Mon Sep 17 00:00:00 2001 From: Samer Ayadi Date: Thu, 24 Sep 2026 15:40:43 +0200 Subject: [PATCH] TRO-Fluss: sepliste (Roh-JSON) als Gegenprobe fuer die Separatorenzahl je Host Optional: findet tro_flow.py neben der CSV ein _raw.json (dxfmakros- Rohattribute), vergleicht es je Strecke/Gefaellestrecke-Host die aus der CSV abgeleitete Separatorenzahl mit der Anzahl 'SEP'-Eintraege in dessen eigener 'sepliste' und meldet Abweichungen als Finding - ohne TRO-Daten zu aendern, die CSV bleibt die massgebliche Quelle. Faengt an HundM05 sofort einen echten Fall: VF_5 hat laut CSV 3 zugeordnete Separatoren, sepliste nennt nur 2 - passt zum bereits gemeldeten Separator 0018 ohne TrackIds-Mitgliedschaft. Bewusst NICHT verwendet: die sepliste-Koordinaten (der jeweils erste SEP-Eintrag jedes VF-Blocks traegt durchgehend eine kaputte Position, ein Bug im DXF-Export) und das Attribut "ANZAHL_SEPARATOR" (weicht in beiden bisher geprueften Anlagen von TrackIds UND sepliste ab, vermutlich veraltet/ doppelt gezaehlt). Fehlt die Roh-JSON-Datei, entfaellt die Gegenprobe stillschweigend; ist sie vorhanden aber kaputt (z. B. unvollstaendig geschrieben), gibt es eine Warnung statt eines Abbruchs. Co-Authored-By: Claude Sonnet 5 --- lib/tro_flow.py | 93 +++++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 91 insertions(+), 2 deletions(-) diff --git a/lib/tro_flow.py b/lib/tro_flow.py index 0b716e4..3b592e3 100644 --- a/lib/tro_flow.py +++ b/lib/tro_flow.py @@ -81,6 +81,7 @@ from __future__ import annotations import argparse import csv +import json import math import sys from collections.abc import Callable @@ -441,6 +442,62 @@ def _incoming_drop( return drive, cur_h - prev_h +def find_sibling_raw_json(csv_path: Path) -> Path | None: + """_raw.json neben der CSV, falls vorhanden. + + Optionale Roh-Attributsammlung des DXF-Exports (dxfmakros): eine flache + Liste von Bloecken mit 'attribs' (u. a. 'ID') und bei Strecke/ + Gefaellestrecke zusaetzlich einer 'sepliste' (siehe load_sepliste_counts). + Kein Pflichtbestandteil des Exports - existiert die Datei nicht, liefert + diese Funktion None und die Gegenprobe entfaellt einfach. + """ + candidate = csv_path.with_name(f"{csv_path.stem.lower()}_raw.json") + return candidate if candidate.is_file() else None + + +def load_sepliste_counts(raw_json_path: Path, warnings: list[str]) -> dict[str, int]: + """Container-Id (Roh-JSON 'attribs.ID') -> Anzahl 'SEP'-Eintraege in dessen 'sepliste'. + + Reine Gegenprobe gegen die aus der CSV abgeleitete Separatorenzahl je + Host (siehe derive_tros) - NICHT zur Rekonstruktion fehlender Separatoren + gedacht. Zwei Merkmale des Roh-JSON erwiesen sich als unzuverlaessig und + werden deshalb bewusst NICHT verwendet (Review 2026-09-24, HundM05): + + - Der jeweils ERSTE 'SEP'-Eintrag jedes VF-Blocks traegt durchgehend eine + kaputte Position (x = -Block-x, y ein Fantasiewert) - ein Bug im + DXF-Export, keine echte Platzierung. Die sepliste-IDs stammen ausserdem + aus einem anderen Nummerierungslauf als die CSV und lassen sich nicht + darauf abbilden - nur die ANZAHL der 'SEP'-Eintraege ist verwertbar, + nicht ihre Position oder ihre Id. + - Das Attribut "ANZAHL_SEPARATOR" auf dem Block selbst weicht in beiden + bisher geprueften Anlagen sowohl von der TrackIds-Zaehlung als auch von + der sepliste-Zaehlung ab (z. B. HundM05 VF_1: TrackIds/sepliste sagen 4, + ANZAHL_SEPARATOR sagt 9) - vermutlich ein veraltetes/doppelt gezaehltes + Katalog-Attribut, keine verlaessliche Quelle. + + Bei fehlender oder kaputter Datei (z. B. ein noch unvollstaendig + geschriebenes *_raw.json) wird ein Warnhinweis angehaengt und ein leeres + dict geliefert - die Gegenprobe entfaellt dann einfach, ohne den Lauf + abzubrechen. + """ + try: + raw = json.loads(raw_json_path.read_text(encoding="utf-8")) + except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: + warnings.append( + f"{raw_json_path.name}: nicht lesbar/kein gueltiges JSON ({exc}) - " + f"sepliste-Gegenprobe uebersprungen" + ) + return {} + counts: dict[str, int] = {} + for item in raw if isinstance(raw, list) else []: + container_id = str((item.get("attribs") or {}).get("ID") or "") + sepliste = item.get("sepliste") + if not container_id or not sepliste: + continue + counts[container_id] = sum(1 for s in sepliste if s.get("typ") == "SEP") + return counts + + def derive_tros( graph: Graph, elements: list[Element], @@ -449,6 +506,7 @@ def derive_tros( *, resolve_host: Callable[[Element, list[str]], str | None] | None = None, count_exits: Callable[[list[str]], list[str]] | None = None, + sepliste_counts: dict[str, int] | None = None, ) -> list[Tro]: """TRO-Liste aus den Separatoren und ihren angeschlossenen Objekten bilden. @@ -458,6 +516,12 @@ def derive_tros( _switch_exits). tro_geometry.py::make_geometric_resolvers() liefert ein Paar, das beides stattdessen direkt aus den Bauteil-Weltkoordinaten sucht (STRtree) - siehe tro_flow.py --geometry. + + sepliste_counts (optional, siehe load_sepliste_counts): Container-Id -> + Anzahl 'SEP'-Eintraege aus dem Roh-JSON. Reine Gegenprobe - stimmt die aus + der CSV abgeleitete Separatorenzahl eines Strecke/Gefaellestrecke-Hosts + nicht mit dieser Zahl ueberein, wird ein Finding angehaengt; die TRO-Daten + selbst bleiben unangetastet (CSV bleibt die massgebliche Quelle). """ resolve_host = resolve_host or (lambda sep, findings: _default_resolve_host(sep, findings, graph)) count_exits = count_exits or (lambda hosts: _switch_exits(hosts, graph)) @@ -538,6 +602,24 @@ def derive_tros( found.extend(sep_scanners.get(sep, [])) return sorted(found) + # Sepliste-Gegenprobe (siehe load_sepliste_counts/Docstring oben): reine + # Beobachtung, veraendert weder sep_host noch die spaeter abgeleiteten + # TROs - die CSV bleibt die massgebliche Quelle. + if sepliste_counts: + for node_id, node in sorted(graph.nodes.items()): + if node.kind not in ("Strecke", "Gefaellestrecke"): + continue + expected = sepliste_counts.get(node_id) + if expected is None: + continue + actual = len(seps_on([node_id])) + if actual != expected: + findings.append( + f"{node.element.name} ({node_id}): {actual} Separator(en) aus " + f"der CSV zugeordnet, aber {expected} 'SEP'-Eintraege in " + f"sepliste (Roh-JSON) - Gegenprobe weicht ab, bitte pruefen" + ) + tros: list[Tro] = [] # 1) Schwerkraftlinien (Gefaellestrecke mit Separator) -> Speicherblock @@ -1276,6 +1358,7 @@ def analyse_tros( resolve_host: Callable[[Element, list[str]], str | None] | None = None, count_exits: Callable[[list[str]], list[str]] | None = None, method: str | None = None, + sepliste_counts: dict[str, int] | None = None, ) -> Analysis: """TROs ermitteln, verknuepfen, verorten und gegen die Typdefinition pruefen.""" analysis = Analysis() @@ -1283,7 +1366,8 @@ def analyse_tros( if method is not None: analysis.method = method analysis.tros = derive_tros(graph, elements, analysis.findings, analysis.overrides, - resolve_host=resolve_host, count_exits=count_exits) + resolve_host=resolve_host, count_exits=count_exits, + sepliste_counts=sepliste_counts) analysis.boundaries = derive_boundaries(elements, graph) # Fuer Verortung und Verknuepfung zaehlen TROs und Systemgrenzen zusammen: # link_tros() registriert eine Kante nur zu einem Knoten, der Host eines @@ -2073,10 +2157,15 @@ def main(argv: list[str] | None = None) -> int: resolve_host, count_exits = make_geometric_resolvers(elements, tol_mm=args.geo_tol_mm) method = f"geometrisch (STRtree, Toleranz {args.geo_tol_mm:.0f}mm)" + sepliste_counts: dict[str, int] = {} + raw_json_file = find_sibling_raw_json(csv_file) + if raw_json_file is not None: + sepliste_counts = load_sepliste_counts(raw_json_file, warnings) + overrides = load_overrides(csv_file.name, env_dir("SKEL_CFG", "cfg")) analysis = analyse_tros(graph, elements, overrides, resolve_host=resolve_host, count_exits=count_exits, - method=method) + method=method, sepliste_counts=sepliste_counts) if not analysis.tros: print(f"FEHLER: {csv_file} enthaelt keine Separatoren - keine TROs " f"ableitbar.", file=sys.stderr)