s.ayadi
035c86ced9
Adds a --use-cords switch to tro_flow.py (--use-coords accepted too). With it, every TRO node carries its plant coordinate as a pos attribute and the SVG is produced by "neato -n" instead of dot, so the diagram looks like the plant layout rather than a computed left-to-right flow. Scale is 0.25 pt/mm (COORD_SCALE), which turns the ~11 x 20 m plant into roughly 2800 x 5000 pt - large enough that the TRO boxes do not overlap. Without the switch the pos attributes stay in the DOT as information only, at 1:100, and dot ignores them; the default layout is unchanged. Two details neato -n needs: the legend gets a position of its own, placed below left of the plant bounding box, because with -n a node without pos would land at 0,0 in the middle of the drawing. And the switch is refused with a clear message if any TRO has no coordinate, since -n cannot lay out a graph with missing positions. Verified that neato -n reproduces the given coordinates: the positions it reports back differ from the input by a constant offset only (spread 0.08 pt, from Graphviz moving the drawing into the positive quadrant), and no node distance changes by more than 0.08 pt over spans of up to 3600 pt. graph_to_svg() in material_flow.py replaces dot_to_svg() to take an engine and the no-op flag; dot_to_svg() stays as a thin wrapper. It now also treats output on stderr as an error, so neato's "node has no position" warnings do not pass silently. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
1136 lines
42 KiB
Python
1136 lines
42 KiB
Python
# -*- coding: utf-8 -*-
|
|
# Absichtlich ohne Shebang: der Windows-py-Launcher wuerde daraus "python3"
|
|
# ableiten und scheitert, wenn keine PythonCore-Installation registriert ist.
|
|
"""
|
|
tro_flow.py - TROs aus einem CSV-Export ermitteln und ihren Fluss zeichnen.
|
|
|
|
Schwesterprogramm zu material_flow.py:
|
|
|
|
material_flow.py Materialfluss der *mechanischen Objekte*
|
|
(Kreisel, Gefaellestrecke, Strecke)
|
|
tro_flow.py Fluss der *Steuerungsobjekte* (TROs)
|
|
|
|
Gleiche Schalter (--file, --tosvg, --doc); --tosvg erzeugt hier das
|
|
Flussdiagramm der TROs, nicht das der Einzelobjekte.
|
|
|
|
Zusaetzlich --use-cords: jeder TRO traegt die Koordinate seiner Bauteile, und
|
|
das SVG wird mit "neato -n" lagerichtig gezeichnet statt von "dot" angeordnet.
|
|
Der Graph sieht dann aus wie das Anlagenlayout.
|
|
|
|
Was ein TRO ist
|
|
---------------
|
|
Ein TRO ("Transfer Route Object") ist die Steuerungslogik **einer
|
|
Uebergabestelle**. Ausgangspunkt ist der Separator (Vereinzeler/Sperre) und das,
|
|
was an ihm haengt (Host-Strecke, Scanner). Ableitungsregeln nach
|
|
doc/500573_Mubea/TRO_Identifikation_500573.md:
|
|
|
|
Host des Separators -> TRO-Typ
|
|
-----------------------------------------------------------------------
|
|
Gefaellestrecke, >= 2 Linien -> PinStore_Auto (ein Block je Gruppe)
|
|
Gefaellestrecke, einzelne Linie -> 1Sep
|
|
Strecke (angetrieben) -> Vario (je 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 (mit --use-cords lagerichtige pos-Attribute)
|
|
<name>_tro_flow.svg mit --tosvg (dot, mit --use-cords neato -n)
|
|
<name>_tro_doc.md mit --doc
|
|
<name>_tro_errors.md nur wenn TRO-Fehler gefunden wurden
|
|
|
|
Aufruf ueber bin/tro_flow.bat bzw. bin/tro_flow.sh.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
import 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,
|
|
env_dir,
|
|
graph_to_svg,
|
|
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 -->"
|
|
|
|
# Zeichnungsmassstab fuer --use-cords: Graphviz rechnet in Punkt, die Anlage in
|
|
# Millimeter. 0.25 pt/mm entspricht 1:4 - bei ~11 x 20 m Anlage ergibt das eine
|
|
# Zeichnung von etwa 2800 x 5000 pt, in der sich die TRO-Kaesten nicht ueberdecken.
|
|
COORD_SCALE = 0.25
|
|
|
|
# Abstand der Legende zum Anlagenrand in mm
|
|
LEGEND_MARGIN = 3000
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Datenmodell
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class Tro:
|
|
"""Eine ermittelte Uebergabestelle."""
|
|
|
|
tro_id: str
|
|
type_name: str
|
|
label: str
|
|
hosts: list[str]
|
|
separators: list[str]
|
|
scanners: list[str]
|
|
reason: str
|
|
confidence: str
|
|
fb_block: str = ""
|
|
items: dict[str, int] = field(default_factory=dict)
|
|
successors: list[str] = field(default_factory=list)
|
|
predecessors: list[str] = field(default_factory=list)
|
|
colocated: list[str] = field(default_factory=list)
|
|
# Mittelpunkt der Bauteile, aus denen dieser TRO gebildet wurde
|
|
position: tuple[float, float, float] | None = None
|
|
position_from: str = ""
|
|
position_points: int = 0
|
|
|
|
@property
|
|
def x(self) -> float | None:
|
|
return self.position[0] if self.position else None
|
|
|
|
@property
|
|
def y(self) -> float | None:
|
|
return self.position[1] if self.position else None
|
|
|
|
@property
|
|
def z(self) -> float | None:
|
|
return self.position[2] if self.position else None
|
|
|
|
def describe_position(self) -> str:
|
|
"""Koordinate als Text, z. B. '1030, 4700'."""
|
|
if not self.position:
|
|
return "-"
|
|
return f"{self.position[0]:.0f}, {self.position[1]:.0f}"
|
|
|
|
@property
|
|
def connected(self) -> bool:
|
|
return bool(self.successors or self.predecessors or self.colocated)
|
|
|
|
def definition(self) -> TroDefinition | None:
|
|
"""Typdefinition aus tro_catalog (None wenn Typ unbekannt)."""
|
|
return get_tro(self.type_name)
|
|
|
|
|
|
@dataclass
|
|
class Analysis:
|
|
tros: list[Tro] = field(default_factory=list)
|
|
edges: list[tuple[str, str]] = field(default_factory=list)
|
|
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 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]
|
|
|
|
|
|
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]
|
|
if not xs:
|
|
return (0.0, 0.0)
|
|
return (min(xs) - LEGEND_MARGIN, min(ys) - LEGEND_MARGIN)
|
|
|
|
|
|
def render_dot(analysis: Analysis, source: str, use_coords: bool = False) -> str:
|
|
"""
|
|
Gerichtetes Flussdiagramm der TROs als Graphviz-Quelle.
|
|
|
|
use_coords=False Anordnung berechnet 'dot' (Fluss von links nach rechts)
|
|
use_coords=True jeder Knoten wird auf seine Anlagenkoordinate gesetzt und
|
|
mit "neato -n" lagerichtig gezeichnet
|
|
"""
|
|
out: list[str] = []
|
|
add = out.append
|
|
|
|
add("// Automatisch erzeugt von lib/tro_flow.py - nicht manuell aendern.")
|
|
add(f"// Quelle: {source}")
|
|
if use_coords:
|
|
add(f"// Lagerichtig: Knoten auf Anlagenkoordinate, {COORD_SCALE} pt/mm.")
|
|
add("// Rendern mit: neato -n -Tsvg <datei>.dot -o <datei>.svg")
|
|
add("digraph TROFluss {")
|
|
if use_coords:
|
|
# Bei -n bestimmen die pos-Attribute die Lage; rankdir/ranksep entfallen.
|
|
add(" graph [splines=true, outputorder=edgesfirst,")
|
|
add(' fontname="Segoe UI", fontsize=11, labelloc="t",')
|
|
add(f' label="TRO-Fluss lagerichtig ({COORD_SCALE} pt/mm) - '
|
|
f'{_dot_escape(source)}"];')
|
|
else:
|
|
add(" graph [rankdir=LR, splines=spline, nodesep=0.35, ranksep=1.3,")
|
|
add(' fontname="Segoe UI", fontsize=11, labelloc="t",')
|
|
add(f' label="TRO-Fluss - {_dot_escape(source)}"];')
|
|
add(' node [shape=box, style="rounded,filled", fontname="Segoe UI", fontsize=9];')
|
|
add(' edge [fontname="Segoe UI", fontsize=8, color="#2f5597", penwidth=1.2,'
|
|
" arrowsize=0.8];")
|
|
add("")
|
|
|
|
for tro in analysis.tros:
|
|
definition = tro.definition()
|
|
attrs = definition.get_dot_attrs() if definition else {
|
|
"fillcolor": "#ffffff", "color": "#c00000", "fontcolor": "#c00000"
|
|
}
|
|
rendered = ", ".join(f'{k}="{v}"' for k, v in attrs.items())
|
|
# Anlagenkoordinate mitgeben. Ohne --use-cords ignoriert dot das pos,
|
|
# die Angabe bleibt aber als Information im DOT stehen.
|
|
pos = ""
|
|
if tro.position:
|
|
scale = COORD_SCALE if use_coords else 0.01
|
|
pos = (f', pos="{tro.position[0] * scale:.2f},'
|
|
f'{tro.position[1] * scale:.2f}"')
|
|
add(f' "{tro.tro_id}" [label="{_tro_label(tro)}", {rendered}{pos}];')
|
|
add("")
|
|
|
|
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.
|
|
# Bei -n braucht auch sie ein pos, sonst landet sie auf 0,0 mitten in der
|
|
# Anlage - daher unterhalb links vom Anlagenrand.
|
|
rows = "".join(
|
|
f'<tr><td bgcolor="{s.fill}" align="left"> {s.group} </td>'
|
|
f'<td align="left">{s.label}</td></tr>'
|
|
for s in STYLES
|
|
)
|
|
legend_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>>")
|
|
if use_coords:
|
|
lx, ly = _legend_pos(analysis)
|
|
add(f' legende [shape=plaintext, style="", fillcolor="none", '
|
|
f'label={legend_label}, pos="{lx * COORD_SCALE:.2f},'
|
|
f'{ly * COORD_SCALE:.2f}"];')
|
|
else:
|
|
add(" subgraph cluster_legende {")
|
|
add(' label="Legende"; style="rounded"; color="#a6a6a6";')
|
|
add(' fontname="Segoe UI"; fontsize=10; fontcolor="#404040";')
|
|
add(f' legende [shape=plaintext, style="", fillcolor="none", '
|
|
f'label={legend_label}];')
|
|
add(" }")
|
|
add("}")
|
|
|
|
return "\n".join(out) + "\n"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fehlerdatei
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def render_errors(analysis: Analysis, csv_file: Path) -> str:
|
|
out = [GENERATED_MARKER, f"# TRO-Fehler - {csv_file.name}", ""]
|
|
out.append(f"**Erzeugt:** {datetime.now().strftime('%Y-%m-%d %H:%M')} ")
|
|
out.append(f"**Quelle:** `{csv_file}` ")
|
|
out.append(f"**Typdefinition:** `lib/tro_catalog.py` (Stand {CATALOG_AS_OF})")
|
|
out.append("")
|
|
|
|
if analysis.isolated:
|
|
out.append("## Nicht verbundene TROs")
|
|
out.append("")
|
|
out.append(
|
|
"Jeder TRO muss mit mindestens einem anderen TRO verbunden sein. "
|
|
"Fuer die folgenden Uebergabestellen liess sich im Materialfluss kein "
|
|
"Nachbar-TRO finden - weder ueber die Spalte `Nachbarn` noch ueber einen "
|
|
"gemeinsamen Host."
|
|
)
|
|
out.append("")
|
|
out.append("| TRO | Typ | Host | Separatoren | Bezeichnung |")
|
|
out.append("|---|---|---|---|---|")
|
|
for tro in analysis.isolated:
|
|
out.append(
|
|
f"| `{tro.tro_id}` | `{tro.type_name}` | "
|
|
f"{', '.join(f'`{h}`' for h in tro.hosts)} | "
|
|
f"{', '.join(tro.separators) or '-'} | {tro.label} |"
|
|
)
|
|
out.append("")
|
|
out.append("**Moegliche Ursachen**")
|
|
out.append("")
|
|
out.append("- Spalte `Nachbarn` des Host-Objekts ist leer oder unvollstaendig")
|
|
out.append("- `Nachbarn` verweist auf eine unbekannte `TeileId`")
|
|
out.append("- Die Bahnangabe (`-L` / `-R`) eines Kreisels fehlt auf beiden Seiten")
|
|
out.append("")
|
|
|
|
if analysis.unknown_types:
|
|
out.append("## Typen ohne Katalogeintrag")
|
|
out.append("")
|
|
out.append(
|
|
"Die folgenden Typen wurden ermittelt, stehen aber nicht in "
|
|
"`TRO_CATALOG` in `lib/tro_catalog.py`:"
|
|
)
|
|
out.append("")
|
|
for name in analysis.unknown_types:
|
|
holders = [t.tro_id for t in analysis.tros if t.type_name == name]
|
|
out.append(f"- `{name}` ({', '.join(holders)})")
|
|
out.append("")
|
|
|
|
return "\n".join(out) + "\n"
|
|
|
|
|
|
def write_error_file(analysis: Analysis, csv_file: Path, path: Path) -> bool:
|
|
"""Fehlerdatei schreiben; veraltete eigene Datei bei fehlerfreiem Lauf entfernen."""
|
|
if analysis.isolated or analysis.unknown_types:
|
|
path.write_text(render_errors(analysis, csv_file), encoding="utf-8")
|
|
return True
|
|
|
|
if path.is_file():
|
|
try:
|
|
first = path.read_text(encoding="utf-8").splitlines()[:1]
|
|
except OSError:
|
|
first = []
|
|
if first and first[0].strip() == GENERATED_MARKER:
|
|
path.unlink()
|
|
return False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Dokumentation (Markdown)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _mermaid_text(text: str) -> str:
|
|
return text.replace('"', "'").replace("[", "(").replace("]", ")")
|
|
|
|
|
|
def _tro_mermaid(analysis: Analysis) -> list[str]:
|
|
out = ["```mermaid", "flowchart LR"]
|
|
out.extend(f" {line}" for line in mermaid_classdefs())
|
|
|
|
for tro in analysis.tros:
|
|
definition = tro.definition()
|
|
group = definition.get_group() if definition else "ext"
|
|
label = f"{tro.tro_id}<br/>{tro.type_name}<br/>{_mermaid_text(tro.label)}"
|
|
out.append(f' {tro.tro_id}["{label}"]:::{group}')
|
|
|
|
for 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],
|
|
engine: str = "dot",
|
|
) -> None:
|
|
types_used: dict[str, int] = {}
|
|
for tro in analysis.tros:
|
|
types_used[tro.type_name] = types_used.get(tro.type_name, 0) + 1
|
|
|
|
print("")
|
|
print("================================================================")
|
|
print("TRO-FLUSS")
|
|
print("================================================================")
|
|
print(f"Eingabe = {csv_file}")
|
|
print(f"TROs = {len(analysis.tros)}"
|
|
+ (f" ({', '.join(f'{k}: {v}' for k, v in sorted(types_used.items()))})"
|
|
if types_used else ""))
|
|
print(f"TRO-Verbindungen = {len(analysis.edges)}, "
|
|
f"nicht verbunden: {len(analysis.isolated)}")
|
|
print(f"Typdefinition = lib/tro_catalog.py (Stand {CATALOG_AS_OF}), "
|
|
f"{len(TRO_CATALOG)} Typen")
|
|
print(f"Layout = {engine}"
|
|
+ (" (lagerichtig, neato -n auf Anlagenkoordinaten, "
|
|
f"{COORD_SCALE} pt/mm)" if engine == "neato"
|
|
else " (berechnete Anordnung; lagerichtig mit --use-cords)"))
|
|
print(f"DOT = {dot_file}")
|
|
print(f"SVG = {svg_file if svg_file else '- (ohne --tosvg)'}")
|
|
print(f"Doku = {doc_file if doc_file else '- (ohne --doc)'}")
|
|
print(f"Fehlerdatei = {error_file if error_file else '- (keine Fehler)'}")
|
|
print("================================================================")
|
|
|
|
if analysis.isolated:
|
|
print("")
|
|
print(f"FEHLER: {len(analysis.isolated)} TRO(s) ohne Verbindung zu einem "
|
|
f"anderen TRO:")
|
|
for tro in analysis.isolated:
|
|
print(f" x {tro.tro_id} ({tro.type_name}) - {tro.label}")
|
|
|
|
if analysis.unknown_types:
|
|
print("")
|
|
print("FEHLER: TRO-Typen ohne Eintrag in TRO_CATALOG: "
|
|
+ ", ".join(analysis.unknown_types))
|
|
|
|
if analysis.findings:
|
|
print("")
|
|
print(f"Hinweise ({len(analysis.findings)}):")
|
|
for finding in analysis.findings:
|
|
print(f" ? {finding}")
|
|
|
|
if warnings:
|
|
print("")
|
|
print(f"Warnungen aus dem Materialfluss ({len(warnings)}):")
|
|
for warning in warnings:
|
|
print(f" ! {warning}")
|
|
print("")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(
|
|
prog="tro_flow",
|
|
description="Ermittelt aus einem CSV-Export der Anlagenobjekte die "
|
|
"benoetigten TROs und zeichnet ihren Fluss (DOT, optional SVG).",
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
epilog="Beispiele:\n"
|
|
" tro_flow.bat --file mubea.csv\n"
|
|
" tro_flow.bat --file mubea.csv --tosvg\n"
|
|
" tro_flow.bat --file mubea.csv --tosvg --doc\n"
|
|
" tro_flow.bat --file mubea.csv --tosvg --use-cords\n"
|
|
"\n"
|
|
"Exit-Codes:\n"
|
|
" 0 ok\n"
|
|
" 1 Eingabe- oder Aufruffehler\n"
|
|
" 2 SVG konnte nicht erzeugt werden (Graphviz)\n"
|
|
" 3 TRO-Fehler gefunden, Fehlerdatei geschrieben\n",
|
|
)
|
|
parser.add_argument(
|
|
"--file",
|
|
default="export.csv",
|
|
metavar="NAME",
|
|
help="Name der CSV-Eingabedatei in %%SKEL_DATA%% "
|
|
"(oder ein vollstaendiger Pfad). Standard: %(default)s",
|
|
)
|
|
parser.add_argument(
|
|
"--tosvg",
|
|
action="store_true",
|
|
help="Aus der erzeugten DOT-Datei zusaetzlich das TRO-Flussdiagramm als "
|
|
"SVG erzeugen (benoetigt Graphviz 'dot').",
|
|
)
|
|
parser.add_argument(
|
|
"--use-cords",
|
|
"--use-coords",
|
|
dest="use_cords",
|
|
action="store_true",
|
|
help="Lagerichtig zeichnen: jeder TRO wird auf seine Anlagenkoordinate "
|
|
"gesetzt und das SVG von Graphviz 'neato -n' erzeugt statt von "
|
|
f"'dot'. Massstab {COORD_SCALE} pt/mm. Nur mit --tosvg wirksam.",
|
|
)
|
|
parser.add_argument(
|
|
"--doc",
|
|
action="store_true",
|
|
help="Dokumentation der TROs als Markdown in %%SKEL_RESULTS%% erzeugen "
|
|
"(TRO-Liste, Verbindungen, Typdefinition).",
|
|
)
|
|
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"
|
|
|
|
# Ohne Koordinaten ist kein lagerichtiges Layout moeglich
|
|
without = missing_positions(analysis)
|
|
if args.use_cords and without:
|
|
print(f"FEHLER: --use-cords nicht moeglich, {len(without)} TRO(s) ohne "
|
|
f"Koordinate: {', '.join(without)}", file=sys.stderr)
|
|
print(" Ohne pos an jedem Knoten kann 'neato -n' nicht zeichnen.",
|
|
file=sys.stderr)
|
|
return 1
|
|
|
|
dot_file.write_text(
|
|
render_dot(analysis, csv_file.name, use_coords=args.use_cords),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
has_errors = write_error_file(analysis, csv_file, error_target)
|
|
error_file = error_target if has_errors else None
|
|
|
|
engine = "neato" if args.use_cords else "dot"
|
|
svg_error: str | None = None
|
|
if args.tosvg:
|
|
try:
|
|
graph_to_svg(dot_file, svg_file, engine=engine,
|
|
no_op=1 if args.use_cords else 0)
|
|
except RuntimeError as exc:
|
|
svg_error = str(exc)
|
|
created_svg = svg_file if args.tosvg and svg_error is None else None
|
|
|
|
doc_file = None
|
|
if args.doc:
|
|
doc_target.write_text(
|
|
render_doc(analysis, graph, csv_file, dot_file, created_svg, error_file),
|
|
encoding="utf-8",
|
|
)
|
|
doc_file = doc_target
|
|
|
|
report(analysis, csv_file, dot_file, created_svg, error_file, doc_file,
|
|
warnings, engine=engine)
|
|
|
|
if svg_error:
|
|
print(f"FEHLER: SVG nicht erzeugt: {svg_error}", file=sys.stderr)
|
|
return 2
|
|
if has_errors:
|
|
print(f"FEHLER: TRO-Pruefung fehlgeschlagen - siehe {error_target}",
|
|
file=sys.stderr)
|
|
return 3
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|