# -*- 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) ------------------------------- _tro_flow.dot immer (mit --use-cords lagerichtige pos-Attribute) _tro_flow.svg mit --tosvg (dot, mit --use-cords neato -n) _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 math 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, ROTATION_CCW, ROTATION_CW, 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, ) from tro_overrides import ( Overrides, apply_connections, apply_merges, apply_overrides, apply_splits, load_overrides, ) # --------------------------------------------------------------------------- # Konstanten # --------------------------------------------------------------------------- # Ab dieser Anzahl zusammenhaengender Schwerkraftlinien (Gefaellestrecke mit je # einem Separator) wird ein Speicherblock (PinStore_Auto) angenommen. MIN_STORE_LANES = 2 GENERATED_MARKER = "" # Zeichnungsmassstab fuer --use-cords: Graphviz rechnet in Punkt, die Anlage in # Millimeter. 0.25 pt/mm entspricht 1:4 - bei ~11 x 20 m Anlage ergibt das eine # Zeichnung von etwa 2800 x 5000 pt, in der sich die TRO-Kaesten nicht ueberdecken. COORD_SCALE = 0.25 # Abstand der Legende zum Anlagenrand in mm LEGEND_MARGIN = 3000 # --------------------------------------------------------------------------- # Datenmodell # --------------------------------------------------------------------------- @dataclass class Tro: """Eine ermittelte Uebergabestelle.""" tro_id: str type_name: str label: str hosts: list[str] separators: list[str] scanners: list[str] reason: str confidence: str fb_block: str = "" items: dict[str, int] = field(default_factory=dict) successors: list[str] = field(default_factory=list) predecessors: list[str] = field(default_factory=list) colocated: list[str] = field(default_factory=list) # Mittelpunkt der Bauteile, aus denen dieser TRO gebildet wurde position: tuple[float, float, float] | None = None position_from: str = "" position_points: int = 0 @property def x(self) -> float | None: return self.position[0] if self.position else None @property def y(self) -> float | None: return self.position[1] if self.position else None @property def z(self) -> float | None: return self.position[2] if self.position else None def describe_position(self) -> str: """Koordinate als Text, z. B. '1030, 4700'.""" if not self.position: return "-" return f"{self.position[0]:.0f}, {self.position[1]:.0f}" @property def connected(self) -> bool: return bool(self.successors or self.predecessors or self.colocated) def definition(self) -> TroDefinition | None: """Typdefinition aus tro_catalog (None wenn Typ unbekannt).""" return get_tro(self.type_name) @dataclass class Analysis: tros: list[Tro] = field(default_factory=list) edges: list[tuple[str, str]] = field(default_factory=list) # (von, nach) -> Kantenart (transfer / umlauf / weiche / unbestimmt / normal) edge_kinds: dict[tuple[str, str], str] = field(default_factory=dict) isolated: list[Tro] = field(default_factory=list) unknown_types: list[str] = field(default_factory=list) findings: list[str] = field(default_factory=list) # Manuelle Korrekturen aus cfg/tro_overrides.ini (Typkorrekturen, offene Punkte) overrides: Overrides = field(default_factory=Overrides) 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], overrides: Overrides | None = None, ) -> 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" # 4b) Manuell ergaenzte Uebergabestellen (cfg/tro_overrides.ini, [.split.]). # Vor der Nummernvergabe, damit die neuen TROs in derselben Reihenfolge # eine Nummer bekommen wie die abgeleiteten. if overrides is not None and overrides.splits: def _neuer_tro(spec, hosts, label): return Tro( tro_id="", type_name=spec.type_name, label=label, hosts=hosts, separators=list(spec.separators), scanners=list(spec.scanners), reason=spec.reason or "manuell ergaenzte Uebergabestelle", confidence=spec.confidence, ) apply_splits(tros, overrides, _neuer_tro) # 4c) Zusammenfassungen - nach den Splits, damit auch ein frisch # herausgeloester TRO Teil einer Zusammenfassung sein kann. if overrides is not None and overrides.merges: apply_merges(tros, overrides) # 5) Stabile, laufende Nummern in Reihenfolge der Host-Knoten tros.sort(key=lambda t: (t.hosts[0], t.separators[0] if t.separators else "")) width = max(2, len(str(len(tros)))) for index, tro in enumerate(tros, start=1): tro.tro_id = f"TRO{index:0{width}d}" # 5b) Manuelle Korrekturen aus cfg/tro_overrides.ini. Sie greifen erst hier, # weil sie ueber die TRO-ID adressiert sind, und noch vor Schritt 6, damit # ein manuell gesetzter Typ den SSCC-Hinweis nicht mehr ausloest. if overrides is not None: apply_overrides(tros, overrides) # 6) Scanner am Separator: 1Sep_SSCC laesst sich mechanisch nicht erkennen for tro in tros: if tro.type_name == "1Sep" and tro.scanners: findings.append( f"{tro.tro_id} ({tro.label}): Scanner {', '.join(tro.scanners)} am " f"Separator - als 1Sep eingestuft. Pruefen, ob 1Sep_SSCC " f"(SSCC-Scanner, Endmessung, WCS-Telegramm) erforderlich ist; das " f"ist aus dem mechanischen Layout nicht erkennbar." ) return tros def _tro_edge_kind(edge: Edge) -> str: """Materialkante -> Kantenart des TRO-Flusses (fuer Einfaerbung).""" if edge.note == "Weiche": return "weiche" if edge.kind == EDGE_CIRCLE: return "umlauf" if edge.kind == EDGE_UNRESOLVED: return "unbestimmt" if edge.note in ("Einschleusung", "Ausschleusung"): return "transfer" return "normal" def _bahn_richtung(node) -> tuple[float, float] | None: """Foerderrichtung einer Kreisel-Bahn als Einheitsvektor. Der Kreisel ist eine geschlossene Schleife: auf der einen Bahn laeuft das Material hin, auf der anderen zurueck. Welche Richtung das ist, ergibt sich aus der Laengsachse ("Drehung") und dem Umlaufsinn ("Drehrichtung"): UZS (im Uhrzeigersinn) Bahn R entgegen der Achse, Bahn L mit ihr GUZ (gegen den UZS) umgekehrt Rueckgabe None, wenn eines der beiden Merkmale fehlt - dann bleibt es bei der ungerichteten "gleiche Bahn"-Notiz. """ element = node.element drehung = _as_float(_merkmal(element.merkmale, "Drehung")) umlauf = element.rotation if drehung is None or umlauf not in (ROTATION_CW, ROTATION_CCW): return None rad = math.radians(drehung) achse = (math.cos(rad), math.sin(rad)) mit_achse = (node.lane == LANE_LEFT) == (umlauf == ROTATION_CW) return achse if mit_achse else (-achse[0], -achse[1]) def link_tros( graph: Graph, tros: list[Tro] ) -> tuple[list[tuple[str, str]], dict[tuple[str, str], str]]: """ TROs verknuepfen. Der Materialfluss laeuft von Uebergabestelle zu Uebergabestelle, ggf. durch passive Zwischenknoten (Strecken/Bahnen ohne eigenen TRO). Solche Knoten werden daher durchlaufen ("kontrahiert"), belegte Knoten beenden den Pfad. Jede TRO-Kante bekommt zusaetzlich die Art der *ankommenden* Materialkante (transfer / umlauf / weiche / unbestimmt / normal). """ owners: dict[str, list[str]] = {} for tro in tros: for host in tro.hosts: owners.setdefault(host, []).append(tro.tro_id) forward: dict[str, list[tuple[str, str]]] = {} for edge in graph.edges: kind = _tro_edge_kind(edge) forward.setdefault(edge.src, []).append((edge.dst, kind)) if edge.kind == EDGE_UNRESOLVED: # unbestimmte Richtung: in beide Richtungen begehbar forward.setdefault(edge.dst, []).append((edge.src, kind)) edges: set[tuple[str, str]] = set() edge_kinds: dict[tuple[str, str], str] = {} for tro in tros: seen = set(tro.hosts) queue = [(nxt, k) for host in tro.hosts for (nxt, k) in forward.get(host, [])] while queue: node, arriving = queue.pop() if node in seen: continue seen.add(node) if node in owners: for other in owners[node]: if other != tro.tro_id: pair = (tro.tro_id, other) edges.add(pair) edge_kinds.setdefault(pair, arriving) continue queue.extend(forward.get(node, [])) # Mehrere TROs auf derselben Kreisel-Bahn liegen HINTEREINANDER, nicht # nebeneinander: das Material laeuft die Bahn in Foerderrichtung ab. Sie # werden daher der Reihe nach verkettet statt nur als "gleiche Bahn" # markiert - sonst fehlt z. B. die Kante von der einen Sperre zur naechsten. verkettet: set[str] = set() for host, ids in owners.items(): if len(ids) < 2: continue node = graph.nodes.get(host) if node is None or node.kind != "Kreisel" or not node.lane: continue richtung = _bahn_richtung(node) if richtung is None: continue auf_bahn = [t for t in tros if t.tro_id in ids and t.position] if len(auf_bahn) < 2: continue # Projektion auf die Foerderrichtung: kleinster Wert kommt zuerst auf_bahn.sort(key=lambda t: t.position[0] * richtung[0] + t.position[1] * richtung[1]) for vorher, nachher in zip(auf_bahn, auf_bahn[1:]): pair = (vorher.tro_id, nachher.tro_id) edges.add(pair) edge_kinds.setdefault(pair, "bahn") verkettet.update(pair) # gleicher Host = physisch dieselbe Strecke/Bahn. Verkettete Bahnen sind # oben schon gerichtet verbunden und brauchen die ungerichtete Notiz nicht. for host, ids in owners.items(): for tro_id in ids: tro = next(t for t in tros if t.tro_id == tro_id) if tro_id in verkettet: continue tro.colocated = sorted( set(tro.colocated) | {o for o in ids if o != tro_id and o not in verkettet} ) for tro in tros: tro.successors = sorted(dst for src, dst in edges if src == tro.tro_id) tro.predecessors = sorted(src for src, dst in edges if dst == tro.tro_id) return sorted(edges), edge_kinds def _centroid( points: list[tuple[float, float, float]] ) -> tuple[float, float, float] | None: """Mittelpunkt mehrerer Positionen.""" if not points: return None count = len(points) return ( sum(p[0] for p in points) / count, sum(p[1] for p in points) / count, sum(p[2] for p in points) / count, ) def locate_tros(tros: list[Tro], elements: list[Element], graph: Graph) -> None: """ Jedem TRO eine Koordinate geben. Die Koordinate ist der **Mittelpunkt der Bauteile**, aus denen der TRO gebildet wurde - bei einem 1Sep1Swi also der Mittelwert aus Separator und Weiche, beim Linienspeicher der Mittelwert aller Linien-Separatoren. Hat ein TRO keine Bauteile mit Position (z. B. ein Vario ohne Separator), wird auf die Position seiner Host-Objekte zurueckgegriffen; das steht dann in `position_from`. """ by_id = {element.teile_id: element for element in elements} for tro in tros: # 1) Bauteile des TRO (Separatoren, Scanner) parts = [ by_id[i].position for i in list(tro.separators) + list(tro.scanners) if i in by_id and by_id[i].position ] source = "Bauteile" # 2) Ersatzweise die Host-Objekte if not parts: parts = [ graph.nodes[h].element.position for h in tro.hosts if h in graph.nodes and graph.nodes[h].element.position ] source = "Host-Objekte" tro.position = _centroid(parts) tro.position_points = len(parts) tro.position_from = source if tro.position else "keine Position" def analyse_tros( graph: Graph, elements: list[Element], overrides: Overrides | None = None ) -> Analysis: """TROs ermitteln, verknuepfen, verorten und gegen die Typdefinition pruefen.""" analysis = Analysis() analysis.overrides = overrides if overrides is not None else Overrides() analysis.tros = derive_tros(graph, elements, analysis.findings, analysis.overrides) # Verorten vor dem Verknuepfen: die Reihenfolge mehrerer TROs auf einer # Kreisel-Bahn wird ueber ihre Koordinate bestimmt. locate_tros(analysis.tros, elements, graph) analysis.edges, analysis.edge_kinds = link_tros(graph, analysis.tros) # Vor Ort aufgenommene Topologie schlaegt die abgeleitete. Die Differenz # wird protokolliert, damit sichtbar bleibt, was die Ableitung falsch hatte. if analysis.overrides.connections: abgeleitet = set(analysis.edges) analysis.edges, analysis.edge_kinds = apply_connections( analysis.tros, analysis.overrides ) gesetzt = set(analysis.edges) for src, dst in sorted(abgeleitet - gesetzt): analysis.findings.append( f"abgeleitete Verbindung {src} -> {dst} entfaellt - steht nicht " f"in der gesetzten Topologie" ) for src, dst in sorted(gesetzt - abgeleitet): analysis.findings.append( f"Verbindung {src} -> {dst} nur gesetzt, aus dem Layout nicht " f"ableitbar" ) for tro in analysis.tros: definition = tro.definition() if definition: tro.fb_block = definition.get_fb_block() tro.items = definition.get_items() else: analysis.unknown_types.append(tro.type_name) analysis.unknown_types = sorted(set(analysis.unknown_types)) analysis.isolated = [tro for tro in analysis.tros if not tro.connected] analysis.findings.sort() return analysis # --------------------------------------------------------------------------- # DOT erzeugen - Flussdiagramm der TROs # --------------------------------------------------------------------------- def _tro_label(tro: Tro) -> str: lines = [f"{tro.tro_id} ({tro.type_name})", tro.label] if tro.fb_block: lines.append(tro.fb_block) if tro.items: lines.append(", ".join(f"{c}x {i}" for i, c in sorted(tro.items.items()))) if tro.separators: seps = ", ".join(tro.separators) if len(tro.separators) > 6: seps = f"{tro.separators[0]} … {tro.separators[-1]} ({len(tro.separators)})" lines.append(f"Sep: {seps}") if tro.scanners: lines.append(f"Scan: {', '.join(tro.scanners)}") if tro.position: lines.append(f"x/y: {tro.describe_position()}") return _dot_label(lines) def missing_positions(analysis: Analysis) -> list[str]: """TROs 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) # Kantenstil je Art (an material_flow.py angelehnt: Umlauf hellblau gestrichelt, # Weiche orange). "transfer"/"normal" bleiben der durchgezogene Standardfluss. TRO_EDGE_STYLE = { "umlauf": 'color="#8ea9db", style=dashed, tooltip="Kreisel-Umlauf"', "weiche": 'color="#cc8800", penwidth=1.8, xlabel="Weiche", fontcolor="#cc8800", ' 'tooltip="Weiche (Kreisel-Uebergang)"', "unbestimmt": 'color="#bf8f00", style=dashed, dir=both, tooltip="Richtung unbestimmt"', "transfer": 'tooltip="Uebergabe (Ein-/Ausschleusung)"', } def _tro_edge_attrs(kind: str | None) -> str: return TRO_EDGE_STYLE.get(kind or "normal", "") def render_dot(analysis: Analysis, source: str, use_coords: bool = False) -> str: """ Gerichtetes Flussdiagramm der TROs als Graphviz-Quelle. use_coords=False Anordnung berechnet 'dot' (Fluss von links nach rechts) use_coords=True jeder Knoten wird auf seine Anlagenkoordinate gesetzt und mit "neato -n" lagerichtig gezeichnet """ out: list[str] = [] add = out.append add("// Automatisch erzeugt von lib/tro_flow.py - nicht manuell aendern.") add(f"// Quelle: {source}") if use_coords: add(f"// Lagerichtig: Knoten auf Anlagenkoordinate, {COORD_SCALE} pt/mm.") add("// Rendern mit: neato -n -Tsvg .dot -o .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}" [{_tro_edge_attrs(analysis.edge_kinds.get((src, dst)))}];') seen: set[frozenset] = set() for tro in analysis.tros: for other in tro.colocated: key = frozenset((tro.tro_id, other)) if key in seen: continue seen.add(key) add(f' "{tro.tro_id}" -> "{other}" [dir=none, style=dotted, ' f'color="#7f7f7f", constraint=false, tooltip="gleiche Bahn"];') add("") # Legende: eine Zeile je 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' {s.group} ' f'{s.label}' for s in STYLES ) legend_label = ('<' + rows + '' '' '' '' '' '' '' '' "
durchgezogenUebergabe / Ein-Ausschleusung
blau gestricheltKreisel-Umlauf
orangeWeiche (Kreisel-Uebergang)
grau gepunktetgleiche Bahn / Strecke
>") 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}
{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)} |") if analysis.overrides.applied: add(f"| Manuelle Typkorrekturen | {len(analysis.overrides.applied)} |") if analysis.overrides.open_points: add(f"| Offene Punkte (noch kein TRO) | " f"{len(analysis.overrides.open_points)} |") add("") add("| Typ | FB-Baustein | Anzahl hier | Bauteile je TRO | Farbgruppe |") 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. Manuelle Korrekturen und offene Punkte") add("") overrides = analysis.overrides if not overrides and not overrides.warnings: add("_Keine._ Alles unten Stehende ist rein aus dem Layout abgeleitet.") add("") else: add(f"Quelle: `{overrides.source}`" if overrides.source else "_Keine Override-Datei gefunden._") add("") if overrides.applied: add("### 6.1 Angewandte Typkorrekturen") add("") add("Diese Typen stammen **nicht** aus dem Layout, sondern aus der " "Begehung bzw. dem Review - im Layout sind sie nicht erkennbar.") add("") add("| Korrektur | Begruendung |") add("|---|---|") for entry in overrides.applied: head, _, tail = entry.partition(" (") add(f"| `{head}` | {tail[:-1] if tail.endswith(')') else tail} |") add("") if overrides.open_points: add("### 6.2 Offene Punkte - noch kein TRO") add("") add("Im Review gemeldete Stellen. Die TRO-Struktur ist hier " "**bewusst unveraendert**; die Punkte sind nur markiert (auch in der " "annotierten Zeichnung, Layer `TRO_OPENPOINT`), damit sie in BricsCAD " "geprueft werden koennen.") add("") add("| Punkt | X [mm] | Y [mm] | Z [mm] | erwartet | heute Teil von | Anmerkung |") add("|---|---:|---:|---:|---|---|---|") for point in overrides.open_points: z_text = f"{point.z:.0f}" if point.z is not None else "-" expect = f"`{point.expect}`" if point.expect else "-" owner = f"`{point.belongs_to}`" if point.belongs_to else "-" add(f"| {point.name} | {point.x:.0f} | {point.y:.0f} | {z_text} " f"| {expect} | {owner} | {point.note or '-'} |") add("") if overrides.warnings: add("### 6.3 Probleme mit der Override-Datei") add("") for warning in overrides.warnings: add(f"- {warning}") add("") add("## 7. Ableitungsregeln und Typdefinition") add("") add("Ein TRO ist die Steuerungslogik **einer Uebergabestelle**. Ausgangspunkt " "ist der Separator und das, was an ihm haengt (Host-Objekt, Scanner).") add("") add("| Host des Separators | TRO-Typ |") add("|---|---|") add(f"| Gefaellestrecke, >= {MIN_STORE_LANES} Linien mit gemeinsamem Zu-/Ablauf " "| `PinStore_Auto` (ein Block je Liniengruppe) |") add("| Gefaellestrecke, einzelne Linie | `1Sep` |") add("| Strecke (angetrieben) | `Vario` (je Streckensegment) |") add("| Kreisel-Bahn | `1Sep` |") add("| 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)) overrides = analysis.overrides if overrides.applied: print("") print(f"Manuelle Typkorrekturen ({len(overrides.applied)}) aus " f"{overrides.source}:") for entry in overrides.applied: print(f" * {entry}") if overrides.open_points: print("") print(f"Offene Punkte ({len(overrides.open_points)}) - nur markiert, " f"TRO-Struktur unveraendert:") for point in overrides.open_points: expect = f" erwartet {point.expect}" if point.expect else "" owner = f", heute Teil von {point.belongs_to}" if point.belongs_to else "" print(f" o {point.name} bei {point.describe_position()}{expect}{owner}") if overrides.warnings: print("") print(f"Probleme mit {overrides.source} ({len(overrides.warnings)}):") for warning in overrides.warnings: print(f" ! {warning}") if analysis.findings: print("") print(f"Hinweise ({len(analysis.findings)}):") for finding in analysis.findings: print(f" ? {finding}") if warnings: print("") print(f"Warnungen aus dem Materialfluss ({len(warnings)}):") for warning in warnings: print(f" ! {warning}") print("") # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser( prog="tro_flow", description="Ermittelt aus einem CSV-Export der Anlagenobjekte die " "benoetigten TROs und zeichnet ihren Fluss (DOT, optional SVG).", formatter_class=argparse.RawDescriptionHelpFormatter, epilog="Beispiele:\n" " tro_flow.bat --file mubea.csv\n" " tro_flow.bat --file mubea.csv --tosvg\n" " tro_flow.bat --file mubea.csv --tosvg --doc\n" " tro_flow.bat --file mubea.csv --tosvg --use-cords\n" "\n" "Exit-Codes:\n" " 0 ok\n" " 1 Eingabe- oder Aufruffehler\n" " 2 SVG konnte nicht erzeugt werden (Graphviz)\n" " 3 TRO-Fehler gefunden, Fehlerdatei geschrieben\n", ) parser.add_argument( "--file", default="export.csv", metavar="NAME", help="Name der CSV-Eingabedatei in %%SKEL_DATA%% " "(oder ein vollstaendiger Pfad). Standard: %(default)s", ) parser.add_argument( "--tosvg", action="store_true", help="Aus der erzeugten DOT-Datei zusaetzlich das TRO-Flussdiagramm als " "SVG erzeugen (benoetigt Graphviz 'dot').", ) parser.add_argument( "--use-cords", "--use-coords", dest="use_cords", action="store_true", help="Lagerichtig zeichnen: jeder TRO wird auf seine Anlagenkoordinate " "gesetzt und das SVG von Graphviz 'neato -n' erzeugt statt von " f"'dot'. Massstab {COORD_SCALE} pt/mm. Nur mit --tosvg wirksam.", ) parser.add_argument( "--doc", action="store_true", help="Dokumentation der TROs als Markdown in %%SKEL_RESULTS%% erzeugen " "(TRO-Liste, Verbindungen, Typdefinition).", ) 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 overrides = load_overrides(csv_file.name, env_dir("SKEL_CFG", "cfg")) analysis = analyse_tros(graph, elements, overrides) 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())