Schoenenberger_Systeme_GmbH/sps_skel
8
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
849e985008f60039850c516cdfe716a0f170a3bc
sps_skel/lib/tro_flow.py
T
s.ayadi 849e985008 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) <noreply@anthropic.com>
2026-07-30 13:13:45 +02:00

1051 lines
38 KiB
Python
Raw Blame History

# -*- 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)
-------------------------------
<name>_tro_flow.dot immer
<name>_tro_flow.svg mit --tosvg (Graphviz)
<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 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 = "<!-- erzeugt von lib/tro_flow.py -->"
# ---------------------------------------------------------------------------
# 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'<tr><td bgcolor="{s.fill}" align="left"> {s.group} </td>'
f'<td align="left">{s.label}</td></tr>'
for s in STYLES
)
add(' legende [shape=plaintext, style="", fillcolor="none", label=<'
'<table border="0" cellborder="1" cellspacing="0" cellpadding="3">'
+ rows
+ '<tr><td align="left">durchgezogen</td><td align="left">Materialfluss</td></tr>'
'<tr><td align="left">gepunktet</td><td align="left">gleiche Bahn / Strecke</td></tr>'
"</table>>];")
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 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())
Reference in New Issue View Git Blame Copy Permalink