Compare commits

...
10 Commits
Author SHA1 Message Date
Michael Stangl 1479519bdc Merge branch 'show-bbs-all' 2026-09-25 22:19:19 +02:00
Michael StanglandClaude Opus 5.5 a1bba2b20d material_flow: --show-bbs [flow|all] mit Legende und Tooltips
- neuer Wert fuer --show-bbs: 'flow' (Default, auch ohne Wert) blendet
  die VF/GF-Einzelbausteine (Bogen, Streckenelement, Motor-/Umlenkstation)
  aus, 'all' zeichnet jede CSV-Zeile
- eigene Farben fuer die Einzelbausteine, Legende mit Anzahl je Objektart,
  Tooltip je Rechteck (Zeile, TeileId, TeileArt, Bezeichnung)
- genutzt von dxfmakros/bin/show_exportcsv.bat

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-25 20:25:38 +02:00
Samer AyadiandClaude Sonnet 5 058c60e23d TRO-Fluss: Weichenerkennung ueber CSV-Nachbarn statt Distanzschwelle
_find_switch_separators() ermittelt Weichen-Separatoren jetzt topologisch
(Ausschleuselement unter den CSV-Nachbarn, gleiche Kreisel-Bahn) statt ueber
eine Distanzschwelle - die Fallback-Schwelle paarte an Mubea unbeteiligte
Separatoren faelschlich mit dem naechsten freien Ausschleuselement, u.a. 20
PinStore_Auto-Speicherlinien wurden so faelschlich als 1Sep1Swi erkannt.

Materialfluss: _sole_kreisel_neighbour() als Fallback in _apply_transfers()
fuer AS/ES-Elemente ohne eigenes Merkmal "Kreisel" (Mubea), aufloesbar wenn
die Strecke genau einen Kreisel-Nachbarn hat.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-09-25 15:06:25 +02:00
Samer AyadiandClaude Sonnet 5 e4bfad4dcc TRO-Fluss: Vario/1Sep alternierend rueckwaerts statt nur letzter Separator
Bisher wurde auf einer Streckengruppe nur der jeweils letzte Separator (in
TrackIds-Reihenfolge) auf 1Sep umgestuft, alle davor blieben pauschal Vario.
Eine Streckengruppe ist baulich aber meist eine alternierende VF/GF-Kette
(angetriebener Abschnitt, Vereinzeler, unbetriebener Abschnitt, Vereinzeler,
...), nicht nur am Ende - fuer 2 Separatoren macht das keinen Unterschied,
ab 3 schon. Die Umstufung laeuft jetzt vom letzten Separator rueckwaerts und
alterniert: letzter = 1Sep, davor Vario, davor wieder 1Sep usw.

Bestaetigt an HundM05 VF_1 (4 Separatoren): der neue Positions-Nachweis stuft
den dritten von hinten (0035) unabhaengig auf 1Sep ein - deckungsgleich mit
dem bereits vorhandenen (aber als moeglicherweise unkorrigierte CAD-Vorgabe
mit Vorbehalt behafteten) Hoehen-/Antriebsnachweis (_incoming_drop). Der
Hoehen-/Antriebsnachweis bleibt als unabhaengige Zusatzpruefung erhalten und
kann eine laut Muster erwartete Vario-Position auf 1Sep umstufen, wenn dort
tatsaechlich ein unbetriebenes Gefaelle gemessen wird - nicht umgekehrt.

TRO-Typenzahl auf beiden Referenzanlagen (HundM05, Mubea) unveraendert,
nur die Finding-Texte wurden praeziser.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-09-24 15:55:26 +02:00
Samer AyadiandClaude Sonnet 5 ef3d8170fe TRO-Fluss: sepliste (Roh-JSON) als Gegenprobe fuer die Separatorenzahl je Host
Optional: findet tro_flow.py neben der CSV ein <stamm>_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 <noreply@anthropic.com>
2026-09-24 15:40:43 +02:00
Samer AyadiandClaude Sonnet 5 f4b0e4d508 TRO-Fluss: Separatoren ohne TrackIds-Mitgliedschaft als eigenstaendige 1Sep
Ein per Zuordnung einer Streckengruppe zugeordneter Separator, der nicht Teil
von deren eigener TrackIds-Kette ist, wird direkt als 1Sep eingestuft statt
ueber die Vario/letzter-Separator-Logik zu laufen - vor Ort bei Mubea
VF_1-VF_3 bestaetigt: jede Streckengruppe hat genau einen echten
Kettenseparator, alle weiteren Zuordnung-Treffer sind eigenstaendige
Vereinzeler ohne eigenes VF-Teilstueck. Damit ein solcher Separator nicht
zugleich faelschlich als Weichen-Indiz zaehlt (engster Treffer ueberhaupt bei
Mubea 0005, 163mm), werden echte VF-Kettenmitglieder vor der
Weichenerkennung (_find_switch_separators) ausgenommen - Naehe zum eigenen
Aus-/Einschleuselement ist bei ihnen der normale, erwartete Aufbau, kein
Weichenbeleg.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-09-24 13:16:36 +02:00
Samer AyadiandClaude Sonnet 5 97965e8ec7 TRO-Fluss: Weichenerkennung ohne Bahn-Sperre, Vario/1Sep-Sonderfaelle je Separator
_find_switch_separators: Regel 2 (naechster freier Separator <= 1500mm) lief
bisher nur auf einer Bahn, die Regel 1 (<=500mm) schon bestaetigt hatte. Bei
HundM05 lagen nach einer CAD-Aktualisierung alle 5 echten Weichen-Separatoren
gleichmaessig bei ca. 1013-1016mm - ausserhalb von Regel 1, wodurch die
Bahn-Sperre fuer jede Bahn zublieb und keine einzige Weiche mehr erkannt
wurde. Die Sperre entfaellt; SWITCH_FALLBACK_MM bleibt die alleinige Schranke
gegen Fehlzuordnungen (der eigentliche Schutz vor dem urspruenglichen
Mubea-Fehlfund, der 5-9m entfernte Separatoren betraf).

derive_tros, angetriebene Strecken (Schritt 2): zwei Faelle stufen einen
Separator trotz Host "Strecke" als 1Sep statt Vario ein:

- Letzter Separator vor dem Wiedereinschleusen auf die Kreisel-Bahn (in
  TrackIds-Reihenfolge) - reiner Vereinzeler ohne eigene VarioDrive-Strecke,
  die Kreisel-Bahn treibt ab dort selbst an (HundM05 VF_3 vor Ort bestaetigt).
- Ein nicht-letzter Separator, dessen vorangehendes Segment "horizontal"
  (keine aktive Antriebsrichtung) gefuehrt wird, dabei aber tatsaechlich an
  Hoehe verliert (> 30mm) - ein unbetriebenes Gefaelle-Teilstueck innerhalb
  einer gemischten VF/GF-Streckengruppe (HundM05 VF_1/0035).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-09-23 17:20:18 +02:00
Samer AyadiandClaude Sonnet 5 df6d56090e Materialfluss: Ein-/Ausgabestationen im Chain-Graph zeichnen
BTMT Beladung/Entladung bzw. die generischen Eingang/Ausgang-Symbole waren
bislang von --chain-graph ausgenommen und wurden als Kantenziel einfach
uebersprungen. Die Anlage laeuft als Kreislauf, diese Stationen sind ihre
einzigen Materialquellen/-senken und gehoeren als eigene Knoten in jeden
Graphtyp (coarsebox/compound/subgraph).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-09-23 17:20:02 +02:00
Samer AyadiandClaude Sonnet 5 0477f05180 Materialfluss: --graphtype fuer --chain-graph (coarsebox/compound/subgraph)
--chain-graph zeichnete bisher immer nur die groben Objekte (Kreisel-Bahnen,
Strecke/Gefaellestrecke). Neuer Schalter --graphtype waehlt die Methode:
coarsebox (bisheriges Verhalten, unveraendert), compound (dieselbe
massstabsgetreue SVG, zusaetzlich die Anbauteile - Separator, Scanner,
Ausschleus-/Einschleuselement, Weiche - verschachtelt im Container an ihrer
echten Position, verbunden ueber die bereits aufgeloesten TrackGraph-Kanten)
und subgraph (dieselbe Verschachtelung als geschachtelte Graphviz-Cluster,
schematisch statt lagerichtig, <name>_chain.dot / mit --tosvg auch .svg).

_container_children() liest die Anbauteile direkt aus Element.track_ids -
derselben Quelle, die build_track_graph()/--track schon nutzt - statt sie wie
im Prototyp data/gen_chain_svg.py erneut aus der Nachbarn-Spalte herzuleiten.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-09-22 16:26:02 +02:00
Samer AyadiandClaude Sonnet 5 b197bf8150 TRO-Fluss: Reihenfolge auf der Kreisel-Bahn auch fuer Weichen-TROs
Seit der Weichenerkennung (c788647/fa96a6e) bekommt ein Weichen-TRO als Host
die gefuetterte Strecke statt die Kreisel-Bahn (derive_tros Schritt 3b) - fachlich
richtig fuer den Materialfluss, aber die Positions-Verkettung in link_tros()
gruppierte bisher rein nach Host: sobald nur noch ein einzelner "echter"
Separator auf einer Bahn uebrig blieb (HundM05 Kreisel1 Bahn L: 0020, nachdem
0008/0010/0012/0014 eigene Weichen-TROs wurden), verschwand die Reihenfolge
0008->0010->0012->0014->0020 komplett - im TRO-Flussdiagramm sah es aus, als
haenge jede der 5 Abzweigungen direkt und gleichrangig an genau diesem einen
TRO (Review 2026-09-21: "es gibt viele unnoetige Linien").

_tro_bahn() bestimmt die Kreisel-Bahn eines TROs jetzt zusaetzlich ueber die
Zuordnung seines eigenen Separators, wenn der TRO-Host selbst keine Bahn ist -
link_tros() verkettet danach alle TROs derselben Bahn per Position, nicht nur
die mit identischem Host. Die bestehenden Materialfluss-Kanten (Weiche/Vario
-> Ruecklauf-Bahn) bleiben unveraendert bestehen; die Verkettung ergaenzt nur
die bislang fehlende Reihenfolge der Abzweige.

Die Colocated-Markierung ("gleiche Bahn", gepunktete Linie) wird jetzt pro
Paar statt pro TRO unterdrueckt (verkettete_paare statt verkettet als TRO-Id-
Menge) - sonst waere z. B. TRO01 (Weiche, auf der Bahn verkettet) faelschlich
auch von seiner Colocated-Markierung mit TRO02 (Vario, gleiche Strecke)
ausgenommen worden, obwohl das zwei unabhaengige Beziehungen sind.

HundM05.csv: 24 -> 28 Kanten (4 neue Bahn-Kanten 0008->0010->0012->0014->0020).
Mubea unveraendert (13 TROs, 55 Kanten), keine Fehler in beiden Projekten.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-09-21 13:01:59 +02:00
3 changed files with 1096 additions and 161 deletions

No files matched your search

+5 -3
View File
@@ -241,16 +241,18 @@ See `doc/track.md` for the fuller derivation notes.
| `--use-cords` (alias `--use-coords`) | — | off | Place every node at its plant coordinate and render with `neato -n` (fixed positions) instead of `dot`, so the material graph matches the plant layout — the same option `tro_flow.py` has. The two Kreisel lanes (`-L`/`-R`) are nudged apart by `LANE_NUDGE_MM`; scale is `COORD_SCALE` (0.25 pt/mm, identical to `tro_flow.py`). Only has an effect together with `--tosvg`; fails (exit 1) if any flow node has no coordinate. |
| `--doc` | — | off | Also write a Markdown report (`<name>_material_flow.md`) with object list, connections and plausibility findings. |
| `--connect` | — | off | Also write the topology as `<name>_connect.ini` (item/flow level): `[…nodes]` (Kreisel lanes, Strecke, Gefällestrecke), `[…connections]` with a `kind` per edge (`normal`/`umlauf`/`weiche`/`einschleusung`/`ausschleusung`), and `[…externals]` — the BTMT entry/exit if present, otherwise the open lane-ends listed as entry candidates. A hand-editable bridge into the TRO step, styled after `doc/TRO_Katalog/connect.ini`. |
| `--show-bbs` | — | off | Also write `<name>_bbs.svg`: a plain 2-D top-down view of every CSV row (not just flow nodes — Separator/Scanner/Ein-/Ausschleuselement too), one axis-aligned rectangle per row from the `Position` column (its center, X/Y) and the `Boundingbox` column (its width/depth, X/Y — the Z extent is ignored), with the row's `Bezeichnung` centered in it. A Kreisel is the one exception: instead of one rectangle for the whole ring, it's split into its two Bahn L/R halves (`_kreisel_lane_rects()` — short axis divided in half, `Zuordnung`-majority-vote decides which half is L vs R), each drawn and labelled separately; its K1–K4 points (with `--show-kx`) still belong to the Kreisel as a whole and are drawn once, not per half. No rotation, no Graphviz — hand-built SVG; the `viewBox` stays in plant millimetres (Y flipped so plant "up" renders as screen "up") but the printed page size is scaled to DIN A3 (landscape or portrait, whichever matches the layout's aspect ratio) instead of the plant's real-world extent. Larger objects are drawn first so small ones (Separator, Scanner, …) stay visible on top. |
| `--show-bbs [flow\|all]` | `flow` | off | Also write `<name>_bbs.svg`: a plain 2-D top-down view of every CSV row (not just flow nodes — Separator/Scanner/Ein-/Ausschleuselement too), one axis-aligned rectangle per row from the `Position` column (its center, X/Y) and the `Boundingbox` column (its width/depth, X/Y — the Z extent is ignored), with the row's `Bezeichnung` centered in it. A Kreisel is the one exception: instead of one rectangle for the whole ring, it's split into its two Bahn L/R halves (`_kreisel_lane_rects()` — short axis divided in half, `Zuordnung`-majority-vote decides which half is L vs R), each drawn and labelled separately; its K1–K4 points (with `--show-kx`) still belong to the Kreisel as a whole and are drawn once, not per half. No rotation, no Graphviz — hand-built SVG; the `viewBox` stays in plant millimetres (Y flipped so plant "up" renders as screen "up") but the printed page size is scaled to DIN A3 (landscape or portrait, whichever matches the layout's aspect ratio) instead of the plant's real-world extent. Larger objects are drawn first so small ones (Separator, Scanner, …) stay visible on top. A legend column right of the plant lists every drawn kind with its row count, and each rectangle carries a `<title>` tooltip (row, TeileId, TeileArt, name). Mode `flow` (default, also bare `--show-bbs`) skips the VF/GF chain detail rows exported by dxfmakros since Sept. 2026 (`Bogen`, `Streckenelement`, `Motorstation`, `Umlenkstation` — `BBS_DETAIL_KINDS`), because they fully overlap their Strecke/Gefällestrecke box; `all` draws every row, detail kinds in their own stronger colours. dxfmakros calls this via `bin/show_exportcsv.bat` (mode `all`). |
| `--show-kx` | — | off | Only valid together with `--show-bbs` (else exit 1). Additionally draw each object's connection coordinate systems K1–K4 into the `_bbs.svg`: the K1–K4 columns are 12-char base64 strings (three 24-bit fixed-point values `x,y,z`, factor 10, `csv:trans-encode` in `dxfmakros/Lisp/export.lsp`) decoded to a plant-mm X/Y. Each present K-point is drawn as a small red cross (two lines) with a small `K1`…`K4` label, and consecutive points are joined `K1→K2→K3→K4` by a blue line. Empty K columns are skipped, so an object with only K1/K2 gets two crosses and one link. |
| `--track` | — | off | Additionally write the finer Separator/Ausschleus/Einschleus/Weiche graph described above (`<name>_track.dot`, and `<name>_track.svg` with `--tosvg`). Does not change the normal `_material_flow.*` output. See `doc/track.md`. |
| `--chain-graph` | — | off | Write **one** to-scale "chain-/Lageplan" SVG of the **grobe Objekte** (big items) as `<name>_chain.svg`: every Kreisel split into its Bahn L/R halves (`_kreisel_lane_rects()`, same to-scale split as `--show-bbs`) and every Strecke/Gefällestrecke ("VF"/"GF") drawn as rectangles at their real `Position`/`Boundingbox` coordinates, connected by the already-resolved top-level `graph.edges` (material flow, Kreisel-Umlauf, unresolved). Step 1 of the new chain-graph build-up — no Separator/Scanner/Weiche/Ausschleus-/Einschleuselement and no BTMT Ein-/Ausgang yet; an edge touching a node that isn't drawn (e.g. a station) is simply skipped. Hand-built SVG, no Graphviz. |
| `--chain-graph` | — | off | Write a "chain-/Lageplan" of the **grobe Objekte** (big items): every Kreisel split into its Bahn L/R halves (`_kreisel_lane_rects()`, same split as `--show-bbs`), every Strecke/Gefällestrecke ("VF"/"GF"), and every Ein-/Ausgabestation (BTMT `Beladung`/`Entladung`, or the generic `Eingang`/`Ausgang` symbol where no BTMT station absorbed it — see `_merge_station_symbols()`) — the plant runs as a closed loop, so these stations are its only material sources/sinks and are drawn like any other node, not skipped. Connected by the already-resolved top-level `graph.edges` (material flow, Kreisel-Umlauf, unresolved). Rendering method and output file are chosen by `--graphtype`. |
| `--graphtype` | `coarsebox`\|`compound`\|`subgraph` | `coarsebox` | Only valid together with `--chain-graph` (else exit 1). `coarsebox`: containers only (as above), to real scale, hand-built `<name>_chain.svg`, no Graphviz — the original behaviour. `compound`: the same to-scale picture, **plus** each container's Anbauteile (Separator, Scanner, Ausschleus-/Einschleuselement, Weiche) drawn nested inside it at their real position (`_container_children()` reads them off `Element.track_ids`, the same source `build_track_graph()`/`--track` uses — not a re-derivation from `Nachbarn` like the old `data/gen_chain_svg.py` prototype), linked by the already-resolved `TrackGraph.edges`; still `<name>_chain.svg`, no Graphviz. `subgraph`: the same containment expressed as **nested Graphviz clusters** (`subgraph cluster_*` — one per Kreisel, nested one per Bahn L/R, plus one per Strecke/Gefällestrecke, each containing its Anbauteile as child nodes) — schematic (`dot` auto-layout), **not** to scale; writes `<name>_chain.dot` always, and `<name>_chain.svg` additionally with `--tosvg` (needs Graphviz `dot`, unlike the other two graph types). |
**Output** (in `%SKEL_RESULTS%`): `<name>_material_flow.dot` (always, with `pos`
attributes when `--use-cords` was used), `<name>_material_flow.svg` (with `--tosvg`),
`<name>_material_flow.md` (with `--doc`), `<name>_connect.ini` (with `--connect`),
`<name>_bbs.svg` (with `--show-bbs`), `<name>_track.dot` / `<name>_track.svg` (with `--track`),
`<name>_chain.svg` (with `--chain-graph`).
`<name>_chain.svg` (with `--chain-graph --graphtype coarsebox|compound`, or `--graphtype subgraph --tosvg`),
`<name>_chain.dot` (with `--chain-graph --graphtype subgraph`).
**Exit codes:** `0` ok · `1` input/CLI error · `2` SVG rendering failed
(Graphviz).
+452 -26
View File
@@ -1152,6 +1152,30 @@ def _attach_stations(
return edges
def _sole_kreisel_neighbour(strecke_id: str, by_id: dict[str, Element]) -> str:
"""Kreisel-Basis-Id, wenn die Strecke genau einen Kreisel als Nachbarn hat, sonst ''.
Fallback fuer _apply_transfers, wenn ein Ausschleus-/Einschleuselement
selbst kein Merkmal "Kreisel" traegt (z. B. Mubea: die dortigen AS/ES-
Bloecke haben dieses Merkmal grundsaetzlich nicht, siehe Review
2026-09-25). Bewusst nur bei GENAU einem Kreisel-Nachbarn: bei zwei
verschiedenen Kreiseln (der Normalfall) waere sonst nicht entscheidbar,
welcher zu "aus" (Ausschleusung) und welcher zu "ein" (Einschleusung)
gehoert - dort bleibt es beim bisherigen Hoehenvergleich in _orient(), der
dort zuverlaessig funktioniert (unterschiedliche Kreisel, unterschiedliche
Hoehe). Nur bei einer Ruecklauf-Strecke (beide Enden am selben Kreisel,
siehe unten) ist der einzige Nachbar ohnehin fuer beide Rollen richtig.
"""
strecke = by_id.get(strecke_id)
if strecke is None:
return ""
bases = {
base for base, _lane in strecke.neighbours
if (partner := by_id.get(base)) is not None and partner.is_circle
}
return next(iter(bases)) if len(bases) == 1 else ""
def _kreisel_lane_node(kreisel_base: str, strecke: Element, nodes: dict[str, Node]) -> str | None:
"""Die konkrete Kreisel-Bahn (…-L/…-R), an der die Strecke haengt."""
for base, lane in strecke.neighbours:
@@ -1217,6 +1241,12 @@ def _apply_transfers(
# "Strecke" (VF) oder "Foerderer"/"Förderer" (GF/Streckengruppe).
s = str(_merkmal(el.merkmale, "Strecke", "Foerderer", "Förderer") or "").strip()
if not k and s:
# AS/ES ohne eigenes Merkmal "Kreisel" (z. B. Mubea) - nur bei
# einem einzigen Kreisel-Nachbarn der Strecke aufloesbar, siehe
# _sole_kreisel_neighbour.
k = _sole_kreisel_neighbour(s, by_id)
if k and s:
role = "aus" if el.kind == "Ausschleus" else "ein"
per_strecke.setdefault(s, {})[role] = k
@@ -2279,9 +2309,27 @@ BBS_STYLES: dict[str, dict[str, str]] = {
"Einschleus": dict(fill="#cfe2f3", stroke="#3d85c6", text="#1155cc"),
"Ausschleus": dict(fill="#f9cb9c", stroke="#e69138", text="#b45f06"),
"Weiche": dict(fill="#ead1dc", stroke="#a64d79", text="#741b47"),
# Einzelbausteine einer VF/GF-Kette (nur mit --show-bbs all, siehe
# BBS_DETAIL_KINDS) - sie liegen innerhalb ihrer Strecke/Gefaellestrecke
# und brauchen deshalb kraeftigere Farben, um darauf sichtbar zu bleiben.
"Bogen": dict(fill="#b6d7a8", stroke="#38761d", text="#274e13"),
"Streckenelement": dict(fill="#a2c4c9", stroke="#134f5c", text="#0c343d"),
"Motorstation": dict(fill="#ea9999", stroke="#990000", text="#660000"),
"Umlenkstation": dict(fill="#b4a7d6", stroke="#351c75", text="#20124d"),
}
BBS_DEFAULT_STYLE = dict(fill="#f2f2f2", stroke="#808080", text="#404040")
# Einzelbausteine der VF_n/GF_n-Ketten, die dxfmakros/export_csv.py seit
# Sept. 2026 als eigene Proxy-Zeilen exportiert. Sie ueberlagern die Box ihrer
# Strecke/Gefaellestrecke vollstaendig; --show-bbs (= flow) blendet sie
# deshalb aus, --show-bbs all zeichnet sie mit.
BBS_DETAIL_KINDS = ("Bogen", "Streckenelement", "Motorstation", "Umlenkstation")
BBS_MODES = ("flow", "all")
# Legende (--show-bbs): Kaestchen- und Schriftgroesse in mm.
BBS_LEGEND_BOX_MM = 250.0
BBS_LEGEND_FONT_MM = 180.0
# Rand um die Anlage herum und minimale Schriftgroesse, jeweils in mm.
BBS_MARGIN_MM = 500.0
BBS_MIN_FONT_MM = 40.0
@@ -2312,6 +2360,7 @@ def render_bbs_svg(
source: str,
warnings: list[str],
show_kx: bool = False,
mode: str = "flow",
) -> str:
"""
2D-Draufsicht der ganzen Szene: ein achsparalleles Rechteck je CSV-Zeile,
@@ -2339,9 +2388,19 @@ def render_bbs_svg(
entscheid fuer L/R), siehe _kreisel_lane_rects().
Die K1-K4-Anschlusspunkte (real am Kreisel als Ganzem haengend) werden
trotzdem nur einmal gezeichnet, nicht je Bahn doppelt.
mode 'flow' (Default) laesst die Einzelbausteine der VF/GF-Ketten
(BBS_DETAIL_KINDS) weg, 'all' zeichnet wirklich jede Zeile.
Rechts oben steht in beiden Faellen eine Legende (Farbe, Objektart,
Anzahl); jedes Rechteck traegt einen Tooltip mit Zeile/TeileId/
TeileArt.
"""
if mode not in BBS_MODES:
raise ValueError(f"unbekannter Modus '{mode}' (erlaubt: {', '.join(BBS_MODES)})")
items: list[tuple[Element, float, float, float, float, str | None]] = []
for element in elements:
if mode == "flow" and element.kind in BBS_DETAIL_KINDS:
continue
if element.position is None or element.bbox is None:
warnings.append(
f"{element.describe()}: keine Position/Boundingbox - in --show-bbs nicht gezeichnet"
@@ -2367,7 +2426,24 @@ def render_bbs_svg(
max_x = max(item[1] + item[3] / 2 for item in items) + BBS_MARGIN_MM
min_y = min(item[2] - item[4] / 2 for item in items) - BBS_MARGIN_MM
max_y = max(item[2] + item[4] / 2 for item in items) + BBS_MARGIN_MM
width, height = max_x - min_x, max_y - min_y
plant_w = max_x - min_x
# Legende als eigene Spalte rechts neben der Anlage (Objektart + Anzahl der
# gezeichneten Zeilen; ein Kreisel zaehlt einmal, nicht je Bahn).
counts: dict[str, int] = {}
counted: set[str] = set()
for element, *_rest in items:
if element.teile_id not in counted:
counted.add(element.teile_id)
counts[element.kind] = counts.get(element.kind, 0) + 1
legend = [(kind, f"{kind} ({count})") for kind, count in sorted(counts.items())]
legend_title = f"{len(counted)} Zeilen ({mode})"
row_h = BBS_LEGEND_BOX_MM * 1.5
text_w = max(len(text) for _, text in legend + [("", legend_title)]) * BBS_LEGEND_FONT_MM * 0.6
legend_w = BBS_LEGEND_BOX_MM * 1.5 + text_w + BBS_MARGIN_MM
legend_h = (len(legend) + 1) * row_h + BBS_MARGIN_MM
width = plant_w + legend_w
height = max(max_y - min_y, legend_h)
# Groesste Objekte zuerst zeichnen, damit kleine Anbauteile (Separator,
# Scanner, ...) sichtbar obenauf liegen statt unter einer Strecke/einem
@@ -2391,8 +2467,10 @@ def render_bbs_svg(
y = max_y - (py + bd / 2)
name = f"{element.name or element.teile_id} Bahn {lane}" if lane else (element.name or element.teile_id)
font_size = max(BBS_MIN_FONT_MM, min(bw, bd) * 0.28)
tooltip = _xml_escape(f"Zeile {element.row} | {element.teile_id} | {element.teile_art} | {name}")
add(f' <rect x="{x:.2f}" y="{y:.2f}" width="{bw:.2f}" height="{bd:.2f}" '
f'fill="{style["fill"]}" fill-opacity="0.85" stroke="{style["stroke"]}" stroke-width="6"/>')
f'fill="{style["fill"]}" fill-opacity="0.85" stroke="{style["stroke"]}" stroke-width="6">'
f'<title>{tooltip}</title></rect>')
add(f' <text x="{x + bw / 2:.2f}" y="{y + bd / 2:.2f}" '
f'font-family="Segoe UI, sans-serif" font-size="{font_size:.1f}" fill="{style["text"]}" '
f'text-anchor="middle" dominant-baseline="middle">{_xml_escape(name)}</text>')
@@ -2422,6 +2500,22 @@ def render_bbs_svg(
f'font-family="Segoe UI, sans-serif" font-size="{KX_FONT_MM:.1f}" '
f'fill="{KX_CROSS_COLOR}" dominant-baseline="middle">{label}</text>')
# Legende zuletzt, rechts neben der Anlage.
lx = plant_w + BBS_MARGIN_MM / 2
ly = BBS_MARGIN_MM / 2
add(f' <text x="{lx:.2f}" y="{ly + BBS_LEGEND_BOX_MM / 2:.2f}" font-family="Segoe UI, sans-serif" '
f'font-size="{BBS_LEGEND_FONT_MM:.1f}" font-weight="bold" fill="#000000" '
f'dominant-baseline="middle">{_xml_escape(legend_title)}</text>')
for index, (kind, text) in enumerate(legend, start=1):
style = BBS_STYLES.get(kind, BBS_DEFAULT_STYLE)
ry = ly + index * row_h
add(f' <rect x="{lx:.2f}" y="{ry:.2f}" width="{BBS_LEGEND_BOX_MM:.2f}" '
f'height="{BBS_LEGEND_BOX_MM:.2f}" fill="{style["fill"]}" stroke="{style["stroke"]}" '
f'stroke-width="6"/>')
add(f' <text x="{lx + BBS_LEGEND_BOX_MM * 1.5:.2f}" y="{ry + BBS_LEGEND_BOX_MM / 2:.2f}" '
f'font-family="Segoe UI, sans-serif" font-size="{BBS_LEGEND_FONT_MM:.1f}" '
f'fill="#000000" dominant-baseline="middle">{_xml_escape(text)}</text>')
add("</svg>")
return "\n".join(out) + "\n"
@@ -2479,6 +2573,45 @@ def _kreisel_lane_rects(
return {lane_a: half_a, lane_b: half_b}
def _lane_of_point(pt: tuple[float, float], lane_rects: dict[str, tuple[float, float, float, float]]) -> str:
"""Fallback-Bahnzuordnung ueber die Geometrie (naechstes Rechteck), falls ein
Feinelement keine auswertbare '<Kreisel>-L'/'-R'-Zuordnung traegt (siehe
_container_children)."""
px, py = pt
def dist(rect: tuple[float, float, float, float]) -> float:
x0, y0, x1, y1 = rect
cx, cy = min(max(px, x0), x1), min(max(py, y0), y1)
return (cx - px) ** 2 + (cy - py) ** 2
return min(lane_rects, key=lambda lane: dist(lane_rects[lane]))
def _container_children(element: Element, by_id: dict[str, Element]) -> list[Element]:
"""
Anbauteile (Separator, Scanner, Ausschleus-/Einschleuselement, Weiche) eines
Containers (Kreisel/Strecke/Gefaellestrecke) aus seiner TrackIds-Kette, in
Reihenfolge - fuer die verschachtelte Darstellung in --graphtype compound/
subgraph. Nutzt dieselbe Quelle (Element.track_ids) und dieselbe
Richtungs-Regel fuer Strecke/Gefaellestrecke wie build_track_graph() Schritt 1
(Kreisel-seitiges Element zuerst umdrehen), statt Nachbarn-Heuristiken wie im
Prototyp data/gen_chain_svg.py neu zu erfinden.
"""
chain = list(element.track_ids)
if element.is_transport and chain:
first_el, last_el = by_id.get(chain[0]), by_id.get(chain[-1])
first_kind = first_el.kind if first_el else ""
last_kind = last_el.kind if last_el else ""
if first_kind == "Einschleus" or last_kind == "Ausschleus":
chain = list(reversed(chain))
children: list[Element] = []
for tid in chain:
el = by_id.get(tid)
if el is not None and el.kind in TRACK_KINDS:
children.append(el)
return children
# ---------------------------------------------------------------------------
# Chain-/Lageplan-SVG der groben Objekte erzeugen (--chain-graph)
# ---------------------------------------------------------------------------
@@ -2506,19 +2639,40 @@ def _chain_marker_id(color: str) -> str:
return "chainArrow_" + color.lstrip("#")
def render_chain_svg(elements: list[Element], graph: Graph, source: str) -> str:
def render_chain_svg(
elements: list[Element], graph: Graph, source: str, track: TrackGraph | None = None
) -> str:
"""
Massstabsgetreuer Chain-/Lageplan der groben Objekte (--chain-graph): jeder
Kreisel als seine beiden Bahnen L/R (_kreisel_lane_rects(), Zuordnung aus
allen Elementen fuer den Mehrheitsentscheid - wie --show-bbs) und jede
allen Elementen fuer den Mehrheitsentscheid - wie --show-bbs), jede
Strecke/Gefaellestrecke als ein Rechteck an ihrer echten Position/
Boundingbox. Kanten kommen direkt aus dem schon aufgeloesten groben
Boundingbox, und die BTMT-/generischen Ein-/Ausgabestationen (Beladung/
Entladung/Eingang/Ausgang) ebenso - die Anlage laeuft als Kreislauf, die
Stationen sind seine einzigen Materialquellen/-senken und gehoeren mit in
jeden Chain-Graph. Gezeichnet werden nur Stationen, die auch echte Knoten
im groben Materialfluss-Graphen sind (graph.nodes) - ein durch eine
deckungsgleiche BTMT-Station verschlucktes generisches Eingang/Ausgang-
Symbol (_merge_station_symbols(), siehe build_graph()) taucht folgerichtig
nicht doppelt auf. Kanten kommen direkt aus dem schon aufgeloesten groben
Materialfluss-Graphen (graph.edges) - eine Kante, deren Knoten hier nicht
gezeichnet wird (z. B. eine BTMT-Station), wird schlicht uebersprungen.
gezeichnet wird (z. B. eine Station ohne Position/Boundingbox), wird
schlicht uebersprungen.
track nur fuer --graphtype compound (siehe render_chain_compound_svg): mit
angegebenem TrackGraph werden zusaetzlich die Anbauteile (Separator,
Scanner, Ausschleus-/Einschleuselement, Weiche) jedes Containers an
ihrer echten Position hinein verschachtelt gezeichnet, verbunden mit
den schon aufgeloesten Kanten aus track.edges (_container_children()
liefert die Reihenfolge aus derselben TrackIds-Quelle, die auch
build_track_graph() benutzt - keine eigene Nachbarn-Herleitung wie im
Prototyp data/gen_chain_svg.py). None (Default) = reines --graphtype
coarsebox ohne Anbauteile, wie bisher.
"""
rects: dict[str, tuple[float, float, float, float]] = {}
lane_info: dict[str, tuple[Element, str]] = {}
strecke_labels: dict[str, Element] = {}
station_labels: dict[str, Element] = {}
for element in elements:
if element.position is None or element.bbox is None:
@@ -2535,6 +2689,16 @@ def render_chain_svg(elements: list[Element], graph: Graph, source: str) -> str:
)
strecke_labels[element.teile_id] = element
for node in graph.nodes.values():
element = node.element
if not element.is_station or element.position is None or element.bbox is None:
continue
rects[node.node_id] = (
element.x - element.bbox_x / 2, element.y - element.bbox_y / 2,
element.x + element.bbox_x / 2, element.y + element.bbox_y / 2,
)
station_labels[node.node_id] = element
if not rects:
raise ValueError("keine Kreisel/Strecken mit Position/Boundingbox gefunden")
@@ -2615,10 +2779,15 @@ def render_chain_svg(elements: list[Element], graph: Graph, source: str) -> str:
kreisel, lane = lane_info[node_id]
style = NODE_STYLES.get(kreisel.kind, DEFAULT_NODE_STYLE)
label = f"{kreisel.name or kreisel.teile_id} Bahn {lane}"
else:
elif node_id in strecke_labels:
strecke = strecke_labels[node_id]
style = NODE_STYLES.get(strecke.kind, DEFAULT_NODE_STYLE)
label = f"{strecke.name or node_id} [{node_id}]"
else:
station = station_labels[node_id]
style = NODE_STYLES.get(station.kind, DEFAULT_NODE_STYLE)
role = "Materialeingang" if station.is_entry else "Materialausgang"
label = f"{station.name or node_id} [{node_id}] - {role}"
box_svg.append(
f'<rect x="{sx(x0):.1f}" y="{sy(y1):.1f}" width="{(x1 - x0) * scale:.1f}" '
f'height="{(y1 - y0) * scale:.1f}" fill="{style["fillcolor"]}" fill-opacity="0.3" '
@@ -2629,6 +2798,60 @@ def render_chain_svg(elements: list[Element], graph: Graph, source: str) -> str:
f'fill="{style["color"]}">{_xml_escape(label)}</text>'
)
nested_svg: list[str] = []
if track is not None:
by_id = {e.teile_id: e for e in elements}
kreisel_lane_rects: dict[str, dict[str, tuple[float, float, float, float]]] = {}
for node_id, rect in rects.items():
if node_id in lane_info:
kreisel, lane = lane_info[node_id]
lane_rects = kreisel_lane_rects.setdefault(
kreisel.teile_id, _kreisel_lane_rects(kreisel, elements)
)
children = []
for el in _container_children(kreisel, by_id):
if el.x is None or el.y is None:
continue
base, _, el_lane = el.assignment.partition("-")
if base != kreisel.teile_id or el_lane not in (LANE_LEFT, LANE_RIGHT):
el_lane = _lane_of_point((el.x, el.y), lane_rects)
if el_lane == lane:
children.append(el)
elif node_id in strecke_labels:
strecke = strecke_labels[node_id]
children = [
el for el in _container_children(strecke, by_id)
if el.x is not None and el.y is not None
]
else:
continue # Station - kein Container, keine Anbauteile zum Verschachteln
if not children:
continue
rx0, ry0, rx1, ry1 = rect
child_ids = {el.teile_id for el in children}
edge_color = CHAIN_EDGE_COLOR[EDGE_CIRCLE if node_id in lane_info else EDGE_FLOW]
for te in track.edges:
if te.src not in child_ids or te.dst not in child_ids:
continue
src_el, dst_el = by_id[te.src], by_id[te.dst]
x1, y1 = sx(min(max(src_el.x, rx0), rx1)), sy(min(max(src_el.y, ry0), ry1))
x2, y2 = sx(min(max(dst_el.x, rx0), rx1)), sy(min(max(dst_el.y, ry0), ry1))
nested_svg.append(
f'<line x1="{x1:.1f}" y1="{y1:.1f}" x2="{x2:.1f}" y2="{y2:.1f}" '
f'stroke="{edge_color}" stroke-width="1"/>'
)
for el in children:
style = TRACK_NODE_STYLES.get(el.kind, DEFAULT_NODE_STYLE)
cx, cy = min(max(el.x, rx0), rx1), min(max(el.y, ry0), ry1)
px, py = sx(cx), sy(cy)
title = f"{el.kind} {el.name or el.teile_id} [{el.teile_id}]"
nested_svg.append(
f'<circle cx="{px:.1f}" cy="{py:.1f}" r="3.4" fill="{style["fillcolor"]}" '
f'stroke="{style["color"]}" stroke-width="1.2"><title>{_xml_escape(title)}</title></circle>'
)
heading = f"{Path(source).stem} - Materialfluss (grobe Objekte)"
out: list[str] = []
@@ -2655,10 +2878,160 @@ def render_chain_svg(elements: list[Element], graph: Graph, source: str) -> str:
f'Massstabsgetreu &#8226; Quelle: {_xml_escape(source)}</text>')
out.extend(box_svg)
out.extend(edge_svg)
out.extend(nested_svg)
add("</svg>")
return "\n".join(out) + "\n"
def render_chain_compound_svg(
elements: list[Element], graph: Graph, track: TrackGraph, source: str
) -> str:
"""--graphtype compound: wie render_chain_svg(), aber mit den Anbauteilen
(Separator, Scanner, Ausschleus-/Einschleuselement, Weiche) verschachtelt in
ihrem Container gezeichnet. Duenner Wrapper um render_chain_svg(track=...),
damit Rechteck-/Massstab-/Kanten-Aufbau nicht doppelt existiert."""
return render_chain_svg(elements, graph, source, track=track)
# ---------------------------------------------------------------------------
# Chain-/Lageplan als verschachtelte Graphviz-Cluster (--graphtype subgraph)
# ---------------------------------------------------------------------------
#
# Schematische (nicht massstabsgetreue) Alternative zu render_chain_svg(track=...):
# dieselbe Containment - Kreisel/Bahn/Strecke enthaelt seine Anbauteile - aber als
# verschachtelte Graphviz "subgraph cluster_*"-Bloecke statt handgebauter SVG-
# Rechtecke, gerendert von Graphviz' eigenem Layout (dot). Nuetzlich, wenn die
# Verschachtelung selbst interessiert und nicht die reale Geometrie.
CHAIN_SUBGRAPH_EDGE = {
EDGE_FLOW: '[color="#2f5597", penwidth=1.4]',
EDGE_CIRCLE: '[color="#8ea9db", style=dashed, penwidth=1.2, constraint=false]',
EDGE_UNRESOLVED: '[color="#bf8f00", style=dashed, dir=both]',
}
def _chain_child_dot(node_id: str, element: Element) -> str:
style = TRACK_NODE_STYLES.get(element.kind, DEFAULT_NODE_STYLE)
attrs = ", ".join(f'{key}="{value}"' for key, value in style.items())
label = _dot_label([f"{element.name or element.teile_id}", f"{element.kind} [{element.teile_id}]"])
return f' "{node_id}" [label="{label}", {attrs}];'
def render_chain_subgraph_dot(
elements: list[Element], graph: Graph, track: TrackGraph, source: str
) -> str:
"""
--graphtype subgraph: dieselbe Containment wie render_chain_compound_svg(),
aber als verschachtelte Graphviz-Cluster (Kreisel-Cluster mit einem
Bahn-L/R-Cluster je Bahn darin, Strecke/Gefaellestrecke-Cluster) statt
massstabsgetreuer SVG, plus die Ein-/Ausgabestationen (Beladung/Entladung/
Eingang/Ausgang) als eigene, nicht verschachtelte Knoten - die Anlage
laeuft als Kreislauf, sie sind seine einzigen Materialquellen/-senken.
Rendern mit dot -Tsvg (siehe dot_to_svg()) - NICHT lagerichtig, das Layout
kommt von Graphviz.
"""
by_id = {e.teile_id: e for e in elements}
out: list[str] = []
add = out.append
add("// Automatisch erzeugt von lib/material_flow.py - nicht manuell aendern.")
add(f"// Quelle: {source}")
add("// --chain-graph --graphtype subgraph: verschachtelte Cluster, nicht lagerichtig.")
add("digraph Materialfluss_Chain {")
add(' graph [rankdir=LR, splines=spline, nodesep=0.22, ranksep=1.2,')
add(' fontname="Segoe UI", fontsize=11, labelloc="t",')
add(f' label="Chain-Graph (verschachtelt) - {_dot_escape(source)}"];')
add(' node [shape=box, style="rounded,filled", fontname="Segoe UI", fontsize=8];')
add(' edge [fontname="Segoe UI", fontsize=8, arrowsize=0.8];')
add("")
top_ids: set[str] = set()
for index, element in enumerate(
sorted((e for e in elements if e.is_circle and e.position and e.bbox), key=lambda e: e.teile_id)
):
add(f" subgraph cluster_kreisel_{index} {{")
add(f' label="{_dot_escape(element.kind)} {_dot_escape(element.name or element.teile_id)} '
f'[{element.teile_id}]"; style="rounded,filled"; fillcolor="#eef2fa"; color="#2f5597";')
add(' fontname="Segoe UI"; fontsize=10; fontcolor="#1f3864"; margin=16;')
lane_rects = _kreisel_lane_rects(element, elements)
for lane in sorted(lane_rects):
lane_id = f"{element.teile_id}-{lane}"
top_ids.add(lane_id)
add(f' subgraph "cluster_{lane_id}" {{')
add(f' label="Bahn {lane}"; style="rounded,filled"; fillcolor="#ffffff"; '
f'color="#8ea9db"; fontname="Segoe UI"; fontsize=9; fontcolor="#375623"; margin=10;')
add(f' "{lane_id}" [label="Bahn {lane}", shape=box, style="rounded,filled", '
f'fillcolor="#2f5597", fontcolor="#ffffff", color="#1f3864"];')
children = []
for el in _container_children(element, by_id):
if el.x is None or el.y is None:
continue
base, _, el_lane = el.assignment.partition("-")
if base != element.teile_id or el_lane not in (LANE_LEFT, LANE_RIGHT):
el_lane = _lane_of_point((el.x, el.y), lane_rects)
if el_lane == lane:
children.append(el)
child_ids = {el.teile_id for el in children}
for el in children:
add(_chain_child_dot(el.teile_id, el))
for te in track.edges:
if te.src in child_ids and te.dst in child_ids:
add(f' "{te.src}" -> "{te.dst}" [color="#8ea9db"];')
add(" }")
add(" }")
add("")
for index, element in enumerate(
sorted((e for e in elements if e.is_transport and e.position and e.bbox), key=lambda e: e.teile_id)
):
node_id = element.teile_id
top_ids.add(node_id)
add(f" subgraph cluster_strecke_{index} {{")
add(f' label="{_dot_escape(element.name or node_id)} [{node_id}]"; '
f'style="rounded,filled"; fillcolor="#f5faf0"; color="#548235";')
add(' fontname="Segoe UI"; fontsize=10; fontcolor="#375623"; margin=14;')
style = NODE_STYLES.get(element.kind, DEFAULT_NODE_STYLE)
attrs = ", ".join(f'{key}="{value}"' for key, value in style.items())
add(f' "{node_id}" [label="{_dot_escape(element.name or node_id)}", {attrs}];')
children = [
el for el in _container_children(element, by_id)
if el.x is not None and el.y is not None
]
child_ids = {el.teile_id for el in children}
for el in children:
add(_chain_child_dot(el.teile_id, el))
for te in track.edges:
if te.src in child_ids and te.dst in child_ids:
add(f' "{te.src}" -> "{te.dst}" [color="#2f5597"];')
add(" }")
add("")
# Ein-/Ausgabestationen (Beladung/Entladung/Eingang/Ausgang): die Anlage
# laeuft als Kreislauf, sie sind seine einzigen Materialquellen/-senken -
# eigene (nicht verschachtelte) Knoten, damit ihre schon aufgeloesten
# Kanten unten nicht wie bisher uebersprungen werden.
for node in sorted(graph.nodes.values(), key=lambda n: n.node_id):
if not node.element.is_station:
continue
top_ids.add(node.node_id)
style = NODE_STYLES.get(node.element.kind, DEFAULT_NODE_STYLE)
attrs = ", ".join(f'{key}="{value}"' for key, value in style.items())
role = "Materialeingang" if node.element.is_entry else "Materialausgang"
label = _dot_label([node.element.name or node.node_id, f"{node.element.kind} - {role}"])
add(f' "{node.node_id}" [label="{label}", {attrs}];')
add("")
for edge in graph.edges:
if edge.src not in top_ids or edge.dst not in top_ids:
continue
attrs = CHAIN_SUBGRAPH_EDGE.get(edge.kind, CHAIN_SUBGRAPH_EDGE[EDGE_FLOW])
add(f' "{edge.src}" -> "{edge.dst}" {attrs};')
add("}")
return "\n".join(out) + "\n"
# ---------------------------------------------------------------------------
# SVG erzeugen
# ---------------------------------------------------------------------------
@@ -2935,6 +3308,7 @@ def report(
bbs_file: Path | None = None,
track_dot_file: Path | None = None,
track_svg_file: Path | None = None,
chain_dot_file: Path | None = None,
chain_file: Path | None = None,
) -> None:
kinds: dict[str, int] = {}
@@ -2968,7 +3342,11 @@ def report(
print(f"Bahnzusammenhang = {track_dot_file if track_dot_file else '- (ohne --track)'}")
if track_dot_file:
print(f" SVG = {track_svg_file if track_svg_file else '- (ohne --tosvg)'}")
print(f"Chain-Graph = {chain_file if chain_file else '- (ohne --chain-graph)'}")
if chain_dot_file:
print(f"Chain-Graph = {chain_dot_file}")
print(f" SVG = {chain_file if chain_file else '- (ohne --tosvg)'}")
else:
print(f"Chain-Graph = {chain_file if chain_file else '- (ohne --chain-graph)'}")
print("================================================================")
if stale_svg is not None:
@@ -3020,6 +3398,8 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
" material_flow.bat --file HundM05.csv --show-bbs\n"
" material_flow.bat --file mubea.csv --track --tosvg\n"
" material_flow.bat --file HundM05.csv --chain-graph\n"
" material_flow.bat --file HundM05.csv --chain-graph --graphtype compound\n"
" material_flow.bat --file HundM05.csv --chain-graph --graphtype subgraph --tosvg\n"
"\n"
"Exit-Codes:\n"
" 0 ok\n"
@@ -3064,11 +3444,17 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
parser.add_argument(
"--show-bbs",
dest="show_bbs",
action="store_true",
nargs="?",
const="flow",
default=None,
choices=BBS_MODES,
help="2D-Draufsicht der ganzen Szene als <name>_bbs.svg nach %%SKEL_RESULTS%% "
"schreiben: ein achsparalleles Rechteck je CSV-Zeile aus 'Position' "
"(Mittelpunkt) und 'Boundingbox' (Breite/Tiefe), mit der Bezeichnung "
"im Zentrum, skaliert auf DIN A3. Braucht kein Graphviz.",
"im Zentrum, Legende und Tooltip je Rechteck, skaliert auf DIN A3. "
"Braucht kein Graphviz. 'flow' (Default, auch ohne Wert) laesst die "
"Einzelbausteine der VF/GF-Ketten (Bogen, Streckenelement, Motor-/"
"Umlenkstation) weg, 'all' zeichnet jede Zeile.",
)
parser.add_argument(
"--show-kx",
@@ -3093,13 +3479,26 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
"--chain-graph",
dest="chain_graph",
action="store_true",
help="Massstabsgetreuen Chain-/Lageplan der groben Objekte als "
"<name>_chain.svg nach %%SKEL_RESULTS%% schreiben: jeder Kreisel "
"(je Bahn L/R) und jede Strecke/Gefaellestrecke (VF/GF) als "
"Rechteck an ihrer echten Position/Boundingbox, verbunden durch "
"die bereits aufgeloesten Materialfluss-Kanten. Schritt 1 - noch "
"ohne Anbauteile (Separator, Scanner, Weiche, ...) oder Ein-/"
"Ausgabestationen. Braucht kein Graphviz.",
help="Chain-/Lageplan der groben Objekte (Kreisel je Bahn L/R, Strecke/"
"Gefaellestrecke, plus Beladung/Entladung/Eingang/Ausgang - die "
"Anlage laeuft als Kreislauf, sie sind seine einzigen Material-"
"quellen/-senken) nach %%SKEL_RESULTS%% schreiben, verbunden durch "
"die bereits aufgeloesten Materialfluss-Kanten. Methode/Ausgabedatei "
"je --graphtype.",
)
parser.add_argument(
"--graphtype",
choices=("coarsebox", "compound", "subgraph"),
default="coarsebox",
help="Nur zusammen mit --chain-graph: Darstellungsmethode. 'coarsebox' "
"(Default) = nur die groben Objekte, massstabsgetreu, wie bisher "
"(<name>_chain.svg, kein Graphviz). 'compound' = zusaetzlich die "
"Anbauteile (Separator, Scanner, Ausschleus-/Einschleuselement, "
"Weiche) massstabsgetreu in ihrem Container verschachtelt "
"(<name>_chain.svg, kein Graphviz). 'subgraph' = dieselbe "
"Verschachtelung als geschachtelte Graphviz-Cluster, schematisch "
"statt lagerichtig (<name>_chain.dot, mit --tosvg zusaetzlich "
"<name>_chain.svg).",
)
return parser.parse_args(argv)
@@ -3139,12 +3538,18 @@ def main(argv: list[str] | None = None) -> int:
file=sys.stderr)
return 1
if args.graphtype != "coarsebox" and not args.chain_graph:
print("FEHLER: --graphtype ist nur zusammen mit --chain-graph moeglich.",
file=sys.stderr)
return 1
bbs_file = None
if args.show_bbs:
bbs_target = results / f"{stem}_bbs.svg"
try:
bbs_target.write_text(
render_bbs_svg(elements, csv_file.name, warnings, show_kx=args.show_kx),
render_bbs_svg(elements, csv_file.name, warnings, show_kx=args.show_kx,
mode=args.show_bbs),
encoding="utf-8",
)
except ValueError as exc:
@@ -3214,8 +3619,10 @@ def main(argv: list[str] | None = None) -> int:
track: TrackGraph | None = None
track_dot_file = track_svg_file = None
if args.track:
needs_track = args.track or (args.chain_graph and args.graphtype != "coarsebox")
if needs_track:
track = build_track_graph(elements, by_id, graph, warnings)
if args.track:
track_dot_target = results / f"{stem}_track.dot"
track_dot_target.write_text(render_track_dot(track), encoding="utf-8")
track_dot_file = track_dot_target
@@ -3228,15 +3635,33 @@ def main(argv: list[str] | None = None) -> int:
print(f"FEHLER: Bahnzusammenhang-SVG nicht erzeugt: {exc}", file=sys.stderr)
chain_file = None
chain_dot_file = None
if args.chain_graph:
try:
svg = render_chain_svg(elements, graph, csv_file.name)
except ValueError as exc:
print(f"WARNUNG: Chain-Graph: {exc} - nicht erzeugt.", file=sys.stderr)
if args.graphtype == "subgraph":
chain_dot_target = results / f"{stem}_chain.dot"
chain_dot_target.write_text(
render_chain_subgraph_dot(elements, graph, track, csv_file.name), encoding="utf-8"
)
chain_dot_file = chain_dot_target
if args.tosvg:
chain_svg_target = results / f"{stem}_chain.svg"
try:
dot_to_svg(chain_dot_file, chain_svg_target)
chain_file = chain_svg_target
except RuntimeError as exc:
print(f"FEHLER: Chain-Graph-SVG nicht erzeugt: {exc}", file=sys.stderr)
else:
chain_target = results / f"{stem}_chain.svg"
chain_target.write_text(svg, encoding="utf-8")
chain_file = chain_target
try:
if args.graphtype == "compound":
svg = render_chain_compound_svg(elements, graph, track, csv_file.name)
else:
svg = render_chain_svg(elements, graph, csv_file.name)
except ValueError as exc:
print(f"WARNUNG: Chain-Graph: {exc} - nicht erzeugt.", file=sys.stderr)
else:
chain_target = results / f"{stem}_chain.svg"
chain_target.write_text(svg, encoding="utf-8")
chain_file = chain_target
report(
graph,
@@ -3249,6 +3674,7 @@ def main(argv: list[str] | None = None) -> int:
bbs_file=bbs_file,
track_dot_file=track_dot_file,
track_svg_file=track_svg_file,
chain_dot_file=chain_dot_file,
chain_file=chain_file,
)
+639 -132
View File
@@ -28,9 +28,46 @@ doc/500573_Mubea/TRO_Identifikation_500573.md:
-----------------------------------------------------------------------
Gefaellestrecke, >= 2 Linien -> PinStore_Auto (ein Block je Gruppe)
Gefaellestrecke, einzelne Linie -> 1Sep
Strecke (angetrieben) -> Vario (je Streckensegment)
Strecke (angetrieben) -> Vario (je Separator, siehe unten)
Kreisel-Bahn -> 1Sep
Zwei Ausnahmen stufen einen Separator auf einer "Strecke" trotzdem als 1Sep
statt Vario ein (Review 2026-09-23/25, HundM05):
1. Alternierendes Muster vom Wiedereinschleusen auf die Kreisel-Bahn
rueckwaerts (in TrackIds-Reihenfolge, nur echte Kettenmitglieder - siehe
"eigenstaendige Vereinzeler" unten): der letzte Separator vor dem
Wiedereinschleusen ist ein reiner Vereinzeler ohne eigene VarioDrive-
Strecke (die Kreisel-Bahn treibt ab dort selbst an), der vorletzte wieder
ein eigener Vario-Antriebspunkt, der drittletzte wieder 1Sep usw. - eine
Streckengruppe ist baulich meist eine alternierende VF/GF-Kette
(angetriebener Abschnitt, Vereinzeler, unbetriebener Abschnitt,
Vereinzeler, ...), nicht nur am Ende. Gilt fuer jede Separatoranzahl >= 2
(z. B. VF_3: erster Separator 0041 = Vario, letzter 0040 = 1Sep); bei
genau einem Separator bleibt dieser Vario (einziger Antriebspunkt der
Strecke).
2. Eine "Strecke" (Streckengruppe) ist selbst oft eine gemischte VF/GF-Kette:
sie enthaelt sowohl angetriebene (VF, jede Antriebfahrtrichtung -
horizontal, Auf oder Ab zaehlt gleichermassen als VF) als auch unbetriebene
Gefaelle-Teilstuecke (GF) - siehe Merkmal "Antriebfahrtrichtung"/"VF_Winkel"
je TrackIds-Segment. Ein laut Regel 1 als Vario erwarteter Separator,
dessen unmittelbar vorangehendes Segment als "horizontal" (=keine aktive
Antriebsrichtung) gefuehrt wird, dabei aber tatsaechlich an Hoehe verliert
(Merkmal "Hoehe" der Anbauteile, > 30mm Gefaelle), sitzt auf einem
GF-Teilstueck und wird ebenfalls 1Sep (siehe _incoming_drop) - eine
unabhaengige Zusatzpruefung neben der Positions-Regel, kein Ersatz dafuer;
ein laut Regel 1 bereits 1Sep bleibt 1Sep. Achtung: diese Hoehen-/
Antriebsangaben sind bei mehreren Streckengruppen desselben Typs (z. B.
VF_2/VF_3/VF_4 bei HundM05) identisch und damit vermutlich unkorrigierte
CAD-Blockvorgaben, nicht echte Einzelmessungen - Regel 1 (Position in der
Kette) ist deshalb der robustere Nachweis, Regel 2 nur ein Zusatzfund fuer
Faelle mit echten, individuellen Messwerten (wie VF_1).
Separatoren, deren Zuordnung zwar auf eine Streckengruppe zeigt, die aber
nicht Teil von deren eigener TrackIds-Kette sind, zaehlen bei beiden Regeln
nicht mit - sie sind eigenstaendige Vereinzeler (plain 1Sep, siehe
derive_tros Schritt 2) und treiben das alternierende Muster nicht mit.
Ein Scanner am Separator aendert den Typ NICHT: in connect.ini der
Referenzanlage sind Separatoren mit Scanner ueberwiegend 1Sep (z. B.
"TRO104 = 1Sep | TRO 104 (1Sep)+Scanner"). 1Sep_SSCC ist dem Sonderfall mit
@@ -58,6 +95,7 @@ from __future__ import annotations
import argparse
import csv
import json
import math
import sys
from collections.abc import Callable
@@ -73,6 +111,7 @@ from material_flow import (
LANE_RIGHT,
ROTATION_CCW,
ROTATION_CW,
TRANSPORT_KINDS,
UNASSIGNED,
Element,
Graph,
@@ -93,6 +132,7 @@ from tro_catalog import (
STYLES,
TRO_CATALOG,
TroDefinition,
TroItem,
get_tro,
mermaid_classdefs,
)
@@ -113,23 +153,14 @@ from tro_overrides import (
# einem Separator) wird ein Speicherblock (PinStore_Auto) angenommen.
MIN_STORE_LANES = 2
# Toleranz (mm), innerhalb derer ein Separator als "sitzt an einem
# Ausschleuselement" gilt (siehe _find_switch_separators, Regel 1). Kalibriert
# an hundmM05_1.csv: die tatsaechlich am Uebergang montierten Separatoren
# liegen 250-350mm vom Element entfernt, die naechsten unbeteiligten (auf der
# eigenen Streckenmitte montierten) Separatoren erst ab ca. 885mm - die
# Grenze liegt bewusst in der Mitte dieser Luecke.
SWITCH_PROXIMITY_MM = 500.0
# Obergrenze (mm) fuer Regel 2 in _find_switch_separators (naechster freier
# Separator ohne Treffer aus Regel 1). Grosszuegiger als SWITCH_PROXIMITY_MM,
# aber bewusst NICHT unbegrenzt: an HundM05 (Review 2026-09-18) lagen die noch
# echten Weichen bis 1139mm vom Ausschleuselement entfernt - das deckt dieser
# Wert. Ohne diese Grenze hat sich Regel 2 an einer groesseren Anlage (Mubea,
# 500573) als Fehlerquelle erwiesen: uebrig gebliebene, tatsaechlich
# unbeteiligte Separatoren wurden dem naechsten noch freien Ausschleuselement
# zugeordnet, auch wenn das 5-9 Meter entfernt lag.
SWITCH_FALLBACK_MM = 1500.0
# Mindest-Hoehenverlust (mm) auf dem der Reihe nach vorangehenden TrackIds-
# Segment, damit ein Separator auf einer "Strecke" trotzdem als 1Sep statt
# Vario gilt (siehe _incoming_drop). Muss nur "horizontal" gefuehrte, aber
# tatsaechlich Gefaelle-Segmente erwischen (HundM05 Separator 0035: -104mm auf
# 3,8m), nicht das Rauschen eines echten flachen VF-Laufs (HundM05 Separator
# 0036: +26mm auf 0,5m). 30mm liegt bewusst weit unter dem echten Befund und
# weit ueber dem Rauschen.
GEFAELLE_DROP_MM = 30.0
GENERATED_MARKER = "<!-- erzeugt von lib/tro_flow.py -->"
@@ -281,7 +312,7 @@ def _default_resolve_host(sep: Element, findings: list[str], graph: Graph) -> st
def _find_switch_separators(
separators: list[Element], elements: list[Element], tol_mm: float
separators: list[Element], elements: list[Element]
) -> dict[str, tuple[Element, float]]:
"""Separator-Id -> (Ausschleuselement der Weiche, Abstand mm).
@@ -296,79 +327,160 @@ def _find_switch_separators(
zusaetzliches Bauteil des TROs (siehe derive_tros/scanners_for) - viele
Separatoren tragen einen Scanner, ohne eine Weiche zu sein.
Zwei Durchlaeufe, um trotzdem jedem Ausschleuselement genau einen
Separator zuzuordnen:
Frueher ueber eine reine Distanzschwelle ermittelt (<=500mm, mit einer
grosszuegigeren Fallback-Schwelle bis 1500mm fuer weiter entfernte
Faelle). Das erwies sich als nicht plausibel genug in beide Richtungen
(Review 2026-09-25): die Fallback-Schwelle paarte an Mubea wiederholt
voellig unbeteiligte Separatoren mit dem naechstgelegenen noch freien
Ausschleuselement, ohne echten Bezug zwischen beiden - vor Ort geprueft,
keine Weiche an der Stelle. Eine reine Nahdistanz ohne Fallback haette
umgekehrt an HundM05 alle 5 echten Weichen verfehlt, die dort
gleichmaessig bei 1013-1016mm liegen (Review 2026-09-18) - Distanz allein
ist also weder an Mubea noch an HundM05 ein verlaessliches Kriterium.
1. Geometrische Naehe (Distanz <= tol_mm). HundM05 hat an seinen
Kreisel<->Strecke-Uebergaengen kein eigenes "ILS Weiche"-Bauteil
(anders als ein Kreisel<->Kreisel-Beruehrpunkt, siehe Mubea) - das von
dxfmakros synthetisierte Ausschleuselement (Kreisel -> Strecke, am
Streckenanfang) uebernimmt dort faktisch die Rolle der Weiche. Sitzt
ein Separator direkt daneben, gehoert er zu dieser Weiche - unabhaengig
davon, ob seine CSV-Spalte "Zuordnung" auf die Kreisel-Bahn oder auf
die Strecke zeigt (in den Daten kommt beides vor, siehe derive_tros).
2. Naechstgelegener freier Separator je noch unbelegtem Ausschleuselement,
jedes Paar nur einmal vergeben (kleinste Distanz zuerst, hoechstens
SWITCH_FALLBACK_MM), und nur auf einer Bahn, die bereits durch Regel 1
einen bestaetigten Treffer hat (eine Bahn ohne jeden nahen Treffer hat
vermutlich gar keine Weiche, z. B. Kreisel1 Bahn R bei HundM05). Grund
fuer den zweiten Durchlauf (HundM05, Review 2026-09-18): an einem
Kreisel mit 5 Abzweigen lagen die tatsaechlichen Schalt-Separatoren pro
Abzweig unterschiedlich weit vom Uebergabeelement entfernt (418-1139mm
in der damaligen Registrierung) - eine einzelne mm-Schwelle (Regel 1)
trifft nicht alle. SWITCH_FALLBACK_MM begrenzt das trotzdem: ohne
Obergrenze hat Regel 2 an einer groesseren Anlage (Mubea) uebrig
gebliebene, tatsaechlich unbeteiligte Separatoren dem naechsten noch
freien Ausschleuselement zugeordnet, auch wenn das 5-9 Meter entfernt
lag - siehe SWITCH_FALLBACK_MM.
Stattdessen wird die im CSV-Export bereits vorhandene topologische
Nachbarschaft verwendet (Spalte "Nachbarn", Element.neighbours): dxfmakros
traegt dort den tatsaechlich anliegenden Nachbarn ein (CAD-Topologie),
nicht nur den geometrisch naechsten - an HundM05 steht das echte
Ausschleuselement trotz 1013-1016mm Abstand direkt in den Nachbarn des
Schalt-Separators, an Mubea NICHT in den Nachbarn der zu Unrecht per
Fallback getroffenen Separatoren. Ein Separator ist eine Weiche, wenn:
1. ein Ausschleuselement unter seinen Nachbarn steht, UND
2. dieses Ausschleuselement zur selben Kreisel-Bahn gehoert wie der
Separator (gleiche "Zuordnung", z. B. beide auf Kreisel "0002" -
Element.track_ids des betreffenden Kreisels enthaelt das
Ausschleuselement). Das ist nicht nur eine Absicherung gegen einen
ungenau exportierten "Nachbarn"-Eintrag, sondern noetig: ein
Ausschleuselement markiert nicht nur den Kreisel->Strecke-Uebergang,
sondern an einem PinStore_Auto-Speicherblock auch den Ablauf jeder
einzelnen Schwerkraftlinie zum gemeinsamen Sammler - jeder
Linien-Separator hat dort ein Ausschleuselement als CSV-Nachbarn, ohne
damit eine echte Weiche zu sein (Review 2026-09-25, Mubea: 20
Speicherlinien faelschlich als 1Sep1Swi einsortiert, weil ihre
Kreisel-Zuordnung fehlt und Kriterium 2 dafuer faelschlich nicht griff).
Ist die Kreisel-Zuordnung des Separators nicht aufloesbar (er sitzt gar
nicht auf einer Kreisel-Bahn), ist er deshalb KEIN Weichenkandidat -
Kriterium 1 allein reicht nicht.
Die Distanz im Rueckgabewert ist nur noch Beiwerk fuer die Doku ("XXXmm
von Ausschleuselement YYYY entfernt"), kein Kriterium mehr.
"""
transfers = [e for e in elements if e.kind == "Ausschleus"]
transfers = {e.teile_id: e for e in elements if e.kind == "Ausschleus"}
kreisel_by_id = {e.teile_id: e for e in elements if e.kind == "Kreisel"}
result: dict[str, tuple[Element, float]] = {}
for sep in separators:
if sep.x is None or sep.y is None:
sep_kreisel_id = sep.assignment.split("-", 1)[0] if sep.assignment else ""
kreisel = kreisel_by_id.get(sep_kreisel_id)
if kreisel is None:
continue
best: tuple[float, Element] | None = None
for transfer in transfers:
if transfer.x is None or transfer.y is None:
for base, _lane in sep.neighbours:
transfer = transfers.get(base)
if transfer is None:
continue
d = math.hypot(transfer.x - sep.x, transfer.y - sep.y)
if d <= tol_mm and (best is None or d < best[0]):
best = (d, transfer)
if best is not None:
result[sep.teile_id] = (best[1], best[0])
# Bahnen mit mindestens einem durch Regel 1 bestaetigten Treffer - nur
# dort darf Regel 2 ueberhaupt suchen (siehe Docstring).
switch_bahnen = {sep.assignment for sep in separators
if sep.teile_id in result and sep.assignment}
used_transfers = {hit[0].teile_id for hit in result.values()}
remaining_seps = [s for s in separators if s.teile_id not in result
and s.x is not None and s.y is not None
and s.assignment in switch_bahnen]
candidates: list[tuple[float, Element, Element]] = []
for sep in remaining_seps:
for transfer in transfers:
if transfer.teile_id in used_transfers or transfer.x is None or transfer.y is None:
if transfer.teile_id not in kreisel.track_ids:
continue
d = math.hypot(transfer.x - sep.x, transfer.y - sep.y)
if d <= SWITCH_FALLBACK_MM:
candidates.append((d, sep, transfer))
candidates.sort(key=lambda c: c[0])
claimed_seps: set[str] = set()
for d, sep, transfer in candidates:
if sep.teile_id in claimed_seps or transfer.teile_id in used_transfers:
continue
result[sep.teile_id] = (transfer, d)
claimed_seps.add(sep.teile_id)
used_transfers.add(transfer.teile_id)
d = (
math.hypot(transfer.x - sep.x, transfer.y - sep.y)
if None not in (sep.x, sep.y, transfer.x, transfer.y)
else 0.0
)
result[sep.teile_id] = (transfer, d)
break
return result
def _incoming_drop(
host: Element, sep_id: str, elements_by_id: dict[str, Element]
) -> tuple[str | None, float | None]:
"""Antriebfahrtrichtung und Hoehendifferenz (mm) des TrackIds-Segments,
das unmittelbar vor sep_id in host.track_ids liegt.
host ist eine "Strecke" (Streckengruppe); ihr Merkmal "Antriebfahrtrichtung"
ist eine Kommaliste, ein Eintrag je Segment zwischen zwei aufeinander-
folgenden TrackIds-Elementen (siehe derive_tros/Docstring-Tabelle). Fehlt
sep_id in der Kette, ist es das erste Element (kein vorangehendes Segment)
oder fehlt einem der beiden Enden das Merkmal "Hoehe" (z. B. Aus-/
Einschleuselemente tragen keins), liefert die Funktion (None, None) - der
Aufrufer behandelt das als "kein Gefaelle-Nachweis", bleibt also bei Vario.
"""
chain = host.track_ids
if sep_id not in chain:
return None, None
idx = chain.index(sep_id)
if idx == 0:
return None, None
prev = elements_by_id.get(chain[idx - 1])
cur = elements_by_id.get(sep_id)
if prev is None or cur is None:
return None, None
prev_h = _as_float(_merkmal(prev.merkmale, "Höhe", "Hoehe"))
cur_h = _as_float(_merkmal(cur.merkmale, "Höhe", "Hoehe"))
if prev_h is None or cur_h is None:
return None, None
drive_list = [d.strip() for d in (host.drive_dir or "").split(",")]
drive = drive_list[idx - 1] if idx - 1 < len(drive_list) else None
return drive, cur_h - prev_h
def find_sibling_raw_json(csv_path: Path) -> Path | None:
"""<CSV-Stamm, klein geschrieben>_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],
@@ -377,6 +489,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.
@@ -386,6 +499,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))
@@ -396,13 +515,33 @@ def derive_tros(
scanners = sorted(
(e for e in elements if e.kind == "Scanner"), key=lambda e: e.teile_id
)
elements_by_id = {e.teile_id: e for e in elements}
# Separatoren, die bereits ein echtes VF-Kettenmitglied ihrer EIGENEN
# Zuordnung-Streckengruppe sind (siehe Schritt 2/_incoming_drop unten),
# werden von der Weichenerkennung ausgenommen: Naehe zum eigenen Aus-/
# Einschleuselement ist bei ihnen der NORMALE, erwartete Aufbau jeder
# angetriebenen Streckengruppe (TrackIds = [Aus-, 1x Separator, Ein-
# schleuselement], siehe Mubea VF_1-VF_3), keine Weichenevidenz. Ohne
# diesen Ausschluss faengt Regel 1 faelschlich genau diesen Separator ein
# (Mubea 0005, 163mm - der engste, staerkste Treffer ueberhaupt, aber kein
# Weichen-Indiz - Review 2026-09-24).
own_vf_separators = {
sep.teile_id
for sep in separators
if sep.assignment
and (host := elements_by_id.get(sep.assignment))
and host.kind == "Strecke"
and sep.teile_id in host.track_ids
}
switch_candidates = [s for s in separators if s.teile_id not in own_vf_separators]
# Separatoren an einem Ausschleuselement (siehe _find_switch_separators)
# werden VOR der normalen Host-Aufloesung herausgenommen: sie bilden gleich
# unten (3b) ihre eigene 1Sep1Swi-Uebergabestelle und duerfen nicht zugleich
# als gewoehnlicher Separator ihres Zuordnung-Hosts (Kreisel-Bahn oder
# Strecke - beides kommt vor) mitgezaehlt werden.
switch_hits = _find_switch_separators(separators, elements, SWITCH_PROXIMITY_MM)
switch_hits = _find_switch_separators(switch_candidates, elements)
# Separator -> Host-Knoten
sep_host: dict[str, str] = {}
@@ -446,6 +585,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
@@ -486,23 +643,132 @@ def derive_tros(
)
)
# 2) Angetriebene Strecken -> je Segment ein Vario
# 2) Angetriebene Strecken -> je SEPARATOR ein eigener Vario-TRO. Katalog
# (TRO_CATALOG in tro_catalog.py) definiert Vario als genau 1x Separator +
# 1x VarioDrive - ein Vario-TRO ist die Paarung eines einzelnen VF-
# Teilstuecks mit dem einen Separator, den es antreibt, kein Sammel-TRO
# fuer die ganze Streckengruppe (Review 2026-09-21/24). Mehrere
# Separatoren auf demselben Streckensegment sind deshalb mehrere
# eigenstaendige Uebergabestellen. Ein Segment ganz ohne Separator bildet
# folgerichtig KEIN eigenes TRO (kein Separator zum Paaren) - anders als
# bisher wird es nicht mehr als Platzhalter-Vario mit leerer
# Separatorliste gefuehrt.
for node_id in sorted(n for n, node in graph.nodes.items() if node.kind == "Strecke"):
seps = seps_on([node_id])
if not seps:
continue
element = graph.nodes[node_id].element
drive = element.drive_dir or "unbekannt"
tros.append(
Tro(
tro_id="",
type_name="Vario",
label=element.name,
hosts=[node_id],
separators=seps,
scanners=scanners_for(seps),
reason=f"angetriebenes Streckensegment (Antriebfahrtrichtung: {drive})",
confidence="hoch",
reason = f"angetriebenes Streckensegment (Antriebfahrtrichtung: {drive})"
# Alternierendes Muster vom Wiedereinschleusen auf die Kreisel-Bahn
# rueckwaerts (in echter TrackIds-Reihenfolge, nicht sortierte
# Teile-Id): eine Streckengruppe ist baulich meist eine alternierende
# VF/GF-Kette, deshalb wechselt auch der TRO-Typ ab dem letzten
# Separator (naechster Nachbar der Kreisel-Bahn, reiner Vereinzeler
# ohne eigene VarioDrive-Strecke - die Kreisel-Bahn treibt ab dort
# selbst an) rueckwaerts durch: letzter = 1Sep, davor Vario, davor
# wieder 1Sep usw. Bestaetigt am Beispiel VF_3 (Review 2026-09-23):
# TRO11 (erster Separator 0041, direkt nach dem Ausschleusen) ist
# Vario, TRO10 (zweiter/letzter Separator 0040, vor dem
# Wiedereinschleusen) ist 1Sep; siehe auch VF_1 (0034, letzter von
# vieren). Fuer genau 2 Separatoren identisch zur frueheren "nur der
# letzte wird umgestuft"-Regel; ab 3 weicht es ab (Review 2026-09-25)
# - vorher blieben alle bis auf den letzten pauschal Vario, obwohl die
# Kette baulich mehrfach zwischen VF und GF wechselt. Bei genau einem
# Separator gibt es kein "erster vs. letzter", der bleibt Vario
# (einziger Antriebspunkt der Strecke).
seps_set = set(seps)
track_seps = [tid for tid in element.track_ids if tid in seps_set]
track_seps_set = set(track_seps)
alt_type: dict[str, str] = {}
alt_offset: dict[str, int] = {}
if len(track_seps) > 1:
for offset, tid in enumerate(reversed(track_seps)):
alt_offset[tid] = offset
alt_type[tid] = "1Sep" if offset % 2 == 0 else "Vario"
for sep_id in seps:
if sep_id not in track_seps_set:
# Zuordnung zeigt zwar auf diese Streckengruppe, der Separator
# ist aber nicht Teil von deren eigener TrackIds-Kette - kein
# VF-Teilstueck treibt ihn an, also plain 1Sep statt Vario
# (Mubea VF_1-VF_3 vor Ort bestaetigt, Review 2026-09-24: pro
# Streckengruppe genau 1 echter Kettenseparator, alle
# uebrigen Zuordnung-Treffer sind eigenstaendige Vereinzeler).
tros.append(
Tro(
tro_id="",
type_name="1Sep",
label=element.name,
hosts=[node_id],
separators=[sep_id],
scanners=scanners_for([sep_id]),
reason=(
f"Separator {sep_id}: Zuordnung zeigt auf "
f"{element.name}, ist aber nicht Teil von dessen "
f"TrackIds-Kette - kein eigenes VF treibt ihn an"
),
confidence="mittel",
)
)
findings.append(
f"{element.name} Separator {sep_id}: Zuordnung ohne "
f"TrackIds-Mitgliedschaft - als 1Sep statt Vario eingestuft"
)
continue
type_name = alt_type.get(sep_id, "Vario")
sep_reason = reason
confidence = "hoch"
if type_name == "1Sep":
confidence = "mittel"
sep_reason = (
f"alternierendes VF/GF-Muster vom Wiedereinschleusen von "
f"{element.name} rueckwaerts ({alt_offset[sep_id] + 1}. "
f"Separator von hinten) - unbetriebener Abschnitt ohne "
f"eigenes VarioDrive (Review 2026-09-23/25)"
)
findings.append(
f"{element.name} Separator {sep_id}: alternierendes Muster "
f"({alt_offset[sep_id] + 1}. von hinten) - als 1Sep statt "
f"Vario eingestuft"
)
# Hoehen-/Antriebsdaten des vorangehenden Segments (Strecke/
# Gefaellestrecke) bleiben eine unabhaengige Zusatzpruefung (siehe
# Docstring oben, Regel 2): sie kann eine laut Muster als Vario
# erwartete Position auf 1Sep umstufen, wenn dort tatsaechlich ein
# unbetriebenes Gefaelle gemessen wurde - nicht umgekehrt, ein
# bereits per Muster bestimmtes 1Sep bleibt 1Sep.
if type_name == "Vario":
seg_drive, seg_drop = _incoming_drop(element, sep_id, elements_by_id)
if seg_drive == "horizontal" and seg_drop is not None and seg_drop <= -GEFAELLE_DROP_MM:
type_name = "1Sep"
confidence = "mittel"
sep_reason = (
f"Separator {sep_id} liegt auf einem unbetriebenen "
f"Gefaelle-Teilstueck der Streckengruppe {element.name} "
f"(Hoehe faellt um {-seg_drop:.0f}mm, keine aktive "
f"Antriebfahrtrichtung) - kein Vario trotz Host 'Strecke' "
f"und trotz alternierendem Muster"
)
findings.append(
f"{element.name} Separator {sep_id}: {-seg_drop:.0f}mm Gefaelle "
f"ohne Antrieb auf dem vorangehenden Segment - als 1Sep statt "
f"Vario eingestuft (widerspricht alternierendem Muster)"
)
tros.append(
Tro(
tro_id="",
type_name=type_name,
label=element.name,
hosts=[node_id],
separators=[sep_id],
scanners=scanners_for([sep_id]),
reason=sep_reason,
confidence=confidence,
)
)
)
# 3) Separatoren auf Kreisel-Bahnen -> je Separator eine Uebergabestelle
for sep_id in sorted(
@@ -624,26 +890,35 @@ def derive_tros(
# Typnamen fuer Systemgrenz-Knoten (siehe derive_boundaries) - bewusst NICHT
# in TRO_CATALOG: eine BTMT Be-/Entladestation ist kein Steuerobjekt mit
# eigenem FB-Baustein, nur der tatsaechliche Materialein-/ausgang.
# in TRO_CATALOG: eine Be-/Entladestation (BTMT oder generisches Eingang/
# Ausgang-Symbol) ist kein Steuerobjekt mit eigenem FB-Baustein, nur der
# tatsaechliche Materialein-/ausgang.
BOUNDARY_ENTRY = "Materialeingang"
BOUNDARY_EXIT = "Materialausgang"
BOUNDARY_TYPES = (BOUNDARY_ENTRY, BOUNDARY_EXIT)
def derive_boundaries(elements: list[Element], graph: Graph) -> list[Tro]:
"""BTMT Be-/Entladestationen als sichtbare Systemgrenz-Knoten.
"""Be-/Entladestationen als sichtbare Systemgrenz-Knoten.
Ohne diese Knoten "verschwindet" der Materialfluss stillschweigend an der
Station: link_tros() registriert eine Kante nur, wenn die Traversierung
einen Knoten erreicht, der ein TRO-Host ist. Eine BTMT-Station ist nie
ein TRO-Host (sie hat keinen Separator), also wird sie beim Durchlaufen
als passiver Zwischenknoten behandelt - und da eine Entladung (Senke)
keine ausgehende Kante mehr hat, bricht die Suche dort einfach ab, ohne
eine Kante einzutragen. Eine Beladung (Quelle) wird umgekehrt nie als
Ausgangspunkt einer Traversierung betrachtet, weil kein TRO dort beginnt.
Damit fehlte bislang genau der Teil des Flusses, den man am Anfang oder
Ende eines Durchlaufs durch die Anlage sieht.
einen Knoten erreicht, der ein TRO-Host ist. Eine Station ist nie ein
TRO-Host (sie hat keinen Separator), also wird sie beim Durchlaufen als
passiver Zwischenknoten behandelt - und da eine Entladung/ein Ausgang
(Senke) keine ausgehende Kante mehr hat, bricht die Suche dort einfach
ab, ohne eine Kante einzutragen. Eine Beladung/ein Eingang (Quelle) wird
umgekehrt nie als Ausgangspunkt einer Traversierung betrachtet, weil kein
TRO dort beginnt. Damit fehlte bislang genau der Teil des Flusses, den
man am Anfang oder Ende eines Durchlaufs durch die Anlage sieht - die
Anlage laeuft als Kreislauf, diese Stationen sind ihre einzigen
Materialquellen/-senken (siehe auch material_flow.py --chain-graph).
Erfasst sowohl die konkrete BTMT-Station (Beladung/Entladung) als auch das
generische Eingang/Ausgang-Symbol (element.is_station, STATION_KINDS in
material_flow.py) - ein durch eine deckungsgleiche BTMT-Station
verschlucktes Symbol taucht dank des graph.nodes-Filters unten trotzdem
nicht doppelt auf (_merge_station_symbols(), siehe build_graph()).
Kein eigener TRO-Typ (keine FB_ILS_MTRO_*-Zuordnung, kein Eintrag in
TRO_CATALOG) - nur ein sichtbarer Marker fuer den echten Anfang/das echte
@@ -653,7 +928,7 @@ def derive_boundaries(elements: list[Element], graph: Graph) -> list[Tro]:
"""
boundaries: list[Tro] = []
for element in elements:
if element.kind not in ("Beladung", "Entladung"):
if not element.is_station:
continue
if element.teile_id not in graph.nodes:
continue
@@ -665,7 +940,7 @@ def derive_boundaries(elements: list[Element], graph: Graph) -> list[Tro]:
hosts=[element.teile_id],
separators=[],
scanners=[],
reason="Systemgrenze (BTMT Be-/Entladestation) - kein TRO, "
reason="Systemgrenze (Be-/Entladestation) - kein TRO, "
"nur Materialein-/ausgang",
confidence="hoch",
)
@@ -714,8 +989,32 @@ def _bahn_richtung(node) -> tuple[float, float] | None:
return achse if mit_achse else (-achse[0], -achse[1])
def _tro_bahn(tro: Tro, graph: Graph, sep_assignment: dict[str, str]) -> str | None:
"""Kreisel-Bahn, auf der ein TRO physisch sitzt - fuer die Positions-
Verkettung in link_tros(). Meist einfach der TRO-Host (wenn er direkt auf
der Bahn liegt, z. B. ein gewoehnlicher Kreisel-Separator). Bei einem
Weichen-TRO zeigt der Host dagegen auf die gefuetterte Strecke, nicht auf
die Bahn (siehe derive_tros Schritt 3b) - dort zaehlt stattdessen die
Zuordnung des eigenen Separators, falls die auf eine Kreisel-Bahn zeigt.
Zeigt sie stattdessen auf eine Strecke (z. B. HundM05 Separator 0016/0018,
die auf ihrer eigenen Strecke montiert sind, nicht auf der Kreisel-Bahn),
gibt es keine Bahn-Zugehoerigkeit - der TRO gehoert dann nicht in die
Verkettung.
"""
for host in tro.hosts:
node = graph.nodes.get(host)
if node and node.kind == "Kreisel":
return host
for sep_id in tro.separators:
bahn = sep_assignment.get(sep_id)
node = graph.nodes.get(bahn or "")
if node and node.kind == "Kreisel":
return bahn
return None
def link_tros(
graph: Graph, tros: list[Tro]
graph: Graph, tros: list[Tro], elements: list[Element]
) -> tuple[list[tuple[str, str]], dict[tuple[str, str], str]]:
"""
TROs verknuepfen.
@@ -730,6 +1029,7 @@ def link_tros(
for tro in tros:
for host in tro.hosts:
owners.setdefault(host, []).append(tro.tro_id)
by_id = {t.tro_id: t for t in tros}
forward: dict[str, list[tuple[str, str]]] = {}
for edge in graph.edges:
@@ -758,21 +1058,179 @@ def link_tros(
continue
queue.extend(forward.get(node, []))
# Mehrere TROs auf derselben Kreisel-Bahn liegen HINTEREINANDER, nicht
# nebeneinander: das Material laeuft die Bahn in Foerderrichtung ab. Sie
# werden daher der Reihe nach verkettet statt nur als "gleiche Bahn"
# markiert - sonst fehlt z. B. die Kante von der einen Sperre zur naechsten.
verkettet: set[str] = set()
for host, ids in owners.items():
if len(ids) < 2:
# verkettete_paare (ungerichtet) unterdrueckt die "gleiche Bahn"-Punktlinie
# weiter unten - dafuer reicht Ortsgleichheit unabhaengig von der Richtung.
# directed_paare (gerichtet) markiert dagegen genau die Kanten, die durch
# eine dieser Regeln bewusst gesetzt wurden - fuer den Weichen-
# Redundanzfilter weiter unten muss das richtungsscharf sein: TRO12->TRO01
# (echte Umlaufkette) darf die entgegengesetzte TRO01->TRO12 (grober,
# redundanter Contraction-Walk-Fund) NICHT mitdecken, nur weil es dieselben
# zwei TROs sind (Review 2026-09-21, Bug beim ersten Versuch dieses Filters).
verkettete_paare: set[frozenset[str]] = set()
directed_paare: set[tuple[str, str]] = set()
# 1. Ground-Truth-Umlaufreihenfolge je Kreisel aus Element.track_ids
# (dieselbe nach Drehrichtung-Winkel sortierte Liste, die auch --track/
# build_track_graph verwendet - siehe doc/track.md). Laeuft ZUERST, damit
# der Positions-Fallback weiter unten weiss, welche Kreisel damit schon
# praezise abgedeckt sind. Ersetzt fuer diese Kreisel die aus dem groben
# Materialfluss-Graph kontrahierten "Nabe -> jeder erreichbare Zweig"-
# Kanten weiter unten: die kennen keine Reihenfolge und zeigen daher u. U.
# rueckwaerts (Review 2026-09-21 - TRO11 zeigte auf alle 5 Abzweige
# gleichrangig, obwohl es in der echten Umlaufreihenfolge der LETZTE
# Separator vor der Entladung ist).
elem_owner: dict[str, str] = {}
for tro in tros:
if tro.type_name in BOUNDARY_TYPES:
elem_owner.setdefault(tro.hosts[0], tro.tro_id)
else:
for sep_id in tro.separators:
elem_owner.setdefault(sep_id, tro.tro_id)
def _add_chain_pair(a: str, b: str) -> None:
if a == b:
return
# Eine Systemgrenze ist nur in EINE Richtung gueltig (siehe
# derive_boundaries: Beladung ist Quelle, nie Ziel; Entladung ist
# Senke, nie Quelle) - unabhaengig davon, wie TrackIds sie in der
# geschlossenen Umlaufliste nacheinander auffuehrt.
if by_id[a].type_name == BOUNDARY_EXIT or by_id[b].type_name == BOUNDARY_ENTRY:
return
pair = (a, b)
edges.add(pair)
edge_kinds.setdefault(pair, "bahn")
verkettete_paare.add(frozenset(pair))
directed_paare.add(pair)
trackids_pairs.add(pair)
trackids_pairs: set[tuple[str, str]] = set()
covered_kreisel_bases: set[str] = set()
# 0. Reihenfolge der Vario-TROs INNERHALB einer Strecke, aus derselben
# Element.track_ids-Quelle wie der Kreisel-Umlauf unten, hier aber linear
# (AS -> SEP.. -> ES, keine geschlossene Schleife) - siehe build_track_graph()
# in material_flow.py, dessen Richtungskorrektur hier repliziert wird. Seit
# derive_tros Schritt 2 einen eigenen Vario-TRO je Separator anlegt (Review
# 2026-09-21: "jedes Vario-TRO soll genau 1 Separator haben"), muessen
# mehrere Varios auf derselben Strecke in echter Reihenfolge verkettet
# werden - strecke_first_owned/strecke_last_owned merken sich den jeweils
# ersten/letzten fuer die Anschluss-Kanten an Weiche bzw. Wiedereinschleusung.
elements_by_id = {e.teile_id: e for e in elements}
strecke_first_owned: dict[str, str] = {}
strecke_last_owned: dict[str, str] = {}
for el in elements:
if el.kind not in TRANSPORT_KINDS:
continue
node = graph.nodes.get(host)
if node is None or node.kind != "Kreisel" or not node.lane:
chain = list(el.track_ids)
if len(chain) >= 2:
first_el, last_el = elements_by_id.get(chain[0]), elements_by_id.get(chain[-1])
if ((first_el and first_el.kind == "Einschleus")
or (last_el and last_el.kind == "Ausschleus")):
chain = list(reversed(chain))
# Ein Separator, der (Zuordnung) zwar auf dieser Strecke steht, aber
# selbst eine Weiche ist (z. B. HundM05 Separator 0016 auf VF_5 - siehe
# derive_tros Schritt 3b), gehoert NICHT in die Vario-Kette der Strecke:
# er ist der Kreisel-Bahn zugehoerig und dort schon (Schritt 1 unten)
# verkettet. Ohne diesen Ausschluss wuerde er faelschlich als erster
# oder letzter "Vario" gelten (Review 2026-09-21).
owned = [
elem_owner[tid] for tid in chain if tid in elem_owner
and not ((d := by_id[elem_owner[tid]].definition()) and d.has_item(TroItem.SWITCH))
]
if not owned:
# Segment ganz ohne Separator (derive_tros Schritt 2: leere
# Separatorliste) - der einzelne Vario-TRO dieses Hosts bleibt
# trotzdem Anschlusspunkt fuer Weiche/Wiedereinschleusung.
solo = [tid for tid in owners.get(el.teile_id, [])
if by_id[tid].type_name == "Vario"]
if len(solo) == 1:
owned = solo
for a, b in zip(owned, owned[1:]):
_add_chain_pair(a, b)
if owned:
strecke_first_owned[el.teile_id] = owned[0]
strecke_last_owned[el.teile_id] = owned[-1]
# Einschleuselement-Id -> das Vario-TRO, das dort real wieder einmuendet
# (der LETZTE Vario-TRO der Strecke in echter Reihenfolge, nicht irgendein
# beliebiger - Review 2026-09-21).
einschleus_vario: dict[str, str] = {}
for el in elements:
if el.kind != "Einschleus":
continue
strecke_id = str(_merkmal(el.merkmale, "Strecke", "Förderer") or "")
vario_id = strecke_last_owned.get(strecke_id)
if vario_id:
einschleus_vario[el.teile_id] = vario_id
for el in elements:
if not el.is_circle or len(el.track_ids) < 2:
continue
chain = list(el.track_ids)
if el.rotation in (ROTATION_CW, ROTATION_CCW):
chain.append(chain[0]) # Umlauf schliessen
# Hauptkette: Separatoren/Systemgrenzen in Reihenfolge, Aus-/
# Einschleuselemente bewusst uebersprungen (unowned) - das beschreibt
# den BYPASS fuer Verkehr, der NICHT abzweigt (z. B. TRO01 -> TRO03,
# wenn dazwischen niemand abbiegt).
owned = [(i, elem_owner[tid]) for i, tid in enumerate(chain) if tid in elem_owner]
for (_, a), (_, b) in zip(owned, owned[1:]):
_add_chain_pair(a, b)
if owned:
covered_kreisel_bases.add(el.teile_id)
# Zusaetzlich der reale Wiedereintritt eines abgezweigten Vario-
# Segments: vom Einschleuselement zum naechsten Halt DANACH in
# derselben Kette - ersetzt fuer dieses Vario-TRO den groben "Strecke
# -> Bahn"-Kontraktionsfund (der pauschal auf DEN EINEN TRO zeigt, der
# zufaellig noch die Kreisel-Bahn als Host hat, unabhaengig davon,
# welche Strecke tatsaechlich zurueckkehrt - Review 2026-09-21: "warum
# zeigen alle Varios auf TRO12"). Unabhaengig von der Hauptkette oben,
# die an dieser Stelle bewusst durchlaeuft (Bypass, siehe dort).
for i, tid in enumerate(chain):
vario_id = einschleus_vario.get(tid)
if vario_id is None:
continue
nxt = next((o for j, o in owned if j > i), None)
if nxt is None and owned:
nxt = owned[0][1] # Umlauf geschlossen - wickelt um
if nxt is not None and nxt != vario_id:
_add_chain_pair(vario_id, nxt)
# Nur TROs, fuer die die Umlaufkette tatsaechlich einen praezisen
# Nachfolger gefunden hat (precise_sources) - ein TRO ohne brauchbare
# TrackIds-Abdeckung behaelt seine bisherigen (ungenaueren, aber nicht
# leeren) Kanten aus dem Contraction-Walk als Rueckfall. Angewendet wird
# der Filter erst ganz unten, nachdem auch die Weichen-Kanten feststehen.
precise_sources = {a for a, _ in trackids_pairs}
# 2. Positions-Fallback: mehrere TROs auf derselben Kreisel-Bahn liegen
# HINTEREINANDER, nicht nebeneinander - der Reihe nach verkettet statt nur
# als "gleiche Bahn" markiert. NUR fuer Kreisel, die Schritt 1 NICHT schon
# praezise abgedeckt hat (covered_kreisel_bases) - sonst gruppiert dieser
# Fallback rein nach CSV-Spalte "Zuordnung" und uebersieht z. B. bei
# HundM05 die Separatoren 0016/0018 (auf "0005" zugeordnet, nicht auf der
# Bahn) und zieht faelschlich eine direkte Kante 0014->0020 an ihnen
# vorbei (Review 2026-09-21).
sep_assignment = {e.teile_id: e.assignment for e in elements
if e.kind == "Separator" and e.assignment}
bahn_gruppen: dict[str, list[Tro]] = {}
for tro in tros:
bahn = _tro_bahn(tro, graph, sep_assignment)
if bahn and bahn.split("-")[0] not in covered_kreisel_bases:
bahn_gruppen.setdefault(bahn, []).append(tro)
for bahn, gruppe in bahn_gruppen.items():
if len(gruppe) < 2:
continue
node = graph.nodes[bahn]
if not node.lane:
continue
richtung = _bahn_richtung(node)
if richtung is None:
continue
auf_bahn = [t for t in tros if t.tro_id in ids and t.position]
auf_bahn = [t for t in gruppe if t.position]
if len(auf_bahn) < 2:
continue
# Projektion auf die Foerderrichtung: kleinster Wert kommt zuerst
@@ -782,18 +1240,60 @@ def link_tros(
pair = (vorher.tro_id, nachher.tro_id)
edges.add(pair)
edge_kinds.setdefault(pair, "bahn")
verkettet.update(pair)
verkettete_paare.add(frozenset(pair))
directed_paare.add(pair)
# gleicher Host = physisch dieselbe Strecke/Bahn. Verkettete Bahnen sind
# oben schon gerichtet verbunden und brauchen die ungerichtete Notiz nicht.
# Ein Weichen-TRO auf derselben Strecke wie ein Vario-TRO ist keine blosse
# Ortsgleichheit, sondern der zweite der beiden moeglichen Wege an der
# Weiche: entweder weiter auf der Kreisel-Bahn zum naechsten Separator
# (verkettete_paare oben) oder durch die Weiche hindurch in dieses Segment
# - konkret in dessen ERSTEN Vario-TRO (strecke_first_owned oben; bei
# mehreren Separatoren auf derselben Strecke sind die weiteren bereits
# ueber die Strecken-Kette oben an den ersten angeschlossen). Gerichtete
# Kante statt "gleiche Bahn"-Punktlinie, mit dem bereits vorhandenen
# "weiche"-Kantenstil (orange, Label "Weiche"). Review 2026-09-21.
for tro in tros:
definition = tro.definition()
if not (definition and definition.has_item(TroItem.SWITCH)) or len(tro.hosts) != 1:
continue
ziel_id = strecke_first_owned.get(tro.hosts[0])
if ziel_id is None or ziel_id == tro.tro_id:
continue
pair = (tro.tro_id, ziel_id)
edges.add(pair)
edge_kinds.setdefault(pair, "weiche")
verkettete_paare.add(frozenset(pair))
directed_paare.add(pair)
# Jedes TRO, fuer das oben ein praeziser Nachfolger in der Umlaufkette
# gefunden wurde (precise_sources), hat damit seine gesamte reale
# Weiterfuehrung bereits vollstaendig beschrieben (Bahn-Fortsetzung und/
# oder Weiche). Jede WEITERE von ihm ausgehende Kante stammt aus dem
# groben Contraction-Walk ganz oben (ueber denselben Host wie ein anderes
# TRO - z. B. TRO01 -> TRO12 zusaetzlich zu TRO01 -> TRO02, oder TRO02 ->
# TRO12 zusaetzlich zu TRO02 -> TRO03) und ist redundant bzw. schlicht
# falsch: das jeweilige Ziel-TRO berichtet dieselbe Weiterfuehrung bereits
# selbst, praeziser. Review 2026-09-21: erst "jede Weiche hat 3 statt 2
# ausgehende Kanten", dann "warum zeigen alle Varios auf TRO12" - beides
# derselbe Fehler, hier fuer jedes TRO vereinheitlicht statt nur fuer
# Weichen. directed_paare (NICHT das ungerichtete verkettete_paare!) -
# sonst deckt eine TRO12->TRO01-Umlaufkante faelschlich auch die
# Gegenrichtung TRO01->TRO12 ab, nur weil dasselbe TRO-Paar beteiligt ist.
edges = {e for e in edges
if not (e[0] in precise_sources and e not in directed_paare)}
# gleicher Host = physisch dieselbe Strecke/Bahn. Ein bereits verkettetes
# PAAR (Bahn-Reihenfolge oben oder Weiche gerade eben) braucht die
# ungerichtete "gleiche Bahn"-Notiz nicht zusaetzlich - ein Weichen-TRO
# kann durchaus auf der Bahn verkettet UND mit seinem Vario-TRO ueber die
# Weiche verbunden sein, das sind zwei unabhaengige Beziehungen.
for host, ids in owners.items():
for tro_id in ids:
tro = next(t for t in tros if t.tro_id == tro_id)
if tro_id in verkettet:
continue
tro.colocated = sorted(
set(tro.colocated) | {o for o in ids if o != tro_id and o not in verkettet}
)
others = {o for o in ids if o != tro_id
and frozenset((tro_id, o)) not in verkettete_paare}
if others:
tro.colocated = sorted(set(tro.colocated) | others)
for tro in tros:
tro.successors = sorted(dst for src, dst in edges if src == tro.tro_id)
@@ -861,6 +1361,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()
@@ -868,7 +1369,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
@@ -878,7 +1380,7 @@ def analyse_tros(
# Verorten vor dem Verknuepfen: die Reihenfolge mehrerer TROs auf einer
# Kreisel-Bahn wird ueber ihre Koordinate bestimmt.
locate_tros(linked, elements, graph)
analysis.edges, analysis.edge_kinds = link_tros(graph, linked)
analysis.edges, analysis.edge_kinds = link_tros(graph, linked, elements)
# Vor Ort aufgenommene Topologie schlaegt die abgeleitete. Die Differenz
# wird protokolliert, damit sichtbar bleibt, was die Ableitung falsch hatte.
@@ -1400,10 +1902,10 @@ def render_doc(
add("| Gefaellestrecke, einzelne Linie | `1Sep` |")
add("| Strecke (angetrieben) | `Vario` (je Streckensegment) |")
add("| Kreisel-Bahn | `1Sep` |")
add(f"| Separator <= {SWITCH_PROXIMITY_MM:.0f}mm (oder naechstgelegen bis "
f"{SWITCH_FALLBACK_MM:.0f}mm) von einem Ausschleuselement "
"| `1Sep1Swi` (Element uebernimmt die fehlende Weiche; ein Einschleuselement "
"zaehlt nicht - passiver Merge ohne Richtungsentscheidung) |")
add("| Separator mit Ausschleuselement unter seinen CSV-Nachbarn (gleiche "
"Kreisel-Bahn) | `1Sep1Swi` (Element uebernimmt die fehlende Weiche; "
"ein Einschleuselement zaehlt nicht - passiver Merge ohne "
"Richtungsentscheidung) |")
add("| zusaetzlich: 2 bzw. 3 abgehende Wege | Aufwertung zu `1Sep1Swi` / `1Sep2Swi` |")
add("")
add("Ein **Scanner** am Separator aendert den Typ nicht. In der Referenzanlage "
@@ -1658,10 +2160,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)