Schoenenberger_Systeme_GmbH/sps_skel
8
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

TRO flow svg file can be created.

Browse Source
This commit is contained in:
Samer Ayadi
2026-09-10 11:29:22 +02:00
parent 6fda4951e1
commit 4bab45705d
4 changed files with 738 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/tro_flow.py
+290 -25
View File
@@ -60,6 +60,7 @@ 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
@@ -88,6 +89,7 @@ from material_flow import (
from tro_catalog import (
CATALOG_AS_OF,
ITEM_FB,
STYLE_EXT,
STYLES,
TRO_CATALOG,
TroDefinition,
@@ -111,6 +113,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 Aus-/
# Einschleuselement" gilt (siehe _find_switch_separators). 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
GENERATED_MARKER = "<!-- erzeugt von lib/tro_flow.py -->"
# Zeichnungsmassstab fuer --use-cords: Graphviz rechnet in Punkt, die Anlage in
@@ -187,9 +197,14 @@ class Analysis:
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:
for tro in self.tros + self.boundaries:
if tro.tro_id == tro_id:
return tro
return None
@@ -243,13 +258,71 @@ def _switch_exits(hosts: list[str], graph: Graph) -> list[str]:
)
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 -> (naechstgelegenes Aus-/Einschleuselement, Abstand mm),
nur wenn Abstand <= 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) bzw. Einschleuselement (Strecke -> Kreisel,
am Streckenende) 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).
"""
transfers = [e for e in elements if e.kind in ("Ausschleus", "Einschleus")]
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])
return result
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."""
"""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
)
@@ -257,30 +330,39 @@ def derive_tros(
(e for e in elements if e.kind == "Scanner"), key=lambda e: e.teile_id
)
# Separator -> Host-Knoten (Spalte "Zuordnung")
# Separatoren an einem Aus-/Einschleuselement (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:
host = sep.assignment
if host and host != UNASSIGNED and host in graph.nodes:
if sep.teile_id in switch_hits:
continue
host = resolve_host(sep, findings)
if host is not None:
sep_host[sep.teile_id] = host
else:
findings.append(
f"Separator {sep.teile_id} '{sep.name}': Zuordnung '{host or '-'}' "
f"zeigt auf keinen Flussknoten - keinem TRO zugeordnet"
)
# 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:
if target in sep_host or target in switch_hits:
sep_scanners.setdefault(target, []).append(scanner.teile_id)
mounted = scanner.assignment
if mounted and mounted != UNASSIGNED and mounted != sep_host[target]:
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 {sep_host[target]}"
f"liest aber an Separator {target} auf {target_host}"
)
else:
findings.append(
@@ -380,11 +462,41 @@ def derive_tros(
)
)
# 3b) Separatoren an einem Aus-/Einschleuselement -> 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" im Merkmal, 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") or "")
ende = str(_merkmal(transfer.merkmale, "Streckenende") or "")
own_scanners = sorted(sep_scanners.get(sep_id, []))
art = "Ausschleuselement" if transfer.kind == "Ausschleus" else "Einschleuselement"
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 = _switch_exits(tro.hosts, graph)
exits = count_exits(tro.hosts)
if not 2 <= len(exits) <= 3:
continue
upgraded = "1Sep1Swi" if len(exits) == 2 else "1Sep2Swi"
@@ -443,6 +555,60 @@ def derive_tros(
return 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.
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.
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.
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 element.kind not in ("Beladung", "Entladung"):
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 (BTMT 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":
@@ -620,23 +786,38 @@ def locate_tros(tros: list[Tro], elements: list[Element], graph: Graph) -> None:
def analyse_tros(
graph: Graph, elements: list[Element], overrides: Overrides | None = None
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()
analysis.tros = derive_tros(graph, elements, analysis.findings, analysis.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(analysis.tros, elements, graph)
analysis.edges, analysis.edge_kinds = link_tros(graph, analysis.tros)
locate_tros(linked, elements, graph)
analysis.edges, analysis.edge_kinds = link_tros(graph, linked)
# 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(
analysis.tros, analysis.overrides
linked, analysis.overrides
)
gesetzt = set(analysis.edges)
for src, dst in sorted(abgeleitet - gesetzt):
@@ -688,14 +869,16 @@ def _tro_label(tro: Tro) -> str:
def missing_positions(analysis: Analysis) -> list[str]:
"""TROs ohne Koordinate - mit ihnen ist kein lagerichtiges Layout moeglich."""
return [t.tro_id for t in analysis.tros if not t.position]
"""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)."""
xs = [t.position[0] for t in analysis.tros if t.position]
ys = [t.position[1] for t in analysis.tros if t.position]
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)
@@ -764,6 +947,20 @@ def render_dot(analysis: Analysis, source: str, use_coords: bool = False) -> str
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)))}];')
@@ -903,6 +1100,10 @@ def _tro_mermaid(analysis: Analysis) -> list[str]:
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}")
@@ -935,6 +1136,7 @@ def render_doc(
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 "
@@ -952,6 +1154,7 @@ def render_doc(
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)} |")
@@ -1037,6 +1240,24 @@ def render_doc(
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:
@@ -1107,6 +1328,8 @@ 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 von einem Aus-/Einschleuselement "
"| `1Sep1Swi` (Element uebernimmt die fehlende Weiche) |")
add("| zusaetzlich: 2 bzw. 3 abgehende Wege | Aufwertung zu `1Sep1Swi` / `1Sep2Swi` |")
add("")
add("Ein **Scanner** am Separator aendert den Typ nicht. In der Referenzanlage "
@@ -1167,6 +1390,7 @@ def report(
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 ""))
@@ -1284,9 +1508,36 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
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)
@@ -1320,14 +1571,28 @@ def main(argv: list[str] | None = None) -> int:
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)
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
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"