Samer AyadiandClaude Sonnet 5
97965e8ec7
_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>
2088 lines
89 KiB
Python
2088 lines
89 KiB
Python
# -*- coding: utf-8 -*-
|
|
# Absichtlich ohne Shebang: der Windows-py-Launcher wuerde daraus "python3"
|
|
# ableiten und scheitert, wenn keine PythonCore-Installation registriert ist.
|
|
"""
|
|
tro_flow.py - TROs aus einem CSV-Export ermitteln und ihren Fluss zeichnen.
|
|
|
|
Schwesterprogramm zu material_flow.py:
|
|
|
|
material_flow.py Materialfluss der *mechanischen Objekte*
|
|
(Kreisel, Gefaellestrecke, Strecke)
|
|
tro_flow.py Fluss der *Steuerungsobjekte* (TROs)
|
|
|
|
Gleiche Schalter (--file, --tosvg, --doc); --tosvg erzeugt hier das
|
|
Flussdiagramm der TROs, nicht das der Einzelobjekte.
|
|
|
|
Zusaetzlich --use-cords: jeder TRO traegt die Koordinate seiner Bauteile, und
|
|
das SVG wird mit "neato -n" lagerichtig gezeichnet statt von "dot" angeordnet.
|
|
Der Graph sieht dann aus wie das Anlagenlayout.
|
|
|
|
Was ein TRO ist
|
|
---------------
|
|
Ein TRO ("Transfer Route Object") ist die Steuerungslogik **einer
|
|
Uebergabestelle**. Ausgangspunkt ist der Separator (Vereinzeler/Sperre) und das,
|
|
was an ihm haengt (Host-Strecke, Scanner). Ableitungsregeln nach
|
|
doc/500573_Mubea/TRO_Identifikation_500573.md:
|
|
|
|
Host des Separators -> TRO-Typ
|
|
-----------------------------------------------------------------------
|
|
Gefaellestrecke, >= 2 Linien -> PinStore_Auto (ein Block je Gruppe)
|
|
Gefaellestrecke, einzelne Linie -> 1Sep
|
|
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, HundM05):
|
|
|
|
1. 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. 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).
|
|
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 (nicht letzter) 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). 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).
|
|
|
|
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
|
|
SSCC-Scanner, Endmessung und WCS-Telegramm vorbehalten (nur TRO105/TRO110) und
|
|
laesst sich aus dem mechanischen Layout nicht erkennen - es wird nur ein
|
|
Hinweis ausgegeben.
|
|
|
|
Zusaetzlich: hat ein Host 2 bzw. 3 abgehende Flusskanten, wird 1Sep zu
|
|
1Sep1Swi bzw. 1Sep2Swi aufgewertet (Weiche).
|
|
|
|
Typen, FB-Bausteine, Bauteile und Farben kommen ausschliesslich aus
|
|
lib/tro_catalog.py.
|
|
|
|
Ausgabe (alles in SKEL_RESULTS)
|
|
-------------------------------
|
|
<name>_tro_flow.dot immer (mit --use-cords lagerichtige pos-Attribute)
|
|
<name>_tro_flow.svg mit --tosvg (dot, mit --use-cords neato -n)
|
|
<name>_tro_doc.md mit --doc
|
|
<name>_tro_errors.md nur wenn TRO-Fehler gefunden wurden
|
|
|
|
Aufruf ueber bin/tro_flow.bat bzw. bin/tro_flow.sh.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
import math
|
|
import sys
|
|
from collections.abc import Callable
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
from material_flow import (
|
|
EDGE_CIRCLE,
|
|
EDGE_FLOW,
|
|
EDGE_UNRESOLVED,
|
|
LANE_LEFT,
|
|
LANE_RIGHT,
|
|
ROTATION_CCW,
|
|
ROTATION_CW,
|
|
TRANSPORT_KINDS,
|
|
UNASSIGNED,
|
|
Element,
|
|
Graph,
|
|
_as_float,
|
|
_dot_escape,
|
|
_dot_label,
|
|
_merkmal,
|
|
build_graph,
|
|
env_dir,
|
|
graph_to_svg,
|
|
read_elements,
|
|
resolve_input,
|
|
)
|
|
from tro_catalog import (
|
|
CATALOG_AS_OF,
|
|
ITEM_FB,
|
|
STYLE_EXT,
|
|
STYLES,
|
|
TRO_CATALOG,
|
|
TroDefinition,
|
|
TroItem,
|
|
get_tro,
|
|
mermaid_classdefs,
|
|
)
|
|
from tro_overrides import (
|
|
Overrides,
|
|
apply_connections,
|
|
apply_merges,
|
|
apply_overrides,
|
|
apply_splits,
|
|
load_overrides,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Konstanten
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# Ab dieser Anzahl zusammenhaengender Schwerkraftlinien (Gefaellestrecke mit je
|
|
# 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 -->"
|
|
|
|
# Zeichnungsmassstab fuer --use-cords: Graphviz rechnet in Punkt, die Anlage in
|
|
# Millimeter. 0.25 pt/mm entspricht 1:4 - bei ~11 x 20 m Anlage ergibt das eine
|
|
# Zeichnung von etwa 2800 x 5000 pt, in der sich die TRO-Kaesten nicht ueberdecken.
|
|
COORD_SCALE = 0.25
|
|
|
|
# Abstand der Legende zum Anlagenrand in mm
|
|
LEGEND_MARGIN = 3000
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Datenmodell
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class Tro:
|
|
"""Eine ermittelte Uebergabestelle."""
|
|
|
|
tro_id: str
|
|
type_name: str
|
|
label: str
|
|
hosts: list[str]
|
|
separators: list[str]
|
|
scanners: list[str]
|
|
reason: str
|
|
confidence: str
|
|
fb_block: str = ""
|
|
items: dict[str, int] = field(default_factory=dict)
|
|
successors: list[str] = field(default_factory=list)
|
|
predecessors: list[str] = field(default_factory=list)
|
|
colocated: list[str] = field(default_factory=list)
|
|
# Mittelpunkt der Bauteile, aus denen dieser TRO gebildet wurde
|
|
position: tuple[float, float, float] | None = None
|
|
position_from: str = ""
|
|
position_points: int = 0
|
|
|
|
@property
|
|
def x(self) -> float | None:
|
|
return self.position[0] if self.position else None
|
|
|
|
@property
|
|
def y(self) -> float | None:
|
|
return self.position[1] if self.position else None
|
|
|
|
@property
|
|
def z(self) -> float | None:
|
|
return self.position[2] if self.position else None
|
|
|
|
def describe_position(self) -> str:
|
|
"""Koordinate als Text, z. B. '1030, 4700'."""
|
|
if not self.position:
|
|
return "-"
|
|
return f"{self.position[0]:.0f}, {self.position[1]:.0f}"
|
|
|
|
@property
|
|
def connected(self) -> bool:
|
|
return bool(self.successors or self.predecessors or self.colocated)
|
|
|
|
def definition(self) -> TroDefinition | None:
|
|
"""Typdefinition aus tro_catalog (None wenn Typ unbekannt)."""
|
|
return get_tro(self.type_name)
|
|
|
|
|
|
@dataclass
|
|
class Analysis:
|
|
tros: list[Tro] = field(default_factory=list)
|
|
edges: list[tuple[str, str]] = field(default_factory=list)
|
|
# (von, nach) -> Kantenart (transfer / umlauf / weiche / unbestimmt / normal)
|
|
edge_kinds: dict[tuple[str, str], str] = field(default_factory=dict)
|
|
isolated: list[Tro] = field(default_factory=list)
|
|
unknown_types: list[str] = field(default_factory=list)
|
|
findings: list[str] = field(default_factory=list)
|
|
# Manuelle Korrekturen aus cfg/tro_overrides.ini (Typkorrekturen, offene Punkte)
|
|
overrides: Overrides = field(default_factory=Overrides)
|
|
# Wie Host/Fanout ermittelt wurden - siehe derive_tros(resolve_host, count_exits)
|
|
method: str = "Materialfluss-Graph (Zuordnung/Nachbarn)"
|
|
# Systemgrenzen (BTMT Be-/Entladestationen) - siehe derive_boundaries; kein
|
|
# TRO, nur damit der tatsaechliche Materialein-/ausgang sichtbar wird.
|
|
boundaries: list[Tro] = field(default_factory=list)
|
|
|
|
def by_id(self, tro_id: str) -> Tro | None:
|
|
for tro in self.tros + self.boundaries:
|
|
if tro.tro_id == tro_id:
|
|
return tro
|
|
return None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TRO-Ermittlung
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _group_lanes(lanes: list[str], graph: Graph) -> list[list[str]]:
|
|
"""
|
|
Schwerkraftlinien zu Speicherbloecken gruppieren.
|
|
|
|
Linien, die sich einen Zu- oder Ablaufknoten teilen, gehoeren zum selben
|
|
Speicher. So entstehen bei zwei getrennten Speichern auch zwei TROs.
|
|
"""
|
|
adjacency: dict[str, set[str]] = {}
|
|
for lane in lanes:
|
|
neighbours: set[str] = set()
|
|
for edge in graph.edges:
|
|
if edge.src == lane:
|
|
neighbours.add(edge.dst)
|
|
elif edge.dst == lane:
|
|
neighbours.add(edge.src)
|
|
adjacency[lane] = neighbours
|
|
|
|
groups: list[dict] = []
|
|
for lane in sorted(lanes):
|
|
current = {"lanes": [lane], "nodes": set(adjacency[lane]) | {lane}}
|
|
for group in [g for g in groups if current["nodes"] & g["nodes"]]:
|
|
current["lanes"].extend(group["lanes"])
|
|
current["nodes"] |= group["nodes"]
|
|
groups.remove(group)
|
|
groups.append(current)
|
|
|
|
for group in groups:
|
|
group["lanes"].sort()
|
|
groups.sort(key=lambda g: g["lanes"][0])
|
|
return [group["lanes"] for group in groups]
|
|
|
|
|
|
def _switch_exits(hosts: list[str], graph: Graph) -> list[str]:
|
|
"""Abgehende Flusskanten eines TRO-Hosts (ohne Kreisel-Umlauf)."""
|
|
return sorted(
|
|
{
|
|
edge.dst
|
|
for edge in graph.edges
|
|
if edge.kind == EDGE_FLOW and edge.src in hosts and edge.dst not in hosts
|
|
}
|
|
)
|
|
|
|
|
|
def _default_resolve_host(sep: Element, findings: list[str], graph: Graph) -> str | None:
|
|
"""Host eines Separators aus der CSV-Spalte 'Zuordnung' (Default-Resolver)."""
|
|
host = sep.assignment
|
|
if host and host != UNASSIGNED and host in graph.nodes:
|
|
return host
|
|
findings.append(
|
|
f"Separator {sep.teile_id} '{sep.name}': Zuordnung '{host or '-'}' "
|
|
f"zeigt auf keinen Flussknoten - keinem TRO zugeordnet"
|
|
)
|
|
return None
|
|
|
|
|
|
def _find_switch_separators(
|
|
separators: list[Element], elements: list[Element], tol_mm: float
|
|
) -> dict[str, tuple[Element, float]]:
|
|
"""Separator-Id -> (Ausschleuselement der Weiche, Abstand mm).
|
|
|
|
Ein Separator ist eine Weiche, wenn ihm ein ILS Ausschleuselement
|
|
zugeordnet werden kann - das Element ist die Weiche (siehe unten), der
|
|
Separator nur der Steuerpunkt davor. NICHT das Einschleuselement: das
|
|
Wiedereinschleusen ist ein passiver Merge ohne Richtungsentscheidung, nur
|
|
das Ausschleusen (Verlassen der Kreisel-Bahn) ist eine Weichenstelle - vor
|
|
Ort bestaetigt (HundM05, Review 2026-09-21): ein Separator an einem
|
|
Einschleuselement ist keine eigene Uebergabestelle. Ein eventueller
|
|
Scanner am selben Separator ist ebenfalls kein Kriterium, sondern nur ein
|
|
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:
|
|
|
|
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) - unabhaengig davon, ob Regel 1 auf derselben Bahn
|
|
bereits einen Treffer hatte. 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.
|
|
|
|
Frueher lief Regel 2 nur auf einer Bahn, die Regel 1 bereits bestaetigt
|
|
hatte (eine Bahn ohne jeden nahen Treffer hat vermutlich gar keine
|
|
Weiche). Diese Bahn-Sperre wurde entfernt (Review 2026-09-23): bei
|
|
einer neueren HundM05-Revision lagen ALLE 5 echten Schalt-Separatoren
|
|
gleichmaessig bei ca. 1013-1016mm - ausserhalb von Regel 1 (500mm), aber
|
|
innerhalb von SWITCH_FALLBACK_MM. Ohne einen einzigen Regel-1-Treffer
|
|
blieb die Bahn-Sperre fuer jede Bahn zu, Regel 2 kam nie zum Zug, und
|
|
keine der 5 Weichen wurde erkannt. SWITCH_FALLBACK_MM bleibt die
|
|
einzige Schranke gegen Fehlzuordnungen (siehe Mubea-Fall oben) - sollte
|
|
sich das an einer weiteren Anlage als zu grosszuegig erweisen, braucht
|
|
es dort eine gezieltere Loesung statt der Bahn-Sperre.
|
|
"""
|
|
transfers = [e for e in elements if e.kind == "Ausschleus"]
|
|
result: dict[str, tuple[Element, float]] = {}
|
|
|
|
for sep in separators:
|
|
if sep.x is None or sep.y is None:
|
|
continue
|
|
best: tuple[float, Element] | None = None
|
|
for transfer in transfers:
|
|
if transfer.x is None or transfer.y 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])
|
|
|
|
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]
|
|
|
|
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:
|
|
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)
|
|
|
|
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 derive_tros(
|
|
graph: Graph,
|
|
elements: list[Element],
|
|
findings: list[str],
|
|
overrides: Overrides | None = None,
|
|
*,
|
|
resolve_host: Callable[[Element, list[str]], str | None] | None = None,
|
|
count_exits: Callable[[list[str]], list[str]] | None = None,
|
|
) -> list[Tro]:
|
|
"""TRO-Liste aus den Separatoren und ihren angeschlossenen Objekten bilden.
|
|
|
|
resolve_host(sep, findings) -> Host-Knoten-Id oder None; Default liest die
|
|
CSV-Spalte "Zuordnung" (siehe _default_resolve_host). count_exits(hosts) ->
|
|
Liste abgehender Nachbarn eines TRO-Hosts; Default liest graph.edges (siehe
|
|
_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.
|
|
"""
|
|
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))
|
|
|
|
separators = sorted(
|
|
(e for e in elements if e.kind == "Separator"), key=lambda e: e.teile_id
|
|
)
|
|
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 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)
|
|
|
|
# Separator -> Host-Knoten
|
|
sep_host: dict[str, str] = {}
|
|
for sep in separators:
|
|
if sep.teile_id in switch_hits:
|
|
continue
|
|
host = resolve_host(sep, findings)
|
|
if host is not None:
|
|
sep_host[sep.teile_id] = host
|
|
|
|
# Scanner -> Separator; "Naechster Separator" ist die funktionale Zuordnung,
|
|
# "Zuordnung" nur der Montageort.
|
|
sep_scanners: dict[str, list[str]] = {}
|
|
for scanner in scanners:
|
|
target = str(_merkmal(scanner.merkmale, "Naechster Separator") or "").strip()
|
|
if target in sep_host or target in switch_hits:
|
|
sep_scanners.setdefault(target, []).append(scanner.teile_id)
|
|
mounted = scanner.assignment
|
|
target_host = sep_host.get(target)
|
|
# Fuer Switch-Separatoren (3b) gibt es keinen einzelnen Host zum
|
|
# Vergleichen (sie haengen zwischen zwei Ketten) - der Abgleich
|
|
# entfaellt dort, ohne den Scanner deshalb als "unbekannt" zu werten.
|
|
if (mounted and mounted != UNASSIGNED and target_host is not None
|
|
and mounted != target_host):
|
|
findings.append(
|
|
f"Scanner {scanner.teile_id} '{scanner.name}': montiert an {mounted}, "
|
|
f"liest aber an Separator {target} auf {target_host}"
|
|
)
|
|
else:
|
|
findings.append(
|
|
f"Scanner {scanner.teile_id} '{scanner.name}': 'Naechster Separator' "
|
|
f"'{target or '-'}' unbekannt - keinem TRO zugeordnet"
|
|
)
|
|
|
|
def seps_on(hosts: list[str]) -> list[str]:
|
|
return sorted(s for s, h in sep_host.items() if h in hosts)
|
|
|
|
def scanners_for(seps: list[str]) -> list[str]:
|
|
found: list[str] = []
|
|
for sep in seps:
|
|
found.extend(sep_scanners.get(sep, []))
|
|
return sorted(found)
|
|
|
|
tros: list[Tro] = []
|
|
|
|
# 1) Schwerkraftlinien (Gefaellestrecke mit Separator) -> Speicherblock
|
|
lanes = sorted(
|
|
{h for h in sep_host.values() if graph.nodes[h].kind == "Gefaellestrecke"}
|
|
)
|
|
for group in _group_lanes(lanes, graph):
|
|
seps = seps_on(group)
|
|
if len(group) >= MIN_STORE_LANES:
|
|
tros.append(
|
|
Tro(
|
|
tro_id="",
|
|
type_name="PinStore_Auto",
|
|
label=f"Speicher ({len(group)} Schwerkraftlinien)",
|
|
hosts=group,
|
|
separators=seps,
|
|
scanners=scanners_for(seps),
|
|
reason=(
|
|
f"{len(group)} Gefaellestrecken mit je eigenem Separator, "
|
|
f"gemeinsamer Zu-/Ablauf - Muster eines Linienspeichers"
|
|
),
|
|
confidence="hoch",
|
|
)
|
|
)
|
|
else:
|
|
for lane in group:
|
|
lane_seps = seps_on([lane])
|
|
tros.append(
|
|
Tro(
|
|
tro_id="",
|
|
type_name="1Sep",
|
|
label=graph.nodes[lane].element.name,
|
|
hosts=[lane],
|
|
separators=lane_seps,
|
|
scanners=scanners_for(lane_seps),
|
|
reason="einzelne Gefaellestrecke mit Separator (kein Linienspeicher)",
|
|
confidence="mittel",
|
|
)
|
|
)
|
|
|
|
# 2) Angetriebene Strecken -> je SEPARATOR ein eigener Vario-TRO. Katalog
|
|
# (TRO_CATALOG in tro_catalog.py) definiert Vario als 1x Separator + 1x
|
|
# VarioDrive - mehrere Separatoren auf demselben Streckensegment sind
|
|
# mehrere eigenstaendige Uebergabestellen, keine gemeinsame (Review
|
|
# 2026-09-21). Ein Segment ganz ohne Separator behaelt weiterhin einen
|
|
# einzelnen TRO (leere Separatorliste), damit das Antriebssegment selbst
|
|
# nicht verschwindet.
|
|
for node_id in sorted(n for n, node in graph.nodes.items() if node.kind == "Strecke"):
|
|
seps = seps_on([node_id])
|
|
element = graph.nodes[node_id].element
|
|
drive = element.drive_dir or "unbekannt"
|
|
reason = f"angetriebenes Streckensegment (Antriebfahrtrichtung: {drive})"
|
|
|
|
# Letzter Separator vor dem Wiedereinschleusen auf die Kreisel-Bahn
|
|
# (in echter TrackIds-Reihenfolge, nicht sortierte Teile-Id) ist ein
|
|
# reiner Vereinzeler ohne eigene VarioDrive-Strecke: die Kreisel-Bahn
|
|
# uebernimmt den Antrieb ab dort selbst - 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. Gilt unabhaengig von der
|
|
# Separatoranzahl - siehe auch VF_1 (0034, letzter von vieren). Nur
|
|
# bei genau einem Separator gibt es kein "erster vs. letzter", der
|
|
# bleibt Vario (er ist der einzige Antriebspunkt der Strecke).
|
|
seps_set = set(seps)
|
|
track_seps = [tid for tid in element.track_ids if tid in seps_set]
|
|
last_sep_id = track_seps[-1] if len(track_seps) > 1 else None
|
|
|
|
for sep_id in seps or [None]:
|
|
sep_list = [sep_id] if sep_id else []
|
|
type_name = "Vario"
|
|
sep_reason = reason
|
|
confidence = "hoch"
|
|
if sep_id is not None and sep_id == last_sep_id:
|
|
type_name = "1Sep"
|
|
confidence = "mittel"
|
|
sep_reason = (
|
|
f"letzter Separator vor dem Wiedereinschleusen von "
|
|
f"{element.name} auf die Kreisel-Bahn - reiner Vereinzeler, "
|
|
f"kein eigenes VarioDrive mehr noetig (Review 2026-09-23)"
|
|
)
|
|
findings.append(
|
|
f"{element.name} Separator {sep_id}: letzter Separator vor dem "
|
|
f"Wiedereinschleusen - als 1Sep statt Vario eingestuft"
|
|
)
|
|
elif sep_id is not None:
|
|
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'"
|
|
)
|
|
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"
|
|
)
|
|
tros.append(
|
|
Tro(
|
|
tro_id="",
|
|
type_name=type_name,
|
|
label=element.name,
|
|
hosts=[node_id],
|
|
separators=sep_list,
|
|
scanners=scanners_for(sep_list),
|
|
reason=sep_reason,
|
|
confidence=confidence,
|
|
)
|
|
)
|
|
|
|
# 3) Separatoren auf Kreisel-Bahnen -> je Separator eine Uebergabestelle
|
|
for sep_id in sorted(
|
|
s for s, h in sep_host.items() if graph.nodes[h].kind == "Kreisel"
|
|
):
|
|
host = sep_host[sep_id]
|
|
own_scanners = sorted(sep_scanners.get(sep_id, []))
|
|
node = graph.nodes[host]
|
|
lane_name = _merkmal(node.element.merkmale, "Name") or node.element.name
|
|
tros.append(
|
|
Tro(
|
|
tro_id="",
|
|
type_name="1Sep",
|
|
label=f"Separator {sep_id} auf {lane_name} Bahn {node.lane}",
|
|
hosts=[host],
|
|
separators=[sep_id],
|
|
scanners=own_scanners,
|
|
reason=(
|
|
"Separator mit Scanner auf angetriebener Kreisel-Bahn"
|
|
if own_scanners
|
|
else "einzelner Separator auf angetriebener Kreisel-Bahn"
|
|
),
|
|
confidence="mittel",
|
|
)
|
|
)
|
|
|
|
# 3b) Separatoren an einem Ausschleuselement -> eigene 1Sep1Swi-
|
|
# Uebergabestelle. Das Element uebernimmt die Rolle der (hier fehlenden)
|
|
# Weiche; siehe _find_switch_separators. Host bewusst die Strecke (nicht
|
|
# die Kreisel-Bahn): beide Elementarten tragen "Strecke" oder "Foerderer"
|
|
# im Merkmal (beide Schreibweisen kommen nebeneinander im selben Export
|
|
# vor), und die Strecke hat bereits reale, gerichtete Kanten zum Kreisel
|
|
# (_apply_transfers) - darueber haengt sich die Verknuepfung (link_tros)
|
|
# von selbst ein, ohne einen weiteren synthetischen Host zu erfinden.
|
|
for sep_id, (transfer, distance) in sorted(switch_hits.items()):
|
|
strecke_id = str(_merkmal(transfer.merkmale, "Strecke", "Förderer") or "")
|
|
ende = str(_merkmal(transfer.merkmale, "Streckenende") or "")
|
|
own_scanners = sorted(sep_scanners.get(sep_id, []))
|
|
art = "Ausschleuselement"
|
|
tros.append(
|
|
Tro(
|
|
tro_id="",
|
|
type_name="1Sep1Swi",
|
|
label=f"Separator {sep_id} an {art} {transfer.teile_id} "
|
|
f"(Strecke {strecke_id}, {ende})",
|
|
hosts=[strecke_id] if strecke_id in graph.nodes else [transfer.teile_id],
|
|
separators=[sep_id],
|
|
scanners=own_scanners,
|
|
reason=(
|
|
f"{distance:.0f}mm von {art} {transfer.teile_id} entfernt - "
|
|
f"kein eigenes Weichen-Bauteil im Layout, das {art} "
|
|
f"uebernimmt diese Rolle"
|
|
),
|
|
confidence="mittel",
|
|
)
|
|
)
|
|
|
|
# 4) Weichen-Aufwertung: 2 bzw. 3 abgehende Flusskanten
|
|
for tro in tros:
|
|
if tro.type_name != "1Sep":
|
|
continue
|
|
exits = count_exits(tro.hosts)
|
|
if not 2 <= len(exits) <= 3:
|
|
continue
|
|
upgraded = "1Sep1Swi" if len(exits) == 2 else "1Sep2Swi"
|
|
findings.append(
|
|
f"{tro.label}: {len(exits)} abgehende Wege ({', '.join(exits)}) - "
|
|
f"als Weiche {upgraded} statt {tro.type_name} eingestuft"
|
|
)
|
|
tro.type_name = upgraded
|
|
tro.reason = f"{tro.reason}; {len(exits)} abgehende Wege -> Weiche"
|
|
tro.confidence = "niedrig"
|
|
|
|
# 4b) Manuell ergaenzte Uebergabestellen (cfg/tro_overrides.ini, [.split.]).
|
|
# Vor der Nummernvergabe, damit die neuen TROs in derselben Reihenfolge
|
|
# eine Nummer bekommen wie die abgeleiteten.
|
|
if overrides is not None and overrides.splits:
|
|
def _neuer_tro(spec, hosts, label):
|
|
return Tro(
|
|
tro_id="",
|
|
type_name=spec.type_name,
|
|
label=label,
|
|
hosts=hosts,
|
|
separators=list(spec.separators),
|
|
scanners=list(spec.scanners),
|
|
reason=spec.reason or "manuell ergaenzte Uebergabestelle",
|
|
confidence=spec.confidence,
|
|
)
|
|
|
|
apply_splits(tros, overrides, _neuer_tro)
|
|
|
|
# 4c) Zusammenfassungen - nach den Splits, damit auch ein frisch
|
|
# herausgeloester TRO Teil einer Zusammenfassung sein kann.
|
|
if overrides is not None and overrides.merges:
|
|
apply_merges(tros, overrides)
|
|
|
|
# 5) Stabile, laufende Nummern in Reihenfolge der Host-Knoten
|
|
tros.sort(key=lambda t: (t.hosts[0], t.separators[0] if t.separators else ""))
|
|
width = max(2, len(str(len(tros))))
|
|
for index, tro in enumerate(tros, start=1):
|
|
tro.tro_id = f"TRO{index:0{width}d}"
|
|
|
|
# 5b) Manuelle Korrekturen aus cfg/tro_overrides.ini. Sie greifen erst hier,
|
|
# weil sie ueber die TRO-ID adressiert sind, und noch vor Schritt 6, damit
|
|
# ein manuell gesetzter Typ den SSCC-Hinweis nicht mehr ausloest.
|
|
if overrides is not None:
|
|
apply_overrides(tros, overrides)
|
|
|
|
# 6) Scanner am Separator: 1Sep_SSCC laesst sich mechanisch nicht erkennen
|
|
for tro in tros:
|
|
if tro.type_name == "1Sep" and tro.scanners:
|
|
findings.append(
|
|
f"{tro.tro_id} ({tro.label}): Scanner {', '.join(tro.scanners)} am "
|
|
f"Separator - als 1Sep eingestuft. Pruefen, ob 1Sep_SSCC "
|
|
f"(SSCC-Scanner, Endmessung, WCS-Telegramm) erforderlich ist; das "
|
|
f"ist aus dem mechanischen Layout nicht erkennbar."
|
|
)
|
|
return tros
|
|
|
|
|
|
# Typnamen fuer Systemgrenz-Knoten (siehe derive_boundaries) - bewusst NICHT
|
|
# 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]:
|
|
"""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 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
|
|
Ende des Materialflusses. Eigene Nummerierung ("SYS1", "SYS2", ...)
|
|
getrennt von den TRO-IDs, damit auf den ersten Blick klar bleibt: das
|
|
sind keine Steuerobjekte.
|
|
"""
|
|
boundaries: list[Tro] = []
|
|
for element in elements:
|
|
if not element.is_station:
|
|
continue
|
|
if element.teile_id not in graph.nodes:
|
|
continue
|
|
boundaries.append(
|
|
Tro(
|
|
tro_id="",
|
|
type_name=BOUNDARY_ENTRY if element.is_entry else BOUNDARY_EXIT,
|
|
label=element.name,
|
|
hosts=[element.teile_id],
|
|
separators=[],
|
|
scanners=[],
|
|
reason="Systemgrenze (Be-/Entladestation) - kein TRO, "
|
|
"nur Materialein-/ausgang",
|
|
confidence="hoch",
|
|
)
|
|
)
|
|
boundaries.sort(key=lambda t: t.hosts[0])
|
|
width = max(1, len(str(len(boundaries))))
|
|
for index, boundary in enumerate(boundaries, start=1):
|
|
boundary.tro_id = f"SYS{index:0{width}d}"
|
|
return boundaries
|
|
|
|
|
|
def _tro_edge_kind(edge: Edge) -> str:
|
|
"""Materialkante -> Kantenart des TRO-Flusses (fuer Einfaerbung)."""
|
|
if edge.note == "Weiche":
|
|
return "weiche"
|
|
if edge.kind == EDGE_CIRCLE:
|
|
return "umlauf"
|
|
if edge.kind == EDGE_UNRESOLVED:
|
|
return "unbestimmt"
|
|
if edge.note in ("Einschleusung", "Ausschleusung"):
|
|
return "transfer"
|
|
return "normal"
|
|
|
|
|
|
def _bahn_richtung(node) -> tuple[float, float] | None:
|
|
"""Foerderrichtung einer Kreisel-Bahn als Einheitsvektor.
|
|
|
|
Der Kreisel ist eine geschlossene Schleife: auf der einen Bahn laeuft das
|
|
Material hin, auf der anderen zurueck. Welche Richtung das ist, ergibt sich
|
|
aus der Laengsachse ("Drehung") und dem Umlaufsinn ("Drehrichtung"):
|
|
|
|
UZS (im Uhrzeigersinn) Bahn R entgegen der Achse, Bahn L mit ihr
|
|
GUZ (gegen den UZS) umgekehrt
|
|
|
|
Rueckgabe None, wenn eines der beiden Merkmale fehlt - dann bleibt es bei
|
|
der ungerichteten "gleiche Bahn"-Notiz.
|
|
"""
|
|
element = node.element
|
|
drehung = _as_float(_merkmal(element.merkmale, "Drehung"))
|
|
umlauf = element.rotation
|
|
if drehung is None or umlauf not in (ROTATION_CW, ROTATION_CCW):
|
|
return None
|
|
rad = math.radians(drehung)
|
|
achse = (math.cos(rad), math.sin(rad))
|
|
mit_achse = (node.lane == LANE_LEFT) == (umlauf == ROTATION_CW)
|
|
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], elements: list[Element]
|
|
) -> tuple[list[tuple[str, str]], dict[tuple[str, str], str]]:
|
|
"""
|
|
TROs verknuepfen.
|
|
|
|
Der Materialfluss laeuft von Uebergabestelle zu Uebergabestelle, ggf. durch
|
|
passive Zwischenknoten (Strecken/Bahnen ohne eigenen TRO). Solche Knoten
|
|
werden daher durchlaufen ("kontrahiert"), belegte Knoten beenden den Pfad.
|
|
Jede TRO-Kante bekommt zusaetzlich die Art der *ankommenden* Materialkante
|
|
(transfer / umlauf / weiche / unbestimmt / normal).
|
|
"""
|
|
owners: dict[str, list[str]] = {}
|
|
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:
|
|
kind = _tro_edge_kind(edge)
|
|
forward.setdefault(edge.src, []).append((edge.dst, kind))
|
|
if edge.kind == EDGE_UNRESOLVED:
|
|
# unbestimmte Richtung: in beide Richtungen begehbar
|
|
forward.setdefault(edge.dst, []).append((edge.src, kind))
|
|
|
|
edges: set[tuple[str, str]] = set()
|
|
edge_kinds: dict[tuple[str, str], str] = {}
|
|
for tro in tros:
|
|
seen = set(tro.hosts)
|
|
queue = [(nxt, k) for host in tro.hosts for (nxt, k) in forward.get(host, [])]
|
|
while queue:
|
|
node, arriving = queue.pop()
|
|
if node in seen:
|
|
continue
|
|
seen.add(node)
|
|
if node in owners:
|
|
for other in owners[node]:
|
|
if other != tro.tro_id:
|
|
pair = (tro.tro_id, other)
|
|
edges.add(pair)
|
|
edge_kinds.setdefault(pair, arriving)
|
|
continue
|
|
queue.extend(forward.get(node, []))
|
|
|
|
# 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
|
|
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 gruppe if t.position]
|
|
if len(auf_bahn) < 2:
|
|
continue
|
|
# Projektion auf die Foerderrichtung: kleinster Wert kommt zuerst
|
|
auf_bahn.sort(key=lambda t: t.position[0] * richtung[0]
|
|
+ t.position[1] * richtung[1])
|
|
for vorher, nachher in zip(auf_bahn, auf_bahn[1:]):
|
|
pair = (vorher.tro_id, nachher.tro_id)
|
|
edges.add(pair)
|
|
edge_kinds.setdefault(pair, "bahn")
|
|
verkettete_paare.add(frozenset(pair))
|
|
directed_paare.add(pair)
|
|
|
|
# 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)
|
|
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)
|
|
tro.predecessors = sorted(src for src, dst in edges if dst == tro.tro_id)
|
|
|
|
return sorted(edges), edge_kinds
|
|
|
|
|
|
def _centroid(
|
|
points: list[tuple[float, float, float]]
|
|
) -> tuple[float, float, float] | None:
|
|
"""Mittelpunkt mehrerer Positionen."""
|
|
if not points:
|
|
return None
|
|
count = len(points)
|
|
return (
|
|
sum(p[0] for p in points) / count,
|
|
sum(p[1] for p in points) / count,
|
|
sum(p[2] for p in points) / count,
|
|
)
|
|
|
|
|
|
def locate_tros(tros: list[Tro], elements: list[Element], graph: Graph) -> None:
|
|
"""
|
|
Jedem TRO eine Koordinate geben.
|
|
|
|
Die Koordinate ist der **Mittelpunkt der Bauteile**, aus denen der TRO
|
|
gebildet wurde - bei einem 1Sep1Swi also der Mittelwert aus Separator und
|
|
Weiche, beim Linienspeicher der Mittelwert aller Linien-Separatoren.
|
|
|
|
Hat ein TRO keine Bauteile mit Position (z. B. ein Vario ohne Separator),
|
|
wird auf die Position seiner Host-Objekte zurueckgegriffen; das steht dann
|
|
in `position_from`.
|
|
"""
|
|
by_id = {element.teile_id: element for element in elements}
|
|
|
|
for tro in tros:
|
|
# 1) Bauteile des TRO (Separatoren, Scanner)
|
|
parts = [
|
|
by_id[i].position
|
|
for i in list(tro.separators) + list(tro.scanners)
|
|
if i in by_id and by_id[i].position
|
|
]
|
|
source = "Bauteile"
|
|
|
|
# 2) Ersatzweise die Host-Objekte
|
|
if not parts:
|
|
parts = [
|
|
graph.nodes[h].element.position
|
|
for h in tro.hosts
|
|
if h in graph.nodes and graph.nodes[h].element.position
|
|
]
|
|
source = "Host-Objekte"
|
|
|
|
tro.position = _centroid(parts)
|
|
tro.position_points = len(parts)
|
|
tro.position_from = source if tro.position else "keine Position"
|
|
|
|
|
|
def analyse_tros(
|
|
graph: Graph,
|
|
elements: list[Element],
|
|
overrides: Overrides | None = None,
|
|
*,
|
|
resolve_host: Callable[[Element, list[str]], str | None] | None = None,
|
|
count_exits: Callable[[list[str]], list[str]] | None = None,
|
|
method: str | None = None,
|
|
) -> Analysis:
|
|
"""TROs ermitteln, verknuepfen, verorten und gegen die Typdefinition pruefen."""
|
|
analysis = Analysis()
|
|
analysis.overrides = overrides if overrides is not None else Overrides()
|
|
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)
|
|
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
|
|
# Eintrags in der uebergebenen Liste ist - eine BTMT-Station muss also mit
|
|
# in dieser Liste stehen, sonst bleibt sie unsichtbar (siehe derive_boundaries).
|
|
linked = analysis.tros + analysis.boundaries
|
|
# 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, elements)
|
|
|
|
# Vor Ort aufgenommene Topologie schlaegt die abgeleitete. Die Differenz
|
|
# wird protokolliert, damit sichtbar bleibt, was die Ableitung falsch hatte.
|
|
if analysis.overrides.connections:
|
|
abgeleitet = set(analysis.edges)
|
|
analysis.edges, analysis.edge_kinds = apply_connections(
|
|
linked, analysis.overrides
|
|
)
|
|
gesetzt = set(analysis.edges)
|
|
for src, dst in sorted(abgeleitet - gesetzt):
|
|
analysis.findings.append(
|
|
f"abgeleitete Verbindung {src} -> {dst} entfaellt - steht nicht "
|
|
f"in der gesetzten Topologie"
|
|
)
|
|
for src, dst in sorted(gesetzt - abgeleitet):
|
|
analysis.findings.append(
|
|
f"Verbindung {src} -> {dst} nur gesetzt, aus dem Layout nicht "
|
|
f"ableitbar"
|
|
)
|
|
|
|
for tro in analysis.tros:
|
|
definition = tro.definition()
|
|
if definition:
|
|
tro.fb_block = definition.get_fb_block()
|
|
tro.items = definition.get_items()
|
|
else:
|
|
analysis.unknown_types.append(tro.type_name)
|
|
|
|
analysis.unknown_types = sorted(set(analysis.unknown_types))
|
|
analysis.isolated = [tro for tro in analysis.tros if not tro.connected]
|
|
analysis.findings.sort()
|
|
return analysis
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DOT erzeugen - Flussdiagramm der TROs
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _tro_label(tro: Tro) -> str:
|
|
lines = [f"{tro.tro_id} ({tro.type_name})", tro.label]
|
|
if tro.fb_block:
|
|
lines.append(tro.fb_block)
|
|
if tro.items:
|
|
lines.append(", ".join(f"{c}x {i}" for i, c in sorted(tro.items.items())))
|
|
if tro.separators:
|
|
seps = ", ".join(tro.separators)
|
|
if len(tro.separators) > 6:
|
|
seps = f"{tro.separators[0]} … {tro.separators[-1]} ({len(tro.separators)})"
|
|
lines.append(f"Sep: {seps}")
|
|
if tro.scanners:
|
|
lines.append(f"Scan: {', '.join(tro.scanners)}")
|
|
if tro.position:
|
|
lines.append(f"x/y: {tro.describe_position()}")
|
|
return _dot_label(lines)
|
|
|
|
|
|
def missing_positions(analysis: Analysis) -> list[str]:
|
|
"""TROs/Systemgrenzen ohne Koordinate - mit ihnen ist kein lagerichtiges
|
|
Layout moeglich (auch eine Systemgrenze wird als Knoten gezeichnet)."""
|
|
return [t.tro_id for t in analysis.tros + analysis.boundaries if not t.position]
|
|
|
|
|
|
def _legend_pos(analysis: Analysis) -> tuple[float, float]:
|
|
"""Legende links unterhalb der Anlage platzieren (mm)."""
|
|
all_nodes = analysis.tros + analysis.boundaries
|
|
xs = [t.position[0] for t in all_nodes if t.position]
|
|
ys = [t.position[1] for t in all_nodes if t.position]
|
|
if not xs:
|
|
return (0.0, 0.0)
|
|
return (min(xs) - LEGEND_MARGIN, min(ys) - LEGEND_MARGIN)
|
|
|
|
|
|
# Kantenstil je Art (an material_flow.py angelehnt: Umlauf hellblau gestrichelt,
|
|
# Weiche orange). "transfer"/"normal" bleiben der durchgezogene Standardfluss.
|
|
TRO_EDGE_STYLE = {
|
|
"umlauf": 'color="#8ea9db", style=dashed, tooltip="Kreisel-Umlauf"',
|
|
"weiche": 'color="#cc8800", penwidth=1.8, xlabel="Weiche", fontcolor="#cc8800", '
|
|
'tooltip="Weiche (Kreisel-Uebergang)"',
|
|
"unbestimmt": 'color="#bf8f00", style=dashed, dir=both, tooltip="Richtung unbestimmt"',
|
|
"transfer": 'tooltip="Uebergabe (Ein-/Ausschleusung)"',
|
|
}
|
|
|
|
|
|
def _tro_edge_attrs(kind: str | None) -> str:
|
|
return TRO_EDGE_STYLE.get(kind or "normal", "")
|
|
|
|
|
|
def render_dot(analysis: Analysis, source: str, use_coords: bool = False) -> str:
|
|
"""
|
|
Gerichtetes Flussdiagramm der TROs als Graphviz-Quelle.
|
|
|
|
use_coords=False Anordnung berechnet 'dot' (Fluss von links nach rechts)
|
|
use_coords=True jeder Knoten wird auf seine Anlagenkoordinate gesetzt und
|
|
mit "neato -n" lagerichtig gezeichnet
|
|
"""
|
|
out: list[str] = []
|
|
add = out.append
|
|
|
|
add("// Automatisch erzeugt von lib/tro_flow.py - nicht manuell aendern.")
|
|
add(f"// Quelle: {source}")
|
|
if use_coords:
|
|
add(f"// Lagerichtig: Knoten auf Anlagenkoordinate, {COORD_SCALE} pt/mm.")
|
|
add("// Rendern mit: neato -n -Tsvg <datei>.dot -o <datei>.svg")
|
|
add("digraph TROFluss {")
|
|
if use_coords:
|
|
# Bei -n bestimmen die pos-Attribute die Lage; rankdir/ranksep entfallen.
|
|
add(" graph [splines=true, outputorder=edgesfirst,")
|
|
add(' fontname="Segoe UI", fontsize=11, labelloc="t",')
|
|
add(f' label="TRO-Fluss lagerichtig ({COORD_SCALE} pt/mm) - '
|
|
f'{_dot_escape(source)}"];')
|
|
else:
|
|
add(" graph [rankdir=LR, splines=spline, nodesep=0.35, ranksep=1.3,")
|
|
add(' fontname="Segoe UI", fontsize=11, labelloc="t",')
|
|
add(f' label="TRO-Fluss - {_dot_escape(source)}"];')
|
|
add(' node [shape=box, style="rounded,filled", fontname="Segoe UI", fontsize=9];')
|
|
add(' edge [fontname="Segoe UI", fontsize=8, color="#2f5597", penwidth=1.2,'
|
|
" arrowsize=0.8];")
|
|
add("")
|
|
|
|
for tro in analysis.tros:
|
|
definition = tro.definition()
|
|
attrs = definition.get_dot_attrs() if definition else {
|
|
"fillcolor": "#ffffff", "color": "#c00000", "fontcolor": "#c00000"
|
|
}
|
|
rendered = ", ".join(f'{k}="{v}"' for k, v in attrs.items())
|
|
# Anlagenkoordinate mitgeben. Ohne --use-cords ignoriert dot das pos,
|
|
# die Angabe bleibt aber als Information im DOT stehen.
|
|
pos = ""
|
|
if tro.position:
|
|
scale = COORD_SCALE if use_coords else 0.01
|
|
pos = (f', pos="{tro.position[0] * scale:.2f},'
|
|
f'{tro.position[1] * scale:.2f}"')
|
|
add(f' "{tro.tro_id}" [label="{_tro_label(tro)}", {rendered}{pos}];')
|
|
add("")
|
|
|
|
# Systemgrenzen (BTMT Be-/Entladestationen): kein Katalogtyp, eigene
|
|
# Farbe/Form (ext-Stil, Ellipse statt Box) - deutlich als "kein TRO"
|
|
# erkennbar, aber sichtbar als tatsaechlicher Anfang/Ende des Flusses.
|
|
for boundary in analysis.boundaries:
|
|
rendered = ", ".join(f'{k}="{v}"' for k, v in STYLE_EXT.dot_attrs().items())
|
|
pos = ""
|
|
if boundary.position:
|
|
scale = COORD_SCALE if use_coords else 0.01
|
|
pos = (f', pos="{boundary.position[0] * scale:.2f},'
|
|
f'{boundary.position[1] * scale:.2f}"')
|
|
add(f' "{boundary.tro_id}" [label="{_tro_label(boundary)}", '
|
|
f'shape=ellipse, style=filled, {rendered}{pos}];')
|
|
add("")
|
|
|
|
for src, dst in analysis.edges:
|
|
add(f' "{src}" -> "{dst}" [{_tro_edge_attrs(analysis.edge_kinds.get((src, dst)))}];')
|
|
|
|
seen: set[frozenset] = set()
|
|
for tro in analysis.tros:
|
|
for other in tro.colocated:
|
|
key = frozenset((tro.tro_id, other))
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
add(f' "{tro.tro_id}" -> "{other}" [dir=none, style=dotted, '
|
|
f'color="#7f7f7f", constraint=false, tooltip="gleiche Bahn"];')
|
|
add("")
|
|
|
|
# Legende: eine Zeile je TRO-Typ (Farbe der Gruppe + eigene Form, siehe
|
|
# tro_catalog.TRO_CATALOG), plus eine Zeile fuer die Systemgrenzen und die
|
|
# Kantenarten. Bei -n braucht auch sie ein pos, sonst landet sie auf 0,0
|
|
# mitten in der Anlage - daher unterhalb links vom Anlagenrand.
|
|
rows = "".join(
|
|
f'<tr><td bgcolor="{d.get_style().fill}" align="left"> {d.get_name()} </td>'
|
|
f'<td align="left">{d.describe_symbol()}</td></tr>'
|
|
for d in TRO_CATALOG
|
|
)
|
|
rows += (f'<tr><td bgcolor="{STYLE_EXT.fill}" align="left"> Systemgrenze </td>'
|
|
f'<td align="left">Ellipse (Materialein-/ausgang, kein TRO)</td></tr>')
|
|
legend_label = ('<<table border="0" cellborder="1" cellspacing="0" cellpadding="3">'
|
|
+ rows
|
|
+ '<tr><td align="left"><font color="#2f5597">durchgezogen</font></td>'
|
|
'<td align="left">Uebergabe / Ein-Ausschleusung</td></tr>'
|
|
'<tr><td align="left"><font color="#8ea9db">blau gestrichelt</font></td>'
|
|
'<td align="left">Kreisel-Umlauf</td></tr>'
|
|
'<tr><td align="left"><font color="#cc8800">orange</font></td>'
|
|
'<td align="left">Weiche (Kreisel-Uebergang)</td></tr>'
|
|
'<tr><td align="left">grau gepunktet</td>'
|
|
'<td align="left">gleiche Bahn / Strecke</td></tr>'
|
|
"</table>>")
|
|
if use_coords:
|
|
lx, ly = _legend_pos(analysis)
|
|
add(f' legende [shape=plaintext, style="", fillcolor="none", '
|
|
f'label={legend_label}, pos="{lx * COORD_SCALE:.2f},'
|
|
f'{ly * COORD_SCALE:.2f}"];')
|
|
else:
|
|
add(" subgraph cluster_legende {")
|
|
add(' label="Legende"; style="rounded"; color="#a6a6a6";')
|
|
add(' fontname="Segoe UI"; fontsize=10; fontcolor="#404040";')
|
|
add(f' legende [shape=plaintext, style="", fillcolor="none", '
|
|
f'label={legend_label}];')
|
|
add(" }")
|
|
add("}")
|
|
|
|
return "\n".join(out) + "\n"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fehlerdatei
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def render_errors(analysis: Analysis, csv_file: Path) -> str:
|
|
out = [GENERATED_MARKER, f"# TRO-Fehler - {csv_file.name}", ""]
|
|
out.append(f"**Erzeugt:** {datetime.now().strftime('%Y-%m-%d %H:%M')} ")
|
|
out.append(f"**Quelle:** `{csv_file}` ")
|
|
out.append(f"**Typdefinition:** `lib/tro_catalog.py` (Stand {CATALOG_AS_OF})")
|
|
out.append("")
|
|
|
|
if analysis.isolated:
|
|
out.append("## Nicht verbundene TROs")
|
|
out.append("")
|
|
out.append(
|
|
"Jeder TRO muss mit mindestens einem anderen TRO verbunden sein. "
|
|
"Fuer die folgenden Uebergabestellen liess sich im Materialfluss kein "
|
|
"Nachbar-TRO finden - weder ueber die Spalte `Nachbarn` noch ueber einen "
|
|
"gemeinsamen Host."
|
|
)
|
|
out.append("")
|
|
out.append("| TRO | Typ | Host | Separatoren | Bezeichnung |")
|
|
out.append("|---|---|---|---|---|")
|
|
for tro in analysis.isolated:
|
|
out.append(
|
|
f"| `{tro.tro_id}` | `{tro.type_name}` | "
|
|
f"{', '.join(f'`{h}`' for h in tro.hosts)} | "
|
|
f"{', '.join(tro.separators) or '-'} | {tro.label} |"
|
|
)
|
|
out.append("")
|
|
out.append("**Moegliche Ursachen**")
|
|
out.append("")
|
|
out.append("- Spalte `Nachbarn` des Host-Objekts ist leer oder unvollstaendig")
|
|
out.append("- `Nachbarn` verweist auf eine unbekannte `TeileId`")
|
|
out.append("- Die Bahnangabe (`-L` / `-R`) eines Kreisels fehlt auf beiden Seiten")
|
|
out.append("")
|
|
|
|
if analysis.unknown_types:
|
|
out.append("## Typen ohne Katalogeintrag")
|
|
out.append("")
|
|
out.append(
|
|
"Die folgenden Typen wurden ermittelt, stehen aber nicht in "
|
|
"`TRO_CATALOG` in `lib/tro_catalog.py`:"
|
|
)
|
|
out.append("")
|
|
for name in analysis.unknown_types:
|
|
holders = [t.tro_id for t in analysis.tros if t.type_name == name]
|
|
out.append(f"- `{name}` ({', '.join(holders)})")
|
|
out.append("")
|
|
|
|
return "\n".join(out) + "\n"
|
|
|
|
|
|
def write_error_file(analysis: Analysis, csv_file: Path, path: Path) -> bool:
|
|
"""Fehlerdatei schreiben; veraltete eigene Datei bei fehlerfreiem Lauf entfernen."""
|
|
if analysis.isolated or analysis.unknown_types:
|
|
path.write_text(render_errors(analysis, csv_file), encoding="utf-8")
|
|
return True
|
|
|
|
if path.is_file():
|
|
try:
|
|
first = path.read_text(encoding="utf-8").splitlines()[:1]
|
|
except OSError:
|
|
first = []
|
|
if first and first[0].strip() == GENERATED_MARKER:
|
|
path.unlink()
|
|
return False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Dokumentation (Markdown)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _mermaid_text(text: str) -> str:
|
|
return text.replace('"', "'").replace("[", "(").replace("]", ")")
|
|
|
|
|
|
def _tro_mermaid(analysis: Analysis) -> list[str]:
|
|
out = ["```mermaid", "flowchart LR"]
|
|
out.extend(f" {line}" for line in mermaid_classdefs())
|
|
|
|
for tro in analysis.tros:
|
|
definition = tro.definition()
|
|
group = definition.get_group() if definition else "ext"
|
|
label = f"{tro.tro_id}<br/>{tro.type_name}<br/>{_mermaid_text(tro.label)}"
|
|
out.append(f' {tro.tro_id}["{label}"]:::{group}')
|
|
|
|
for boundary in analysis.boundaries:
|
|
label = f"{boundary.tro_id}<br/>{boundary.type_name}<br/>{_mermaid_text(boundary.label)}"
|
|
out.append(f' {boundary.tro_id}(["{label}"]):::ext')
|
|
|
|
for src, dst in analysis.edges:
|
|
out.append(f" {src} --> {dst}")
|
|
|
|
seen: set[frozenset] = set()
|
|
for tro in analysis.tros:
|
|
for other in tro.colocated:
|
|
key = frozenset((tro.tro_id, other))
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
out.append(f" {tro.tro_id} -. gleiche Bahn .- {other}")
|
|
|
|
out.append("```")
|
|
return out
|
|
|
|
|
|
def render_doc(
|
|
analysis: Analysis,
|
|
graph: Graph,
|
|
csv_file: Path,
|
|
dot_file: Path,
|
|
svg_file: Path | None,
|
|
error_file: Path | None,
|
|
) -> str:
|
|
out: list[str] = [GENERATED_MARKER]
|
|
add = out.append
|
|
|
|
add(f"# TRO-Liste und TRO-Fluss - {csv_file.name}")
|
|
add("")
|
|
add(f"**Erzeugt:** {datetime.now().strftime('%Y-%m-%d %H:%M')} von "
|
|
f"`lib/tro_flow.py` ")
|
|
add(f"**Quelle:** `{csv_file}` ")
|
|
add(f"**Methode:** {analysis.method} ")
|
|
add(f"**Typdefinition:** `lib/tro_catalog.py` (Stand {CATALOG_AS_OF})")
|
|
add("")
|
|
add("> Automatisch aus dem CSV-Export erzeugt. Die TRO-Typen sind aus dem "
|
|
"mechanischen Layout **abgeleitet** - ohne E-Planung (I/O-Liste) und ohne "
|
|
"`FB_Main`-Aufrufstellen gibt es noch keine verbindliche TRO-Nummerierung. "
|
|
"Die IDs unten sind laufende Nummern dieses Laufs.")
|
|
add("")
|
|
|
|
types_used: dict[str, int] = {}
|
|
for tro in analysis.tros:
|
|
types_used[tro.type_name] = types_used.get(tro.type_name, 0) + 1
|
|
|
|
add("## 1. Kurzfassung")
|
|
add("")
|
|
add("| | |")
|
|
add("|---|---|")
|
|
add(f"| Ermittelte TROs | {len(analysis.tros)} |")
|
|
add(f"| Systemgrenzen (Materialein-/ausgang) | {len(analysis.boundaries)} |")
|
|
add(f"| TRO-Verbindungen | {len(analysis.edges)} |")
|
|
add(f"| Nicht verbundene TROs | {len(analysis.isolated)} |")
|
|
add(f"| Hinweise | {len(analysis.findings)} |")
|
|
if analysis.overrides.applied:
|
|
add(f"| Manuelle Typkorrekturen | {len(analysis.overrides.applied)} |")
|
|
if analysis.overrides.open_points:
|
|
add(f"| Offene Punkte (noch kein TRO) | "
|
|
f"{len(analysis.overrides.open_points)} |")
|
|
add("")
|
|
add("| Typ | FB-Baustein | Anzahl hier | Bauteile je TRO | Farbgruppe | Symbol |")
|
|
add("|---|---|---:|---|---|---|")
|
|
for name, count in sorted(types_used.items()):
|
|
definition = get_tro(name)
|
|
add(f"| `{name}` | "
|
|
f"{f'`{definition.get_fb_block()}`' if definition else '_kein Katalogeintrag_'} "
|
|
f"| {count} | {definition.describe_items() if definition else '-'} "
|
|
f"| {definition.get_group() if definition else '-'} "
|
|
f"| {definition.describe_symbol() if definition else '-'} |")
|
|
add("")
|
|
|
|
add("## 2. TRO-Flussdiagramm")
|
|
add("")
|
|
out.extend(_tro_mermaid(analysis))
|
|
add("")
|
|
add(f"Gerendert: [`{dot_file.name}`]({dot_file.name})"
|
|
+ (f" - [`{svg_file.name}`]({svg_file.name})" if svg_file else ""))
|
|
add("")
|
|
|
|
add("## 3. TRO-Liste")
|
|
add("")
|
|
add("| TRO | Typ | FB-Baustein | Bauteile laut Typ | X [mm] | Y [mm] | Z [mm] "
|
|
"| Host-Knoten | Separatoren | Scanner | Begruendung | Vertrauen |")
|
|
add("|---|---|---|---|---:|---:|---:|---|---|---|---|---|")
|
|
for tro in analysis.tros:
|
|
hosts = ", ".join(f"`{h}`" for h in tro.hosts)
|
|
if len(tro.hosts) > 4:
|
|
hosts = f"`{tro.hosts[0]}` … `{tro.hosts[-1]}` ({len(tro.hosts)} Knoten)"
|
|
seps = ", ".join(tro.separators) or "-"
|
|
if len(tro.separators) > 6:
|
|
seps = f"{tro.separators[0]} … {tro.separators[-1]} ({len(tro.separators)})"
|
|
items = ", ".join(f"{c}x {i}" for i, c in sorted(tro.items.items())) or "-"
|
|
cx = f"{tro.x:.0f}" if tro.x is not None else "-"
|
|
cy = f"{tro.y:.0f}" if tro.y is not None else "-"
|
|
cz = f"{tro.z:.0f}" if tro.z is not None else "-"
|
|
add(f"| `{tro.tro_id}` | `{tro.type_name}` | "
|
|
f"{f'`{tro.fb_block}`' if tro.fb_block else '-'} | {items} "
|
|
f"| {cx} | {cy} | {cz} | {hosts} | {seps} | "
|
|
f"{', '.join(tro.scanners) or '-'} | {tro.reason} | {tro.confidence} |")
|
|
add("")
|
|
add("**X/Y/Z** ist der Mittelpunkt der Bauteile, aus denen der TRO gebildet "
|
|
"wurde (Separatoren und Scanner) - beim Linienspeicher also der "
|
|
"Mittelwert aller Linien-Separatoren. Koordinaten in mm im "
|
|
"Anlagen-Koordinatensystem des CSV-Exports.")
|
|
add("")
|
|
add("| TRO | Koordinate aus | Anzahl Punkte |")
|
|
add("|---|---|---:|")
|
|
for tro in analysis.tros:
|
|
add(f"| `{tro.tro_id}` | {tro.position_from} | {tro.position_points} |")
|
|
add("")
|
|
|
|
add("## 4. TRO-Verbindungen")
|
|
add("")
|
|
if analysis.edges:
|
|
add("Richtung des Materialflusses zwischen den Uebergabestellen. "
|
|
"Passive Zwischenknoten ohne eigenen TRO sind zusammengefasst.")
|
|
add("")
|
|
add("| von | nach | Typ von -> Typ nach |")
|
|
add("|---|---|---|")
|
|
for src, dst in analysis.edges:
|
|
a, b = analysis.by_id(src), analysis.by_id(dst)
|
|
add(f"| `{src}` | `{dst}` | "
|
|
f"`{a.type_name if a else '?'}` -> `{b.type_name if b else '?'}` |")
|
|
else:
|
|
add("_Keine TRO-Verbindungen ermittelt._")
|
|
add("")
|
|
|
|
if analysis.isolated:
|
|
add("### Nicht verbundene TROs")
|
|
add("")
|
|
for tro in analysis.isolated:
|
|
add(f"- `{tro.tro_id}` (`{tro.type_name}`) - {tro.label}")
|
|
add("")
|
|
if error_file:
|
|
add(f"Details: [`{error_file.name}`]({error_file.name})")
|
|
add("")
|
|
|
|
if analysis.boundaries:
|
|
add("### Systemgrenzen (Materialein-/ausgang)")
|
|
add("")
|
|
add("BTMT Be-/Entladestationen - kein TRO (kein Separator, kein "
|
|
"FB-Baustein), aber der tatsaechliche Anfang bzw. das Ende des "
|
|
"Materialflusses. Ohne diese Zeile wuerde die Verbindung an der "
|
|
"Station stillschweigend enden (siehe `derive_boundaries` in "
|
|
"`lib/tro_flow.py`).")
|
|
add("")
|
|
add("| Knoten | Typ | Bezeichnung | verbunden mit |")
|
|
add("|---|---|---|---|")
|
|
for boundary in analysis.boundaries:
|
|
other = ", ".join(f"`{x}`" for x in
|
|
sorted(set(boundary.successors) | set(boundary.predecessors))) or "-"
|
|
add(f"| `{boundary.tro_id}` | `{boundary.type_name}` | "
|
|
f"{boundary.label} | {other} |")
|
|
add("")
|
|
|
|
add("## 5. Hinweise")
|
|
add("")
|
|
if analysis.findings:
|
|
for finding in analysis.findings:
|
|
add(f"- {finding}")
|
|
else:
|
|
add("_Keine._")
|
|
add("")
|
|
|
|
add("## 6. Manuelle Korrekturen und offene Punkte")
|
|
add("")
|
|
overrides = analysis.overrides
|
|
if not overrides and not overrides.warnings:
|
|
add("_Keine._ Alles unten Stehende ist rein aus dem Layout abgeleitet.")
|
|
add("")
|
|
else:
|
|
add(f"Quelle: `{overrides.source}`" if overrides.source
|
|
else "_Keine Override-Datei gefunden._")
|
|
add("")
|
|
|
|
if overrides.applied:
|
|
add("### 6.1 Angewandte Typkorrekturen")
|
|
add("")
|
|
add("Diese Typen stammen **nicht** aus dem Layout, sondern aus der "
|
|
"Begehung bzw. dem Review - im Layout sind sie nicht erkennbar.")
|
|
add("")
|
|
add("| Korrektur | Begruendung |")
|
|
add("|---|---|")
|
|
for entry in overrides.applied:
|
|
head, _, tail = entry.partition(" (")
|
|
add(f"| `{head}` | {tail[:-1] if tail.endswith(')') else tail} |")
|
|
add("")
|
|
|
|
if overrides.open_points:
|
|
add("### 6.2 Offene Punkte - noch kein TRO")
|
|
add("")
|
|
add("Im Review gemeldete Stellen. Die TRO-Struktur ist hier "
|
|
"**bewusst unveraendert**; die Punkte sind nur markiert (auch in der "
|
|
"annotierten Zeichnung, Layer `TRO_OPENPOINT`), damit sie in BricsCAD "
|
|
"geprueft werden koennen.")
|
|
add("")
|
|
add("| Punkt | X [mm] | Y [mm] | Z [mm] | erwartet | heute Teil von | Anmerkung |")
|
|
add("|---|---:|---:|---:|---|---|---|")
|
|
for point in overrides.open_points:
|
|
z_text = f"{point.z:.0f}" if point.z is not None else "-"
|
|
expect = f"`{point.expect}`" if point.expect else "-"
|
|
owner = f"`{point.belongs_to}`" if point.belongs_to else "-"
|
|
add(f"| {point.name} | {point.x:.0f} | {point.y:.0f} | {z_text} "
|
|
f"| {expect} | {owner} | {point.note or '-'} |")
|
|
add("")
|
|
|
|
if overrides.warnings:
|
|
add("### 6.3 Probleme mit der Override-Datei")
|
|
add("")
|
|
for warning in overrides.warnings:
|
|
add(f"- {warning}")
|
|
add("")
|
|
|
|
add("## 7. Ableitungsregeln und Typdefinition")
|
|
add("")
|
|
add("Ein TRO ist die Steuerungslogik **einer Uebergabestelle**. Ausgangspunkt "
|
|
"ist der Separator und das, was an ihm haengt (Host-Objekt, Scanner).")
|
|
add("")
|
|
add("| Host des Separators | TRO-Typ |")
|
|
add("|---|---|")
|
|
add(f"| Gefaellestrecke, >= {MIN_STORE_LANES} Linien mit gemeinsamem Zu-/Ablauf "
|
|
"| `PinStore_Auto` (ein Block je Liniengruppe) |")
|
|
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("| zusaetzlich: 2 bzw. 3 abgehende Wege | Aufwertung zu `1Sep1Swi` / `1Sep2Swi` |")
|
|
add("")
|
|
add("Ein **Scanner** am Separator aendert den Typ nicht. In der Referenzanlage "
|
|
"sind Separatoren mit Scanner ueberwiegend `1Sep`. `1Sep_SSCC` ist dem "
|
|
"Sonderfall mit SSCC-Scanner, Endmessung und WCS-Telegramm vorbehalten und "
|
|
"aus einem mechanischen Layout nicht erkennbar - betroffene Stellen stehen "
|
|
"als Hinweis in Abschnitt 5.")
|
|
add("")
|
|
add("Was aus dem CSV **nicht** ableitbar ist: TRO-Nummerierung, Sensor- und "
|
|
"Aktor-Tags, JamAreas, Zielsteuerung und Timing. Das kommt aus der "
|
|
"E-Planung (I/O-Liste) bzw. aus `FB_Main`.")
|
|
add("")
|
|
add(f"Typen, FB-Bausteine, Bauteile und Farben stammen vollstaendig aus "
|
|
f"`TRO_CATALOG` in `lib/tro_catalog.py` (Stand {CATALOG_AS_OF}):")
|
|
add("")
|
|
add("| # | Typ | FB-Baustein | Bauteile je TRO | Farbgruppe | Symbol |")
|
|
add("|---:|---|---|---|---|---|")
|
|
for index, definition in enumerate(TRO_CATALOG, start=1):
|
|
add(f"| {index} | `{definition.get_name()}` | `{definition.get_fb_block()}` "
|
|
f"| {definition.describe_items()} | {definition.get_group()} "
|
|
f"| {definition.describe_symbol()} |")
|
|
add("")
|
|
add("| Bauteil | Sub-FB |")
|
|
add("|---|---|")
|
|
for item in sorted({i for d in TRO_CATALOG for i in d.get_item_names()}):
|
|
fb = ITEM_FB.get(item, "")
|
|
add(f"| {item} | {f'`{fb}`' if fb else '_kein eigener FB_'} |")
|
|
add("")
|
|
add("| Farbgruppe | Bedeutung | Fuellfarbe |")
|
|
add("|---|---|---|")
|
|
for style in STYLES:
|
|
add(f"| {style.group} | {style.label} | `{style.fill}` |")
|
|
add("")
|
|
|
|
return "\n".join(out) + "\n"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Bericht
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def report(
|
|
analysis: Analysis,
|
|
csv_file: Path,
|
|
dot_file: Path,
|
|
svg_file: Path | None,
|
|
error_file: Path | None,
|
|
doc_file: Path | None,
|
|
warnings: list[str],
|
|
engine: str = "dot",
|
|
) -> None:
|
|
types_used: dict[str, int] = {}
|
|
for tro in analysis.tros:
|
|
types_used[tro.type_name] = types_used.get(tro.type_name, 0) + 1
|
|
|
|
print("")
|
|
print("================================================================")
|
|
print("TRO-FLUSS")
|
|
print("================================================================")
|
|
print(f"Eingabe = {csv_file}")
|
|
print(f"Methode = {analysis.method}")
|
|
print(f"TROs = {len(analysis.tros)}"
|
|
+ (f" ({', '.join(f'{k}: {v}' for k, v in sorted(types_used.items()))})"
|
|
if types_used else ""))
|
|
print(f"TRO-Verbindungen = {len(analysis.edges)}, "
|
|
f"nicht verbunden: {len(analysis.isolated)}")
|
|
print(f"Typdefinition = lib/tro_catalog.py (Stand {CATALOG_AS_OF}), "
|
|
f"{len(TRO_CATALOG)} Typen")
|
|
print(f"Layout = {engine}"
|
|
+ (" (lagerichtig, neato -n auf Anlagenkoordinaten, "
|
|
f"{COORD_SCALE} pt/mm)" if engine == "neato"
|
|
else " (berechnete Anordnung; lagerichtig mit --use-cords)"))
|
|
print(f"DOT = {dot_file}")
|
|
print(f"SVG = {svg_file if svg_file else '- (ohne --tosvg)'}")
|
|
print(f"Doku = {doc_file if doc_file else '- (ohne --doc)'}")
|
|
print(f"Fehlerdatei = {error_file if error_file else '- (keine Fehler)'}")
|
|
print("================================================================")
|
|
|
|
if analysis.isolated:
|
|
print("")
|
|
print(f"FEHLER: {len(analysis.isolated)} TRO(s) ohne Verbindung zu einem "
|
|
f"anderen TRO:")
|
|
for tro in analysis.isolated:
|
|
print(f" x {tro.tro_id} ({tro.type_name}) - {tro.label}")
|
|
|
|
if analysis.unknown_types:
|
|
print("")
|
|
print("FEHLER: TRO-Typen ohne Eintrag in TRO_CATALOG: "
|
|
+ ", ".join(analysis.unknown_types))
|
|
|
|
overrides = analysis.overrides
|
|
if overrides.applied:
|
|
print("")
|
|
print(f"Manuelle Typkorrekturen ({len(overrides.applied)}) aus "
|
|
f"{overrides.source}:")
|
|
for entry in overrides.applied:
|
|
print(f" * {entry}")
|
|
|
|
if overrides.open_points:
|
|
print("")
|
|
print(f"Offene Punkte ({len(overrides.open_points)}) - nur markiert, "
|
|
f"TRO-Struktur unveraendert:")
|
|
for point in overrides.open_points:
|
|
expect = f" erwartet {point.expect}" if point.expect else ""
|
|
owner = f", heute Teil von {point.belongs_to}" if point.belongs_to else ""
|
|
print(f" o {point.name} bei {point.describe_position()}{expect}{owner}")
|
|
|
|
if overrides.warnings:
|
|
print("")
|
|
print(f"Probleme mit {overrides.source} ({len(overrides.warnings)}):")
|
|
for warning in overrides.warnings:
|
|
print(f" ! {warning}")
|
|
|
|
if analysis.findings:
|
|
print("")
|
|
print(f"Hinweise ({len(analysis.findings)}):")
|
|
for finding in analysis.findings:
|
|
print(f" ? {finding}")
|
|
|
|
if warnings:
|
|
print("")
|
|
print(f"Warnungen aus dem Materialfluss ({len(warnings)}):")
|
|
for warning in warnings:
|
|
print(f" ! {warning}")
|
|
print("")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(
|
|
prog="tro_flow",
|
|
description="Ermittelt aus einem CSV-Export der Anlagenobjekte die "
|
|
"benoetigten TROs und zeichnet ihren Fluss (DOT, optional SVG).",
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
epilog="Beispiele:\n"
|
|
" tro_flow.bat --file mubea.csv\n"
|
|
" tro_flow.bat --file mubea.csv --tosvg\n"
|
|
" tro_flow.bat --file mubea.csv --tosvg --doc\n"
|
|
" tro_flow.bat --file mubea.csv --tosvg --use-cords\n"
|
|
"\n"
|
|
"Exit-Codes:\n"
|
|
" 0 ok\n"
|
|
" 1 Eingabe- oder Aufruffehler\n"
|
|
" 2 SVG konnte nicht erzeugt werden (Graphviz)\n"
|
|
" 3 TRO-Fehler gefunden, Fehlerdatei geschrieben\n",
|
|
)
|
|
parser.add_argument(
|
|
"--file",
|
|
default="export.csv",
|
|
metavar="NAME",
|
|
help="Name der CSV-Eingabedatei in %%SKEL_DATA%% "
|
|
"(oder ein vollstaendiger Pfad). Standard: %(default)s",
|
|
)
|
|
parser.add_argument(
|
|
"--tosvg",
|
|
action="store_true",
|
|
help="Aus der erzeugten DOT-Datei zusaetzlich das TRO-Flussdiagramm als "
|
|
"SVG erzeugen (benoetigt Graphviz 'dot').",
|
|
)
|
|
parser.add_argument(
|
|
"--use-cords",
|
|
"--use-coords",
|
|
dest="use_cords",
|
|
action="store_true",
|
|
help="Lagerichtig zeichnen: jeder TRO wird auf seine Anlagenkoordinate "
|
|
"gesetzt und das SVG von Graphviz 'neato -n' erzeugt statt von "
|
|
f"'dot'. Massstab {COORD_SCALE} pt/mm. Nur mit --tosvg wirksam.",
|
|
)
|
|
parser.add_argument(
|
|
"--doc",
|
|
action="store_true",
|
|
help="Dokumentation der TROs als Markdown in %%SKEL_RESULTS%% erzeugen "
|
|
"(TRO-Liste, Verbindungen, Typdefinition).",
|
|
)
|
|
parser.add_argument(
|
|
"--geometry",
|
|
action="store_true",
|
|
help="Separator-Host und Weichen-Fanout per STRtree direkt aus den "
|
|
"Bauteil-Weltkoordinaten bestimmen (lib/tro_geometry.py) statt "
|
|
"aus den CSV-Spalten 'Zuordnung'/'Nachbarn' - unabhaengig davon, "
|
|
"ob diese Spalten fuer die Eingabedatei luecken haben.",
|
|
)
|
|
parser.add_argument(
|
|
"--geo-tol-mm",
|
|
type=float,
|
|
default=tro_geometry_default_tolerance_mm(),
|
|
metavar="MM",
|
|
help="Nur mit --geometry: Toleranz (mm), innerhalb derer zwei "
|
|
"Anschlusspunkte als derselbe Uebergabepunkt gelten. "
|
|
"Standard: %(default)s",
|
|
)
|
|
return parser.parse_args(argv)
|
|
|
|
|
|
def tro_geometry_default_tolerance_mm() -> float:
|
|
"""Default fuer --geo-tol-mm; importiert tro_geometry nur bei Bedarf, damit
|
|
--help/normale Laeufe ohne --geometry kein shapely brauchen."""
|
|
try:
|
|
from tro_geometry import DEFAULT_TOLERANCE_MM
|
|
except ImportError:
|
|
return 50.0
|
|
return DEFAULT_TOLERANCE_MM
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = parse_args(argv)
|
|
|
|
csv_file = resolve_input(args.file)
|
|
if csv_file.suffix.lower() != ".csv":
|
|
print(f"FEHLER: --file erwartet eine CSV-Datei, keine '{csv_file.suffix}'-Datei: "
|
|
f"{csv_file}", file=sys.stderr)
|
|
return 1
|
|
if not csv_file.is_file():
|
|
print(f"FEHLER: Eingabedatei nicht gefunden: {csv_file}", file=sys.stderr)
|
|
print(f" Suchpfad fuer Dateinamen: {env_dir('SKEL_DATA', 'data')}",
|
|
file=sys.stderr)
|
|
return 1
|
|
|
|
results = env_dir("SKEL_RESULTS", "results")
|
|
results.mkdir(parents=True, exist_ok=True)
|
|
|
|
warnings: list[str] = []
|
|
try:
|
|
elements = read_elements(csv_file, warnings)
|
|
except (OSError, ValueError, csv.Error) as exc:
|
|
print(f"FEHLER: {csv_file} konnte nicht gelesen werden: {exc}", file=sys.stderr)
|
|
return 1
|
|
|
|
if not elements:
|
|
print(f"FEHLER: {csv_file} enthaelt keine Objekte.", file=sys.stderr)
|
|
return 1
|
|
|
|
graph = build_graph(elements, csv_file.name, warnings)
|
|
if not graph.nodes:
|
|
print(f"FEHLER: {csv_file} enthaelt keine Flussobjekte.", file=sys.stderr)
|
|
return 1
|
|
|
|
resolve_host = count_exits = None
|
|
method = None
|
|
if args.geometry:
|
|
try:
|
|
from tro_geometry import make_geometric_resolvers
|
|
except ImportError as exc:
|
|
print(f"FEHLER: --geometry braucht shapely (pip install shapely): {exc}",
|
|
file=sys.stderr)
|
|
return 1
|
|
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)"
|
|
|
|
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)
|
|
if not analysis.tros:
|
|
print(f"FEHLER: {csv_file} enthaelt keine Separatoren - keine TROs "
|
|
f"ableitbar.", file=sys.stderr)
|
|
return 1
|
|
|
|
stem = csv_file.stem + ("_geo" if args.geometry else "")
|
|
dot_file = results / f"{stem}_tro_flow.dot"
|
|
svg_file = results / f"{stem}_tro_flow.svg"
|
|
doc_target = results / f"{stem}_tro_doc.md"
|
|
error_target = results / f"{stem}_tro_errors.md"
|
|
|
|
# Ohne Koordinaten ist kein lagerichtiges Layout moeglich
|
|
without = missing_positions(analysis)
|
|
if args.use_cords and without:
|
|
print(f"FEHLER: --use-cords nicht moeglich, {len(without)} TRO(s) ohne "
|
|
f"Koordinate: {', '.join(without)}", file=sys.stderr)
|
|
print(" Ohne pos an jedem Knoten kann 'neato -n' nicht zeichnen.",
|
|
file=sys.stderr)
|
|
return 1
|
|
|
|
dot_file.write_text(
|
|
render_dot(analysis, csv_file.name, use_coords=args.use_cords),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
has_errors = write_error_file(analysis, csv_file, error_target)
|
|
error_file = error_target if has_errors else None
|
|
|
|
engine = "neato" if args.use_cords else "dot"
|
|
svg_error: str | None = None
|
|
if args.tosvg:
|
|
try:
|
|
graph_to_svg(dot_file, svg_file, engine=engine,
|
|
no_op=1 if args.use_cords else 0)
|
|
except RuntimeError as exc:
|
|
svg_error = str(exc)
|
|
created_svg = svg_file if args.tosvg and svg_error is None else None
|
|
|
|
doc_file = None
|
|
if args.doc:
|
|
doc_target.write_text(
|
|
render_doc(analysis, graph, csv_file, dot_file, created_svg, error_file),
|
|
encoding="utf-8",
|
|
)
|
|
doc_file = doc_target
|
|
|
|
report(analysis, csv_file, dot_file, created_svg, error_file, doc_file,
|
|
warnings, engine=engine)
|
|
|
|
if svg_error:
|
|
print(f"FEHLER: SVG nicht erzeugt: {svg_error}", file=sys.stderr)
|
|
return 2
|
|
if has_errors:
|
|
print(f"FEHLER: TRO-Pruefung fehlgeschlagen - siehe {error_target}",
|
|
file=sys.stderr)
|
|
return 3
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|