From 849e985008f60039850c516cdfe716a0f170a3bc Mon Sep 17 00:00:00 2001 From: Samer Ayadi Date: Thu, 30 Jul 2026 13:13:45 +0200 Subject: [PATCH] Add TRO flow program deriving the control objects from the layout Sister program to material_flow.py with the same switches, but --tosvg draws the flow diagram of the TROs instead of the individual objects. It reuses read_elements and build_graph, so CSV parsing and direction logic exist once. A TRO is the control logic of one hand-over point, derived from the separator and what is attached to it, following doc/500573_Mubea/TRO_Identifikation_500573.md: Gefaellestrecke, >= 2 lines sharing feed/discharge -> PinStore_Auto per group Gefaellestrecke, single line -> 1Sep Strecke (driven) -> Vario per segment Kreisel lane -> 1Sep A scanner at the separator deliberately does not change the type: in connect.ini separators with a scanner are predominantly 1Sep, and 1Sep_SSCC is reserved for the SSCC/end-measurement/WCS case, which a mechanical layout cannot reveal. Affected points are reported as hints instead. A host with 2 or 3 outgoing flow edges upgrades 1Sep to 1Sep1Swi/1Sep2Swi. Connectivity contracts passive nodes: material runs separator to separator, so a carousel lane carrying no TRO is traversed rather than treated as a dead end. Without that, a lone separator on a spur loop would be reported as an orphan purely as an artifact of the model. Each TRO must connect to at least one other; otherwise an error file is written and the exit code is 3. Each TRO also carries a plant coordinate - the centroid of the components it was built from, so a 1Sep1Swi sits between its separator and its switch. Written into the DOT as a pos attribute (1:100), which dot ignores and "neato -n" can use to draw the diagram to scale. Types, function blocks, components and colours come exclusively from lib/tro_catalog.py. Co-Authored-By: Claude Opus 5 (1M context) --- bin/tro_flow.bat | 33 ++ bin/tro_flow.sh | 34 ++ lib/tro_flow.py | 1050 ++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 1117 insertions(+) create mode 100644 bin/tro_flow.bat create mode 100644 bin/tro_flow.sh create mode 100644 lib/tro_flow.py diff --git a/bin/tro_flow.bat b/bin/tro_flow.bat new file mode 100644 index 0000000..9cd3ee9 --- /dev/null +++ b/bin/tro_flow.bat @@ -0,0 +1,33 @@ +@echo off +REM ================================================================ +REM SPS_SKEL - TROs ermitteln und TRO-Flussdiagramm erzeugen +REM ================================================================ +REM Aufruf: +REM bin\tro_flow.bat --file mubea.csv +REM bin\tro_flow.bat --file mubea.csv --tosvg +REM bin\tro_flow.bat --file mubea.csv --tosvg --doc +REM +REM Die CSV-Datei wird in %SKEL_DATA% gesucht, DOT, SVG, Doku und +REM Fehlerdatei werden in %SKEL_RESULTS% abgelegt. +REM +REM Schwesterprogramm: bin\material_flow.bat (Materialfluss der Objekte) +REM ================================================================ + +setlocal + +REM Umgebung setzen (SKEL_DATA, SKEL_RESULTS, SKEL_LIB, PYTHONPATH ...) +call "%~dp0setenv.bat" >nul + +REM Virtuelle Umgebung nutzen, falls vorhanden (pydantic wird benoetigt) +if exist "%SPS_SKEL%\.venv\Scripts\activate.bat" ( + call "%SPS_SKEL%\.venv\Scripts\activate.bat" +) + +REM Python-Interpreter bestimmen (py-Launcher bevorzugt, sonst python) +set "PY=py" +py --version >nul 2>&1 || set "PY=python" + +"%PY%" "%SKEL_LIB%\tro_flow.py" %* +set "RC=%ERRORLEVEL%" + +endlocal & exit /b %RC% diff --git a/bin/tro_flow.sh b/bin/tro_flow.sh new file mode 100644 index 0000000..a1b4745 --- /dev/null +++ b/bin/tro_flow.sh @@ -0,0 +1,34 @@ +#!/usr/bin/env bash +# ================================================================ +# SPS_SKEL - TROs ermitteln und TRO-Flussdiagramm erzeugen +# ================================================================ +# Aufruf: +# bash bin/tro_flow.sh --file mubea.csv +# bash bin/tro_flow.sh --file mubea.csv --tosvg +# bash bin/tro_flow.sh --file mubea.csv --tosvg --doc +# +# Die CSV-Datei wird in $SKEL_DATA gesucht, DOT, SVG, Doku und +# Fehlerdatei werden in $SKEL_RESULTS abgelegt. +# +# Schwesterprogramm: bin/material_flow.sh (Materialfluss der Objekte) +# ================================================================ + +set -eu + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" + +# Umgebung setzen (SKEL_DATA, SKEL_RESULTS, SKEL_LIB, PYTHONPATH ...) +# shellcheck source=/dev/null +. "$SCRIPT_DIR/setenv.sh" >/dev/null + +# Virtuelle Umgebung nutzen, falls vorhanden (pydantic wird benoetigt) +if [ -f "$SPS_SKEL/.venv/bin/activate" ]; then + # shellcheck source=/dev/null + . "$SPS_SKEL/.venv/bin/activate" +fi + +# Python-Interpreter bestimmen +PY="python3" +command -v python3 >/dev/null 2>&1 || PY="python" + +exec "$PY" "$SKEL_LIB/tro_flow.py" "$@" diff --git a/lib/tro_flow.py b/lib/tro_flow.py new file mode 100644 index 0000000..a00058b --- /dev/null +++ b/lib/tro_flow.py @@ -0,0 +1,1050 @@ +# -*- 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. + +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 Streckensegment) + Kreisel-Bahn -> 1Sep + +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) +------------------------------- + _tro_flow.dot immer + _tro_flow.svg mit --tosvg (Graphviz) + _tro_doc.md mit --doc + _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 sys +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, + UNASSIGNED, + Element, + Graph, + _as_float, + _dot_escape, + _dot_label, + _merkmal, + build_graph, + dot_to_svg, + env_dir, + read_elements, + resolve_input, +) +from tro_catalog import ( + CATALOG_AS_OF, + ITEM_FB, + STYLES, + TRO_CATALOG, + TroDefinition, + get_tro, + mermaid_classdefs, +) + +# --------------------------------------------------------------------------- +# Konstanten +# --------------------------------------------------------------------------- + +# Ab dieser Anzahl zusammenhaengender Schwerkraftlinien (Gefaellestrecke mit je +# einem Separator) wird ein Speicherblock (PinStore_Auto) angenommen. +MIN_STORE_LANES = 2 + +GENERATED_MARKER = "" + + +# --------------------------------------------------------------------------- +# 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) + isolated: list[Tro] = field(default_factory=list) + unknown_types: list[str] = field(default_factory=list) + findings: list[str] = field(default_factory=list) + + def by_id(self, tro_id: str) -> Tro | None: + for tro in self.tros: + 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 derive_tros(graph: Graph, elements: list[Element], findings: list[str]) -> list[Tro]: + """TRO-Liste aus den Separatoren und ihren angeschlossenen Objekten bilden.""" + 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 + ) + + # Separator -> Host-Knoten (Spalte "Zuordnung") + sep_host: dict[str, str] = {} + for sep in separators: + host = sep.assignment + if host and host != UNASSIGNED and host in graph.nodes: + 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: + sep_scanners.setdefault(target, []).append(scanner.teile_id) + mounted = scanner.assignment + if mounted and mounted != UNASSIGNED and mounted != sep_host[target]: + findings.append( + f"Scanner {scanner.teile_id} '{scanner.name}': montiert an {mounted}, " + f"liest aber an Separator {target} auf {sep_host[target]}" + ) + 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 Segment ein Vario + 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" + tros.append( + Tro( + tro_id="", + type_name="Vario", + label=element.name, + hosts=[node_id], + separators=seps, + scanners=scanners_for(seps), + reason=f"angetriebenes Streckensegment (Antriebfahrtrichtung: {drive})", + confidence="hoch", + ) + ) + + # 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", + ) + ) + + # 4) Weichen-Aufwertung: 2 bzw. 3 abgehende Flusskanten + for tro in tros: + if tro.type_name != "1Sep": + continue + exits = _switch_exits(tro.hosts, graph) + 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" + + # 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}" + + # 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 + + +def link_tros(graph: Graph, tros: list[Tro]) -> list[tuple[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. + """ + owners: dict[str, list[str]] = {} + for tro in tros: + for host in tro.hosts: + owners.setdefault(host, []).append(tro.tro_id) + + forward: dict[str, list[str]] = {} + for edge in graph.edges: + forward.setdefault(edge.src, []).append(edge.dst) + if edge.kind == EDGE_UNRESOLVED: + # unbestimmte Richtung: in beide Richtungen begehbar + forward.setdefault(edge.dst, []).append(edge.src) + + edges: set[tuple[str, str]] = set() + for tro in tros: + seen = set(tro.hosts) + queue = [n for host in tro.hosts for n in forward.get(host, [])] + while queue: + node = queue.pop() + if node in seen: + continue + seen.add(node) + if node in owners: + for other in owners[node]: + if other != tro.tro_id: + edges.add((tro.tro_id, other)) + continue + queue.extend(forward.get(node, [])) + + # gleicher Host = physisch dieselbe Strecke/Bahn + for host, ids in owners.items(): + for tro_id in ids: + tro = next(t for t in tros if t.tro_id == tro_id) + tro.colocated = sorted(set(tro.colocated) | {o for o in ids if o != tro_id}) + + 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) + + +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]) -> Analysis: + """TROs ermitteln, verknuepfen, verorten und gegen die Typdefinition pruefen.""" + analysis = Analysis() + analysis.tros = derive_tros(graph, elements, analysis.findings) + analysis.edges = link_tros(graph, analysis.tros) + locate_tros(analysis.tros, elements, graph) + + 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 render_dot(analysis: Analysis, source: str) -> str: + """Gerichtetes Flussdiagramm der TROs als Graphviz-Quelle.""" + out: list[str] = [] + add = out.append + + add("// Automatisch erzeugt von lib/tro_flow.py - nicht manuell aendern.") + add(f"// Quelle: {source}") + add("digraph TROFluss {") + 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: dot ignoriert pos, "neato -n" zeichnet + # damit den Graphen lagerichtig (Millimeter -> Punkt, 1:100). + pos = "" + if tro.position: + pos = f', pos="{tro.position[0] / 100:.2f},{tro.position[1] / 100:.2f}"' + add(f' "{tro.tro_id}" [label="{_tro_label(tro)}", {rendered}{pos}];') + add("") + + for src, dst in analysis.edges: + add(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) + add(f' "{tro.tro_id}" -> "{other}" [dir=none, style=dotted, ' + f'color="#7f7f7f", constraint=false, tooltip="gleiche Bahn"];') + add("") + + # Legende: eine Zeile je Farbgruppe, plus die Kantenarten + add(" subgraph cluster_legende {") + add(' label="Legende"; style="rounded"; color="#a6a6a6";') + add(' fontname="Segoe UI"; fontsize=10; fontcolor="#404040";') + rows = "".join( + f' {s.group} ' + f'{s.label}' + for s in STYLES + ) + add(' legende [shape=plaintext, style="", fillcolor="none", label=<' + '' + + rows + + '' + '' + "
durchgezogenMaterialfluss
gepunktetgleiche Bahn / Strecke
>];") + 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}
{tro.type_name}
{_mermaid_text(tro.label)}" + out.append(f' {tro.tro_id}["{label}"]:::{group}') + + 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"**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"| TRO-Verbindungen | {len(analysis.edges)} |") + add(f"| Nicht verbundene TROs | {len(analysis.isolated)} |") + add(f"| Hinweise | {len(analysis.findings)} |") + add("") + add("| Typ | FB-Baustein | Anzahl hier | Bauteile je TRO | Farbgruppe |") + 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 '-'} |") + 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("") + + add("## 5. Hinweise") + add("") + if analysis.findings: + for finding in analysis.findings: + add(f"- {finding}") + else: + add("_Keine._") + add("") + + add("## 6. 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("| 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 |") + 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()} |") + 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], +) -> 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"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"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)) + + 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" + "\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( + "--doc", + action="store_true", + help="Dokumentation der TROs als Markdown in %%SKEL_RESULTS%% erzeugen " + "(TRO-Liste, Verbindungen, Typdefinition).", + ) + return parser.parse_args(argv) + + +def main(argv: list[str] | None = None) -> int: + args = parse_args(argv) + + csv_file = resolve_input(args.file) + 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 + + analysis = analyse_tros(graph, elements) + 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 + 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" + + dot_file.write_text(render_dot(analysis, csv_file.name), encoding="utf-8") + + has_errors = write_error_file(analysis, csv_file, error_target) + error_file = error_target if has_errors else None + + svg_error: str | None = None + if args.tosvg: + try: + dot_to_svg(dot_file, svg_file) + 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) + + 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())