# -*- coding: utf-8 -*- # Absichtlich ohne Shebang: der Windows-py-Launcher wuerde daraus "python3" # ableiten und scheitert, wenn keine PythonCore-Installation registriert ist. # Der Interpreter wird von bin/material_flow.bat bzw. .sh gesetzt. """ material_flow.py - Materialfluss-Graph aus einem CSV-Export erzeugen. Liest einen CSV-Export der Anlagenobjekte (ILS 2.0) aus dem data-Ordner und erzeugt daraus einen gerichteten Graphen des Materialflusses. Knotenmodell ------------ * Gefaellestrecke / Strecke (Foerderer) -> je ein Knoten * Kreisel -> zwei Knoten (-L / -R); die Richtung zwischen ihnen kommt aus dem Merkmal "Drehrichtung": UZS = rechts nach links, GUZ = links nach rechts * Separator / Scanner -> kein eigener Knoten; sie werden ueber "Zuordnung" (bevorzugt) oder "TrackIds" des Containers (siehe unten) am Label des uebergeordneten Knotens angezeigt Container und ihre Anbauteile (Spalte "TrackIds") -------------------------------------------------- Seit Sept. 2026 listet ein Kreisel/Strecke/Gefaellestrecke in seiner eigenen Spalte "TrackIds" die TeileIds aller Anbauteile (Separator, Scanner, Weiche, Ein-/Ausschleuselement, BTMT Be-/Entladung), die zu ihm gehoeren. Fuer Separator/Scanner/Beladung/Entladung bleibt die praezisere "Zuordnung" des Anbauteils selbst massgeblich, solange sie sich aufloesen laesst; TrackIds schliesst nur die Luecken (fehlende/unaufloesbare Zuordnung, siehe _attach_devices/_attach_stations). Transferelemente (Weiche/Ein-/ Ausschleuselement) werden dagegen weiterhin ausschliesslich ueber ihre eigenen Merkmale ("Kreisel"/"Strecke"/"Kreisel A"/"Kreisel B") verdrahtet - dass sie in TrackIds zweier Container auftauchen (sie verbinden ja genau diese zwei), ist normal, kein Datenfehler. Kantenmodell ------------ Die Spalte "Nachbarn" liefert die Verbindungen (symmetrisch, ohne Richtung). Die Flussrichtung wird aus Hoehe und Antriebsrichtung abgeleitet: * Gefaellestrecke -> bergab: vom hoechsten Nachbarn weg * Strecke, Antriebfahrtrichtung -> "Auf": bergauf, zum hoechsten Nachbarn hin "Ab": bergab, vom hoechsten Nachbarn weg Verbindungen, deren Richtung sich so nicht bestimmen laesst (z. B. zwischen zwei Kreiseln auf gleicher Hoehe), werden als beidseitige Kante gezeichnet und im Bericht als Warnung ausgegeben. Abgrenzung ---------- Dieses Programm behandelt ausschliesslich den Materialfluss der *mechanischen Objekte*. Die Ableitung der Steuerungsobjekte (TROs) und ihr Flussdiagramm liegen im Schwesterprogramm lib/tro_flow.py, das die hier definierten Bausteine (read_elements, build_graph) weiterverwendet. Ausgabe (alles in SKEL_RESULTS) ------------------------------- _material_flow.dot immer _material_flow.svg mit --tosvg (Graphviz) _material_flow.md mit --doc Aufruf ueber bin/material_flow.bat bzw. bin/material_flow.sh. """ from __future__ import annotations import argparse import csv import json import os import re import shutil import subprocess import sys from dataclasses import dataclass, field, replace from datetime import datetime from pathlib import Path # --------------------------------------------------------------------------- # Konstanten # --------------------------------------------------------------------------- CSV_DELIMITER = ";" CSV_ENCODING = "utf-8-sig" TEILEART_PREFIX = "ILS 2.0 " # Dekodierung der CSV-Spalten K1-K4 (Anschluss-Koordinatensysteme eines Blocks): # 12-Zeichen-Base64-Strings mit je drei 24-Bit-Fixed-Point-Werten (x, y, z), # nur Position ohne Rotation. Kodierung siehe csv:trans-encode in # dxfmakros/Lisp/export.lsp (Faktor 10, gleiches Base64-Alphabet wie der # Insertpoint). Rueckgabe in mm im selben Koordinatensystem wie 'Position'. KX_B64_CHARS = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/" KX_B64_INDEX = {c: i for i, c in enumerate(KX_B64_CHARS)} KX_TRANS_FAKTOR = 10.0 # Objektarten, die je einen Materialfluss-Knoten bilden TRANSPORT_KINDS = ("Gefaellestrecke", "Strecke") # Objektarten, die als Kreis aus zwei Bahnen (links/rechts) modelliert werden CIRCLE_KINDS = ("Kreisel",) # Objektarten, die per "Zuordnung" an einem Knoten haengen (kein eigener Knoten) DEVICE_KINDS = ("Separator", "Scanner") # Ein-/Ausgabestationen = Materialquelle/-senke, d.h. die Ein-/Ausgangspunkte # des Materialflusses. Es gibt sie in zwei Auspraegungen: # * konkrete BTMT-Station: "Beladung" (Quelle) / "Entladung" (Senke) # * generisches Symbol: "Eingang" (Quelle) / "Ausgang" (Senke) - der # allgemeinere Bezeichner fuer dasselbe. Liegt ein Symbol nahezu deckungs- # gleich ueber der konkreten BTMT-Station (haeufig, gleiche x/y), wird nur # ein Knoten gezeichnet: die BTMT-Station gewinnt, das Symbol wird # verschluckt (siehe _merge_station_symbols / STATION_MERGE_MM). ENTRY_KINDS = ("Beladung", "Eingang") # -> Materialeingang (Quelle) EXIT_KINDS = ("Entladung", "Ausgang") # -> Materialausgang (Senke) STATION_KINDS = ENTRY_KINDS + EXIT_KINDS # Die generischen Symbole (verschluckbar zugunsten der konkreten BTMT-Station). STATION_SYMBOL_KINDS = ("Eingang", "Ausgang") # Schluesselwoerter zur Erkennung der Stationen (Kleinschreibung, Teilstring). # "eingang"/"ausgang" MUESSEN nach den spezifischeren Mustern kommen, damit z.B. # "BTMT-Beladung" als Beladung (nicht Eingang) klassifiziert wird. STATION_PATTERNS = ( ("Beladung", ("beladung", "beladestation", "btmt-beladung", "btmt beladung")), ("Entladung", ("entladung", "entladestation", "sc_entladung", "abwurf")), ("Eingang", ("eingang",)), ("Ausgang", ("ausgang",)), ) # Max. Abstand (mm), ab dem ein generisches Eingang/Ausgang-Symbol als # "deckungsgleich" mit einer BTMT-Station gilt und verschluckt wird. Die # beobachteten Paare liegen bei 256-431 mm, echte getrennte Stationen >1000 mm. STATION_MERGE_MM = 600.0 # Transfer-/Weichenelemente aus dem erweiterten Export. Sie bilden KEINE eigenen # Knoten, liefern aber die *gerichtete* Verbindung zwischen den Transportobjekten: # Ausschleuselement = Kreisel -> Strecke (am Streckenanfang, Material verlaesst Kreisel) # Einschleuselement = Strecke -> Kreisel (am Streckenende, Material laeuft in Kreisel) # Weiche = Uebergang zwischen zwei Kreiseln # Merkmale: {"Kreisel","Strecke"/"Foerderer","Streckenende"} bzw. {"Kreisel A","Kreisel B"}. # "Strecke" und "Foerderer" sind zwei Schreibweisen desselben Merkmals, die # nebeneinander in ein und demselben Export vorkommen (nicht projektabhaengig # aufteilbar) - siehe _apply_transfers. TRANSFER_KINDS = ("Einschleus", "Ausschleus", "Weiche") TRANSFER_PATTERNS = ( ("Ausschleus", ("ausschleus",)), ("Einschleus", ("einschleus",)), ("Weiche", ("weiche",)), ) # Objektarten des feinen Bahnzusammenhang-Graphen (--track, siehe doc/track.md): # Separator plus die drei Transferarten bilden dort je einen eigenen Knoten - # anders als im normalen Materialfluss-Graphen, wo sie nur Kanten/Zuordnungen # liefern (siehe TRANSPORT_KINDS/CIRCLE_KINDS oben). TRACK_KINDS = ("Separator",) + TRANSFER_KINDS # Container, die in der CSV-Spalte "TrackIds" eine geordnete Liste ihrer # enthaltenen Bahnelemente fuehren (siehe Element.track_ids): Kreisel bringt den # zirkulaeren Umlauf in Drehrichtung mit, Strecke/Gefaellestrecke die lineare # Kette AS -> SEP.. -> ES in Baureihenfolge. Aus diesen Listen wird der # --track-Graph direkt aufgebaut (build_track_graph). TRACK_CONTAINER_KINDS = CIRCLE_KINDS + TRANSPORT_KINDS LANE_LEFT = "L" LANE_RIGHT = "R" LANE_NAMES = {LANE_LEFT: "links", LANE_RIGHT: "rechts"} UNASSIGNED = "nicht zugeordnet" # Drehrichtung eines Kreisels (Merkmal "Drehrichtung"). Sie bestimmt, in welche # Richtung das Material innerhalb des Kreisels zwischen den beiden Bahnen laeuft: # UZS im Uhrzeigersinn -> von rechts nach links (R -> L) # GUZ gegen den Uhrzeigersinn -> von links nach rechts (L -> R) ROTATION_CW = "UZS" ROTATION_CCW = "GUZ" ROTATION_FLOW = {ROTATION_CW: "rechts nach links", ROTATION_CCW: "links nach rechts"} # "0009-L" -> base 0009, lane L ; "0034" -> base 0034, lane None NEIGHBOUR_RE = re.compile(r"^(?P[^\s-]+)(?:-(?P[LR]))?$") # Kantenarten EDGE_FLOW = "flow" EDGE_CIRCLE = "circle" EDGE_UNRESOLVED = "unresolved" # Darstellung je Objektart NODE_STYLES = { "Kreisel": dict(fillcolor="#2f5597", fontcolor="#ffffff", color="#1f3864"), "Gefaellestrecke": dict(fillcolor="#fbe5d6", fontcolor="#833c00", color="#c55a11"), "Strecke": dict(fillcolor="#e2f0d9", fontcolor="#375623", color="#548235"), "Beladung": dict(fillcolor="#d4f0d4", fontcolor="#1e5b1e", color="#338833"), "Entladung": dict(fillcolor="#f4d4d4", fontcolor="#7a1f1f", color="#cc3333"), # Generische Symbole - gleiche Farbe wie die konkreten BTMT-Stationen # (gruen = Eingang/Quelle, rot = Ausgang/Senke). "Eingang": dict(fillcolor="#d4f0d4", fontcolor="#1e5b1e", color="#338833"), "Ausgang": dict(fillcolor="#f4d4d4", fontcolor="#7a1f1f", color="#cc3333"), } DEFAULT_NODE_STYLE = dict(fillcolor="#f2f2f2", fontcolor="#404040", color="#808080") # Ab dieser Anzahl paralleler Gefaellestrecken zwischen denselben zwei # Kreisel-Bahnen (gleicher Zulauf, gleicher Ablauf, gleiche Steigung) werden # sie im Materialfluss-Diagramm zu einem Block zusammengefasst - sonst wird # eine grosse Linienbank (Speicher) unlesbar breit. MIN_GEFAELLE_GROUP = 2 # --------------------------------------------------------------------------- # Datenmodell # --------------------------------------------------------------------------- @dataclass class Element: """Eine Zeile des CSV-Exports.""" number: str kind: str teile_art: str teile_id: str name: str plan_square: str neighbours: list[tuple[str, str | None]] # Geordnete Element-Liste eines Containers (Kreisel/VF/GF) aus der CSV-Spalte # "TrackIds": bei VF/GF die Kette AS -> SEP.. -> ES in Baureihenfolge, bei # Kreisel der zirkulaere Umlauf aller auf ihm liegenden Punkte in # Drehrichtung (Separator/AS/ES/Weiche/BTMT/Eingang/Ausgang). Leer fuer # Nicht-Container. Erzeugt von dxfmakros/export_csv.py, siehe doc/track.md. track_ids: list[str] merkmale: dict row: int position: tuple[float, float, float] | None = None bbox: tuple[float, float, float] | None = None # Anschluss-Koordinatensysteme K1..K4 aus den gleichnamigen CSV-Spalten, # als (label, (x, y)) in mm; leere Spalten fehlen. Nur fuer --show-kx. kpoints: list[tuple[str, tuple[float, float]]] = field(default_factory=list) @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 @property def bbox_x(self) -> float | None: """Boundingbox-Ausdehnung in X (mm), aus der Spalte 'Boundingbox'.""" return self.bbox[0] if self.bbox else None @property def bbox_y(self) -> float | None: """Boundingbox-Ausdehnung in Y (mm).""" return self.bbox[1] if self.bbox else None @property def is_transport(self) -> bool: return self.kind in TRANSPORT_KINDS @property def is_circle(self) -> bool: return self.kind in CIRCLE_KINDS @property def is_device(self) -> bool: return self.kind in DEVICE_KINDS @property def is_entry(self) -> bool: """BTMT-Beladung: Materialeingang (Quelle).""" return self.kind in ENTRY_KINDS @property def is_exit(self) -> bool: """SC_Entladung: Materialausgang (Senke).""" return self.kind in EXIT_KINDS @property def is_station(self) -> bool: return self.kind in STATION_KINDS @property def is_transfer(self) -> bool: """Ein-/Ausschleuselement oder Weiche (liefert Richtung, kein Knoten).""" return self.kind in TRANSFER_KINDS @property def height(self) -> float | None: """Montage-/Einbauhoehe in Meter, falls ableitbar.""" if self.is_circle: return _as_float(_merkmal(self.merkmale, "Höhe in m", "Hoehe in m")) return _as_float(_merkmal(self.merkmale, "Montagehoehe_m", "Montagehöhe_m")) @property def drive_dir(self) -> str: return str(_merkmal(self.merkmale, "Antriebfahrtrichtung") or "").strip() @property def rotation(self) -> str: """Drehrichtung eines Kreisels: UZS (im Uhrzeigersinn) oder GUZ.""" return str(_merkmal(self.merkmale, "Drehrichtung") or "").strip().upper() @property def height_from(self) -> float | None: """Hoehe am Anfang des Objekts in mm (Merkmal Hoehe_Von_mm).""" return _as_float(_merkmal(self.merkmale, "Hoehe_Von_mm", "Höhe_Von_mm")) @property def height_to(self) -> float | None: """Hoehe am Ende des Objekts in mm (Merkmal Hoehe_Bis_mm).""" return _as_float(_merkmal(self.merkmale, "Hoehe_Bis_mm", "Höhe_Bis_mm")) @property def rise(self) -> float | None: """ Hoehenaenderung in Foerderrichtung in mm. Positiv = das Objekt foerdert nach oben, negativ = nach unten. Die Werte koennen absolut sein (Gefaellestrecke: 1941 -> 1466) oder relativ zum Objekt (Strecke: 0 -> 474); nur das Vorzeichen wird ausgewertet. """ start, end = self.height_from, self.height_to if start is None or end is None: return None return end - start @property def assignment(self) -> str: return str(_merkmal(self.merkmale, "Zuordnung") or "").strip() def describe(self) -> str: return f"Zeile {self.row} ({self.kind} {self.teile_id} '{self.name}')" @dataclass class Node: """Ein Knoten des Materialfluss-Graphen.""" node_id: str kind: str element: Element lane: str | None = None devices: dict[str, list[str]] = field(default_factory=dict) @property def height(self) -> float | None: return self.element.height @dataclass class Edge: src: str dst: str kind: str note: str = "" @dataclass class Graph: nodes: dict[str, Node] = field(default_factory=dict) edges: list[Edge] = field(default_factory=list) warnings: list[str] = field(default_factory=list) unassigned: list[Element] = field(default_factory=list) source: str = "" @dataclass class TrackNode: """Ein Knoten des feinen Bahnzusammenhang-Graphen (--track, doc/track.md). element ist None fuer die beiden Hilfsknotenarten: einen Bahn-Platzhalter (placeholder=True - eine Kreisel-Bahn oder Strecke, an der sich kein konkretes Bahnelement auflösen liess, siehe build_track_graph) oder ein generisches Eingang/Ausgang-Symbol (SYMBOL_ENTRY/SYMBOL_EXIT) an einem offenen Bahnende ohne BTMT-Station. """ node_id: str kind: str element: Element | None = None placeholder: bool = False note: str = "" # Zuordnung des naechstgelegenen Scanners (siehe build_track_graph, Schritt 5) # - Scanner bilden weiterhin keinen eigenen Knoten (TRACK_KINDS), aber # render_track_dot() markiert den betroffenen Separator damit per rotem # Punkt statt einen zusaetzlichen Scanner-Knoten einzufuehren. scanner_zuordnung: str | None = None @dataclass class TrackEdge: src: str dst: str note: str = "" @dataclass class TrackGraph: nodes: dict[str, TrackNode] = field(default_factory=dict) edges: list[TrackEdge] = field(default_factory=list) source: str = "" # --------------------------------------------------------------------------- # Hilfsfunktionen # --------------------------------------------------------------------------- def _merkmal(merkmale: dict, *keys: str): """Merkmal ueber mehrere Schreibweisen hinweg lesen.""" for key in keys: if key in merkmale: return merkmale[key] return None def _as_float(value) -> float | None: """Zahl tolerant lesen ('1.94' und '1,94'). Listen wie '2.298,1' -> None.""" if value is None: return None text = str(value).strip() if not text: return None if text.count(",") == 1 and "." not in text: text = text.replace(",", ".") try: return float(text) except ValueError: return None def _dot_escape(text: str) -> str: return text.replace("\\", "\\\\").replace('"', '\\"') def _dot_label(lines: list[str]) -> str: return "\\n".join(_dot_escape(line) for line in lines if line) # --------------------------------------------------------------------------- # Pfade / Umgebung # --------------------------------------------------------------------------- def project_root() -> Path: env = os.environ.get("SPS_SKEL") if env: return Path(env) return Path(__file__).resolve().parent.parent def env_dir(var: str, fallback: str) -> Path: env = os.environ.get(var) return Path(env) if env else project_root() / fallback def resolve_input(name: str) -> Path: """Dateinamen gegen SKEL_DATA aufloesen; Pfadangaben bleiben unveraendert.""" candidate = Path(name) if candidate.is_absolute() or candidate.parent != Path("."): return candidate return env_dir("SKEL_DATA", "data") / name # --------------------------------------------------------------------------- # CSV einlesen # --------------------------------------------------------------------------- def decode_kpoint(raw: str) -> tuple[float, float] | None: """ K1-K4-Base64-String -> (x, y) in mm, oder None bei leer/ungueltig. Die Spalten K1..K4 enthalten je 12 Zeichen (drei 24-Bit-Werte x, y, z); fuer die Draufsicht interessiert nur die xy-Lage. Kodierung siehe csv:trans-encode in dxfmakros/Lisp/export.lsp (vgl. KX_* oben). """ text = (raw or "").strip() if len(text) < 8: # mindestens x und y (2 * 4 Zeichen) return None values: list[int] = [] for start in range(0, len(text) - 3, 4): word = 0 for char in text[start:start + 4]: if char not in KX_B64_INDEX: return None word = (word << 6) | KX_B64_INDEX[char] if word >= 8388608: # Zweierkomplement (24 Bit) word -= 16777216 values.append(word) if len(values) < 2: return None return (values[0] / KX_TRANS_FAKTOR, values[1] / KX_TRANS_FAKTOR) def parse_position(raw: str) -> tuple[float, float, float] | None: """'9510.00, -4882.95, 1942.50' -> (9510.0, -4882.95, 1942.5)""" parts = [p.strip() for p in (raw or "").split(",")] if len(parts) != 3: return None values = [_as_float(p) for p in parts] if any(v is None for v in values): return None return (values[0], values[1], values[2]) def parse_neighbours(raw: str) -> list[tuple[str, str | None]]: """'0009-L, 0002-R' -> [('0009', 'L'), ('0002', 'R')]""" result: list[tuple[str, str | None]] = [] for token in (raw or "").split(","): token = token.strip() if not token: continue match = NEIGHBOUR_RE.match(token) if not match: continue result.append((match.group("base"), match.group("lane"))) return result # def parse_track_ids(raw: str) -> list[str]: # """'0118, 0059, 0119' -> ['0118', '0059', '0119'] (Reihenfolge erhalten).""" # return [token.strip() for token in (raw or "").split(",") if token.strip()] def parse_track_ids(raw: str, *, context: str = "", warnings: list[str] | None = None) -> list[str]: """'0040, 0076, 0036' -> ['0040', '0076', '0036'] Spalte 'TrackIds' (siehe Element.track_ids): anders als 'Nachbarn' ohne Bahn-Suffix (-L/-R) - TrackIds nennt nur die TeileId, nie eine Kreiselbahn. Der ILS-2.0-Export nennt eine TeileId in dieser Spalte gelegentlich doppelt (Datenfehler im CAD-Export, kein sps_skel-Artefakt) - ohne Entduplizierung wuerde das in _build_track_owners wie zwei verschiedene Container aussehen und faelschlich eine "mehrere Container"-Mehrdeutigkeit ausloesen. """ seen: set[str] = set() result: list[str] = [] for token in (raw or "").split(","): token = token.strip() if not token: continue if token in seen: if warnings is not None: warnings.append(f"{context}: TrackIds nennt '{token}' mehrfach - Duplikat ignoriert") continue seen.add(token) result.append(token) return result def _match_kind(patterns, *texts: str) -> str | None: """Erste Musterklasse, deren Schluesselwort in den Texten vorkommt.""" haystack = " ".join(t.lower() for t in texts if t) for kind, keys in patterns: if any(key in haystack for key in keys): return kind return None def classify_station(*texts: str) -> str | None: """BTMT Be-/Entladestation an TeileArt/Bezeichnung erkennen (sonst None).""" return _match_kind(STATION_PATTERNS, *texts) def classify_transfer(*texts: str) -> str | None: """Ein-/Ausschleuselement bzw. Weiche erkennen (sonst None).""" return _match_kind(TRANSFER_PATTERNS, *texts) def read_elements(path: Path, warnings: list[str]) -> list[Element]: """CSV-Export einlesen.""" with path.open("r", encoding=CSV_ENCODING, newline="") as handle: reader = csv.DictReader(handle, delimiter=CSV_DELIMITER) if not reader.fieldnames or "TeileId" not in reader.fieldnames: raise ValueError( f"Unerwartete Kopfzeile in {path.name}: 'TeileId' nicht gefunden. " f"Gefunden: {reader.fieldnames}" ) elements: list[Element] = [] seen: dict[str, Element] = {} for offset, row in enumerate(reader, start=2): teile_id = (row.get("TeileId") or "").strip() if not teile_id: continue teile_art = (row.get("TeileArt") or "").strip() kind = teile_art[len(TEILEART_PREFIX):] if teile_art.startswith(TEILEART_PREFIX) else teile_art # BTMT Be-/Entladestation als Ein-/Ausgang erkennen (sps_skel-seitig) station_kind = classify_station(teile_art, row.get("Bezeichnung") or "") if station_kind: kind = station_kind # Ein-/Ausschleuselement bzw. Weiche erkennen (liefern Richtung) transfer_kind = classify_transfer(teile_art, row.get("Bezeichnung") or "") if transfer_kind: kind = transfer_kind raw_merkmale = (row.get("Merkmale") or "").strip() merkmale: dict = {} if raw_merkmale: try: parsed = json.loads(raw_merkmale) if isinstance(parsed, dict): merkmale = parsed else: warnings.append(f"Zeile {offset}: Merkmale ist kein Objekt - ignoriert") except json.JSONDecodeError as exc: warnings.append(f"Zeile {offset}: Merkmale nicht lesbar ({exc.msg}) - ignoriert") element = Element( number=(row.get("Elementnummer") or "").strip(), kind=kind, teile_art=teile_art, teile_id=teile_id, name=(row.get("Bezeichnung") or "").strip(), plan_square=(row.get("Planquadrat") or "").strip(), neighbours=parse_neighbours(row.get("Nachbarn") or ""), track_ids=parse_track_ids( row.get("TrackIds") or "", context=f"Zeile {offset} ({teile_art} {teile_id})", warnings=warnings ), merkmale=merkmale, row=offset, position=parse_position(row.get("Position") or ""), bbox=parse_position(row.get("Boundingbox") or ""), kpoints=[ (label, xy) for label in ("K1", "K2", "K3", "K4") if (xy := decode_kpoint(row.get(label) or "")) is not None ], ) if teile_id in seen: warnings.append( f"{element.describe()}: TeileId doppelt (bereits in Zeile {seen[teile_id].row}) - ignoriert" ) continue seen[teile_id] = element elements.append(element) return elements # --------------------------------------------------------------------------- # Graph aufbauen # --------------------------------------------------------------------------- def _build_nodes(elements: list[Element], warnings: list[str]) -> dict[str, Node]: nodes: dict[str, Node] = {} for element in elements: if element.is_transport: nodes[element.teile_id] = Node(element.teile_id, element.kind, element) elif element.is_circle: for lane in (LANE_LEFT, LANE_RIGHT): node_id = f"{element.teile_id}-{lane}" nodes[node_id] = Node(node_id, element.kind, element, lane=lane) elif element.is_station: nodes[element.teile_id] = Node(element.teile_id, element.kind, element) elif element.is_transfer: # Ein-/Ausschleuselement / Weiche liefern nur die (gerichtete) # Verbindung, siehe _apply_transfers - kein eigener Knoten. continue elif not element.is_device: warnings.append( f"{element.describe()}: unbekannte TeileArt '{element.teile_art}' - kein Knoten erzeugt" ) return nodes def _build_track_owners(elements: list[Element], warnings: list[str]) -> dict[str, list[str]]: """ Rueckindex der Spalte 'TrackIds': welche(r) Container (Kreisel/Strecke/ Gefaellestrecke) nennt ein Anbauteil (Separator/Scanner/Weiche/Ein-/ Ausschleuselement/Beladung/Entladung) in seiner eigenen 'TrackIds'-Liste. Transferelemente (Weiche/Einschleus/Ausschleus, siehe TRANSFER_KINDS) verbinden zwei Container und werden folgerichtig von beiden genannt - mehrere Nennungen sind dort normal (ihre Verbindung wird ohnehin ueber die eigenen Merkmale "Kreisel"/"Strecke"/"Kreisel A"/"Kreisel B" bestimmt, nicht hierueber). Fuer alle anderen Anbauteile (genau ein Besitzer erwartet) macht das keinen Sinn - siehe _track_owner. """ by_id = {element.teile_id: element for element in elements} owners: dict[str, list[str]] = {} for element in elements: if not (element.is_transport or element.is_circle): continue for fine_id in element.track_ids: if fine_id == element.teile_id: warnings.append(f"{element.describe()}: TrackIds verweist auf sich selbst - ignoriert") continue if fine_id not in by_id: warnings.append( f"{element.describe()}: TrackIds verweist auf unbekanntes Element '{fine_id}' - ignoriert" ) continue owners.setdefault(fine_id, []).append(element.teile_id) return owners def _track_owner( element: Element, owners_of: dict[str, list[str]], by_id: dict[str, Element], warnings: list[str] ) -> Element | None: """ Container-Element zu einem Anbauteil ueber 'TrackIds' (Rueckindex), oder None, wenn TrackIds dazu nichts sagt oder (fuer ein Anbauteil mit genau einem erwarteten Besitzer) mehrdeutig ist. Der Aufrufer faellt dann auf die bisherige Zuordnung/Nachbarn-Ableitung zurueck. """ owners = owners_of.get(element.teile_id) or [] if not owners: return None if len(owners) > 1: warnings.append( f"{element.describe()}: TrackIds nennt mehrere Container ({', '.join(sorted(owners))}) - " f"mehrdeutig, bisherige Zuordnung/Nachbarn-Ableitung verwendet" ) return None return by_id.get(owners[0]) def _owner_node_ids(owner: Element, element: Element, nodes: dict[str, Node]) -> list[str]: """ Knoten-Id(s) eines per TrackIds gefundenen Container-Elements - nur wenn die eigene 'Zuordnung' des Anbauteils schon gescheitert ist (siehe _attach_devices: leer, oder ohne passenden Knoten). TrackIds kennt nur den Kreisel als Ganzes, nicht die Bahn (-L/-R); da eine aufloesbare Zuordnung bereits Vorrang hatte, bleibt hier nur der Fall "Bahn unbekannt" - wie mit E-Planung abgestimmt, wird dann an beide Bahnen angehaengt. """ if owner.is_circle: return [node_id for lane in (LANE_LEFT, LANE_RIGHT) if (node_id := f"{owner.teile_id}-{lane}") in nodes] return [owner.teile_id] if owner.teile_id in nodes else [] def _attach_devices( elements: list[Element], nodes: dict[str, Node], by_id: dict[str, Element], owners_of: dict[str, list[str]], warnings: list[str] ) -> list[Element]: """ Separatoren/Scanner an ihren Knoten haengen. Die eigene 'Zuordnung' des Anbauteils bleibt massgeblich, solange sie auf einen existierenden Knoten zeigt - sie ist praeziser als 'TrackIds', das bei einem Anbauteil direkt an einer Kreisel-Strecke-Grenze (Ausschleus-/ Einschleuselement) auch den benachbarten Kreisel nennt, obwohl das Anbauteil logisch zur Strecke gehoert (siehe _build_track_owners). Erst wenn die Zuordnung fehlt oder ins Leere zeigt (aeltere Exports ohne TrackIds-Spalte eingeschlossen), uebernimmt 'TrackIds' auf dem Container (siehe _track_owner) - das deckt u. a. Anbauteile ab, deren Zuordnung nur den Kreisel ohne Bahn (-L/-R) nennt. """ unassigned: list[Element] = [] for element in elements: if not element.is_device: continue target = element.assignment if target and target != UNASSIGNED and target in nodes: nodes[target].devices.setdefault(element.kind, []).append(element.teile_id) continue owner = _track_owner(element, owners_of, by_id, warnings) if owner is not None: targets = _owner_node_ids(owner, element, nodes) if targets: for node_id in targets: nodes[node_id].devices.setdefault(element.kind, []).append(element.teile_id) continue warnings.append( f"{element.describe()}: TrackIds verweist auf Container {owner.teile_id}, " f"aber kein passender Knoten - nicht dargestellt" ) if not target or target == UNASSIGNED: unassigned.append(element) continue warnings.append( f"{element.describe()}: Zuordnung '{target}' zeigt auf keinen Knoten - nicht dargestellt" ) unassigned.append(element) return unassigned def _collect_links( elements: list[Element], by_id: dict[str, Element], warnings: list[str] ) -> dict[frozenset, dict[str, set[str]]]: """ Verbindungen aus der Spalte 'Nachbarn' sammeln. Ergebnis: {frozenset({idA, idB}): {idA: {Bahnen}, idB: {Bahnen}}} Die Bahn (-L/-R) eines Kreisels steht immer in der Zeile des *Partners*, daher werden beide Blickrichtungen zusammengefuehrt. """ observed: dict[frozenset, dict[str, set[str]]] = {} for element in elements: if not element.neighbours: continue if element.is_device: warnings.append( f"{element.describe()}: Nachbarn an einem Anbauteil werden nicht ausgewertet" ) continue if not (element.is_transport or element.is_circle): continue for base, lane in element.neighbours: if base == element.teile_id: warnings.append(f"{element.describe()}: Nachbar verweist auf sich selbst - ignoriert") continue partner = by_id.get(base) if partner is None: warnings.append(f"{element.describe()}: unbekannter Nachbar '{base}' - ignoriert") continue if partner.is_device or not (partner.is_transport or partner.is_circle): warnings.append( f"{element.describe()}: Nachbar '{base}' ({partner.kind}) ist kein Flussobjekt - ignoriert" ) continue key = frozenset((element.teile_id, base)) entry = observed.setdefault(key, {element.teile_id: set(), base: set()}) if lane: entry[base].add(lane) return observed def _endpoints( base: str, lanes: set[str], by_id: dict[str, Element], warnings: list[str], partner: str ) -> list[str]: """Basis-Id auf konkrete Knoten aufloesen (Kreisel: eine Bahn je Knoten).""" element = by_id[base] if not element.is_circle: return [base] if lanes: return [f"{base}-{lane}" for lane in sorted(lanes)] warnings.append( f"Verbindung {base} <-> {partner}: keine Bahn (-L/-R) angegeben - " f"beide Bahnen des Kreisels {base} werden verbunden" ) return [f"{base}-{LANE_LEFT}", f"{base}-{LANE_RIGHT}"] def drive_uphill(element: Element) -> tuple[bool | None, str]: """ Foerdert das Objekt nach oben? True = Fluss laeuft zum hoechsten Nachbarn hin (bergauf) False = Fluss laeuft vom hoechsten Nachbarn weg (bergab) None = nicht bestimmbar Die Hoehe bestimmt die Richtung NUR bei einer reinen Gefaellestrecke - dort laeuft der Fluss per Definition bergab (Schwerkraft). Ein Foerderer (VF) kann bauartbedingt nach oben ODER unten foerdern; eine Strecke ist oft eine gemischte GF/VF-Kette. Fuer beide ist die Hoehendifferenz KEIN Richtungs- indikator - ihre Richtung kommt aus den Ein-/Ausschleuselementen (_apply_transfers, AS -> ES) bzw. einer eindeutigen Antriebfahrtrichtung, sonst bleibt sie unbestimmt. Bewertet wird daher: 1. Gefaellestrecke: Hoehe_Von/Bis (bergab), sonst per Definition bergab 2. Antriebfahrtrichtung "Auf"/"Ab" (nur wenn eindeutig, nicht bei gemischten Listen wie "horizontal,horizontal,Auf") 3. sonst None (unbestimmt) """ if element.kind == "Gefaellestrecke": rise = element.rise if rise is not None and rise != 0: direction = "Auf" if rise > 0 else "Ab" return rise > 0, ( f"Gefaellestrecke Hoehe_Von {element.height_from:.0f} mm -> Hoehe_Bis " f"{element.height_to:.0f} mm ({direction}, {rise:+.0f} mm)" ) return False, "Gefaellestrecke ohne Hoehenangabe - laeuft bergab" # Foerderer/Strecke: Hoehe sagt NICHTS ueber die Richtung. Nur eine # eindeutige Antriebfahrtrichtung (genau "Auf" bzw. "Ab", keine gemischte # Liste) zaehlt. drive = element.drive_dir.strip().lower() if drive == "auf": return True, f"Antriebfahrtrichtung '{element.drive_dir}' (nach oben)" if drive == "ab": return False, f"Antriebfahrtrichtung '{element.drive_dir}' (nach unten)" return None, ( f"Foerderer/Strecke - Hoehe ist kein Richtungsindikator und " f"Antriebfahrtrichtung '{element.drive_dir or '-'}' nicht eindeutig " f"(Richtung kommt sonst aus AS/ES)" ) def _flow_uphill(element: Element, warnings: list[str]) -> bool | None: """Wie drive_uphill, meldet aber unbestimmbare Faelle als Warnung.""" uphill, reason = drive_uphill(element) if uphill is None: warnings.append(f"{element.describe()}: {reason} - Richtung nicht bestimmbar") return uphill def _orient( pairs: list[tuple[str, str]], nodes: dict[str, Node], warnings: list[str] ) -> list[Edge]: """Flussrichtung je Verbindung aus Hoehe und Antriebsrichtung ableiten.""" incident: dict[str, list[int]] = {} for index, (left, right) in enumerate(pairs): incident.setdefault(left, []).append(index) incident.setdefault(right, []).append(index) # Vorschlaege je Verbindung, gesammelt aus der Sicht der Transportobjekte proposals: dict[int, set[tuple[str, str]]] = {index: set() for index in range(len(pairs))} for node_id, node in sorted(nodes.items()): if node.kind not in TRANSPORT_KINDS: continue indices = incident.get(node_id, []) if not indices: warnings.append( f"{node.element.describe()}: keine Nachbarn - Knoten bleibt unverbunden" ) continue partners = {} for index in indices: left, right = pairs[index] partners[index] = right if left == node_id else left heights = {index: nodes[partner].height for index, partner in partners.items()} missing = [partners[i] for i, h in heights.items() if h is None] if missing: warnings.append( f"{node.element.describe()}: Hoehe von {', '.join(sorted(missing))} unbekannt - " f"Richtung nicht bestimmbar" ) continue uphill = _flow_uphill(node.element, warnings) if uphill is None: continue highest = max(heights.values()) if highest == min(heights.values()): warnings.append( f"{node.element.describe()}: alle Nachbarn auf gleicher Hoehe " f"({highest:.3f} m) - Richtung nicht bestimmbar" ) continue for index, partner in partners.items(): at_top = heights[index] == highest if uphill == at_top: proposals[index].add((node_id, partner)) else: proposals[index].add((partner, node_id)) edges: list[Edge] = [] for index, (left, right) in enumerate(pairs): found = proposals[index] if len(found) == 1: src, dst = next(iter(found)) edges.append(Edge(src, dst, EDGE_FLOW)) elif len(found) > 1: warnings.append( f"Verbindung {left} <-> {right}: widerspruechliche Richtungen " f"({', '.join(f'{s} -> {d}' for s, d in sorted(found))}) - beidseitig gezeichnet" ) edges.append(Edge(left, right, EDGE_UNRESOLVED, "widerspruechlich")) else: warnings.append( f"Verbindung {left} <-> {right}: Richtung nicht bestimmbar " f"(kein Transportobjekt beteiligt) - beidseitig gezeichnet" ) edges.append(Edge(left, right, EDGE_UNRESOLVED, "unbestimmt")) return edges def _nearest_node(station: Element, nodes: dict[str, Node]) -> str | None: """Naechster Nicht-Stations-Knoten zur Station (2D-Abstand).""" if station.x is None or station.y is None: return None best, best_d = None, None for node_id, node in nodes.items(): if node.element.is_station or node.element.x is None or node.element.y is None: continue d = (node.element.x - station.x) ** 2 + (node.element.y - station.y) ** 2 if best_d is None or d < best_d: best, best_d = node_id, d return best def _nearest_kreisel(station: Element, elements: list[Element]) -> Element | None: """Naechstgelegener Kreisel zur Station (2D-Abstand zur Kreisel-Achsposition).""" if station.x is None or station.y is None: return None best, best_d = None, None for el in elements: if not el.is_circle or el.x is None or el.y is None: continue d = (el.x - station.x) ** 2 + (el.y - station.y) ** 2 if best_d is None or d < best_d: best, best_d = el, d return best def _kreisel_containing(station_id: str, elements: list[Element]) -> Element | None: """Kreisel, in dessen TrackIds-Umlauf die Station liegt (oder None).""" for el in elements: if el.is_circle and station_id in el.track_ids: return el return None def _attach_stations( elements: list[Element], by_id: dict[str, Element], nodes: dict[str, Node], owners_of: dict[str, list[str]], warnings: list[str] ) -> list[Edge]: """ Ein-/Ausgabestationen (BTMT Be-/Entladung, Eingang/Ausgang) mit dem Fluss verbinden. Eingang (Beladung) ist Quelle: Kante Station -> Kreisel-Bahn. Ausgang (Entladung) ist Senke: Kante Kreisel-Bahn -> Station. Die Zielbahn kommt in dieser Reihenfolge: 1. Der Kreisel, in dessen TrackIds-Umlauf die Station liegt (verlaesslich, siehe Element.track_ids) - Eingangs-/Ausgangsbahn aus der Drehrichtung (_weiche_lane_roles: UZS -> Eingang R / Ausgang L, GUZ umgekehrt). 2. Ersatzweise: ein Flussobjekt aus den 'Nachbarn' (Separator/Station als Nachbar werden bewusst uebersprungen - sonst haengen zwei BTMT-Punkte aneinander statt an der Bahn). 3. Ersatzweise: der naechstgelegene Kreisel per Abstand, sonst der naechste Flussknoten. """ edges: list[Edge] = [] for element in elements: if not element.is_station: continue sid = element.teile_id if sid not in nodes: continue partners: list[str] = [] # owner = _track_owner(element, owners_of, by_id, warnings) # if owner is not None and owner.is_circle: # roles = _weiche_lane_roles(owner.teile_id, by_id, nodes) # if roles: # eingang, ausgang = roles # lane, role = (eingang, "Eingangsbahn") if element.is_entry else (ausgang, "Ausgangsbahn") # partners = [lane] # warnings.append( # f"{element.describe()}: aus TrackIds an Kreisel {owner.teile_id} " # f"Bahn {lane.split('-')[-1]} angebunden ({role}, Drehrichtung {owner.rotation})" # ) # else: # warnings.append( # f"{element.describe()}: TrackIds nennt Kreisel {owner.teile_id}, aber dessen " # f"Drehrichtung ist unbekannt - Bahn nicht bestimmbar" # ) # elif owner is not None and owner.teile_id in nodes: # partners = [owner.teile_id] # warnings.append(f"{element.describe()}: aus TrackIds an {owner.describe()} angebunden") # if not partners: # # 'Nachbarn' verweist bei BTMT-Stationen haeufig auf Separatoren # # oder die jeweils andere Station (naechste Objekte per Abstand, # # keine echte Flussverbindung) - eine andere Station wird deshalb # # nie als Partner akzeptiert, nur echte Flussobjekte. # for base, lane in element.neighbours: # if lane and f"{base}-{lane}" in nodes: # partners.append(f"{base}-{lane}") # elif base in nodes and not nodes[base].element.is_station: # 1. Kreisel aus dem TrackIds-Umlauf + Bahn aus der Drehrichtung. container = _kreisel_containing(sid, elements) if container is not None: roles = _weiche_lane_roles(container.teile_id, by_id, nodes) if roles is not None: eingang, ausgang = roles lane, role = ((eingang, "Eingangsbahn") if element.is_entry else (ausgang, "Ausgangsbahn")) partners = [lane] warnings.append( f"{element.describe()}: im Umlauf von Kreisel " f"{container.teile_id} - an Bahn {lane.split('-')[-1]} " f"angebunden ({role}, Drehrichtung {container.rotation})" ) # 2. Nachbarn, aber nur echte Flussobjekte (kein Separator/Station). if not partners: for base, lane in element.neighbours: partner_el = by_id.get(base) if partner_el is None or not (partner_el.is_transport or partner_el.is_circle): continue if lane and f"{base}-{lane}" in nodes: partners.append(f"{base}-{lane}") elif base in nodes: partners.append(base) else: partners += [f"{base}-{L}" for L in (LANE_LEFT, LANE_RIGHT) if f"{base}-{L}" in nodes] # 3. Naechster Kreisel per Abstand (Bahn aus Drehrichtung), sonst # naechster Flussknoten. if not partners and (element.is_entry or element.is_exit): kreisel = _nearest_kreisel(element, elements) roles = _weiche_lane_roles(kreisel.teile_id, by_id, nodes) if kreisel else None if roles: eingang, ausgang = roles lane, role = ((eingang, "Eingangsbahn") if element.is_entry else (ausgang, "Ausgangsbahn")) partners = [lane] warnings.append( f"{element.describe()}: nicht in einem TrackIds-Umlauf - " f"an naechsten Kreisel {kreisel.teile_id} Bahn " f"{lane.split('-')[-1]} angebunden ({role}, Drehrichtung " f"{kreisel.rotation})" ) if not partners: nearest = _nearest_node(element, nodes) if nearest: partners = [nearest] warnings.append( f"{element.describe()}: kein Umlauf/Flussnachbar - " f"naechstes Objekt '{nearest}' per Abstand verbunden" ) else: warnings.append(f"{element.describe()}: kein Ziel gefunden - unverbunden") continue for partner in partners: if element.is_entry: edges.append(Edge(sid, partner, EDGE_FLOW)) else: edges.append(Edge(partner, sid, EDGE_FLOW)) return edges def _kreisel_lane_node(kreisel_base: str, strecke: Element, nodes: dict[str, Node]) -> str | None: """Die konkrete Kreisel-Bahn (…-L/…-R), an der die Strecke haengt.""" for base, lane in strecke.neighbours: if base == kreisel_base: node_id = f"{base}-{lane}" if lane else base return node_id if node_id in nodes else None for lane in (LANE_LEFT, LANE_RIGHT): # Fallback: irgendeine Bahn if f"{kreisel_base}-{lane}" in nodes: return f"{kreisel_base}-{lane}" return kreisel_base if kreisel_base in nodes else None def _weiche_lane_roles(kreisel_base: str, by_id: dict[str, Element], nodes: dict[str, Node]) -> tuple[str, str] | None: """Eingangs-/Ausgangsbahn eines Kreisels an einer Kreisel<->Kreisel-Weiche. Ein Kreisel ist intern ein Umlauf zwischen seinen zwei Bahnen (Merkmal "Drehrichtung"); am Beruehrpunkt mit dem anderen Kreisel liegt daher zwingend die eine Bahn als Eingang (freies Ende vor dem internen Umlauf) und die andere als Ausgang (freies Ende danach): UZS (Umlauf R -> L) -> Eingang R, Ausgang L GUZ (Umlauf L -> R) -> Eingang L, Ausgang R None, wenn die Drehrichtung fehlt/unbekannt ist oder eine Bahn im Graph fehlt. """ element = by_id.get(kreisel_base) if element is None or not element.is_circle: return None left, right = f"{kreisel_base}-{LANE_LEFT}", f"{kreisel_base}-{LANE_RIGHT}" if left not in nodes or right not in nodes: return None if element.rotation == ROTATION_CW: return right, left if element.rotation == ROTATION_CCW: return left, right return None def _apply_transfers( elements: list[Element], by_id: dict[str, Element], nodes: dict[str, Node], edges: list[Edge], warnings: list[str] ) -> list[Edge]: """ Kreisel<->Strecke-Kanten anhand der Transferelemente gerichtet setzen. Ausschleuselement: Kreisel -> Strecke (Streckenanfang). Einschleuselement: Strecke -> Kreisel (Streckenende). Weiche: Kreisel<->Kreisel, Bahn und Richtung aus der Drehrichtung beider Kreisel (siehe _weiche_lane_roles); ohne bekannte Drehrichtung bleibt die Verbindung unbestimmt, wird aber weiterhin als Weiche markiert. Die so definierten Kanten sind massgeblich und ersetzen die aus Hoehe/ Nachbarn abgeleitete Richtung fuer genau diese Knotenpaare. """ # Transfer je Strecke sammeln: aus = Einlauf (Kreisel->Strecke, Anfang), # ein = Auslauf (Strecke->Kreisel, Ende) per_strecke: dict[str, dict[str, str]] = {} weiche_pairs: list[set[str]] = [] weiche_ab: list[tuple[str, str]] = [] for el in elements: if el.kind in ("Ausschleus", "Einschleus"): k = str(_merkmal(el.merkmale, "Kreisel") or "").strip() # Der Export benennt die andockende Strecke je nach Bauart als # "Strecke" (VF) oder "Foerderer"/"Förderer" (GF/Streckengruppe). s = str(_merkmal(el.merkmale, "Strecke", "Foerderer", "Förderer") or "").strip() if k and s: role = "aus" if el.kind == "Ausschleus" else "ein" per_strecke.setdefault(s, {})[role] = k elif el.kind == "Weiche": a = str(_merkmal(el.merkmale, "Kreisel A") or "").strip() b = str(_merkmal(el.merkmale, "Kreisel B") or "").strip() if a and b: weiche_pairs.append({a, b}) weiche_ab.append((a, b)) # Weiche zwischen zwei Kreiseln: Bahn und Richtung deterministisch aus der # Drehrichtung beider Kreisel ableiten (Ausgang der einen Bahn -> Eingang # der anderen, in beiden Richtungen) statt aus dem rohen Nachbarn-Kreuzprodukt, # das links/rechts nicht unterscheiden kann. weiche_edges: list[Edge] = [] weiche_resolved: set[frozenset] = set() for a, b in weiche_ab: roles_a = _weiche_lane_roles(a, by_id, nodes) roles_b = _weiche_lane_roles(b, by_id, nodes) if roles_a is None or roles_b is None: warnings.append( f"Weiche {a}<->{b}: Drehrichtung mindestens eines Kreisels unbekannt - " f"Bahnzuordnung nicht bestimmbar, Verbindung bleibt unbestimmt" ) continue in_a, out_a = roles_a in_b, out_b = roles_b weiche_edges.append(Edge(out_a, in_b, EDGE_FLOW, "Weiche")) weiche_edges.append(Edge(out_b, in_a, EDGE_FLOW, "Weiche")) weiche_resolved.add(frozenset({a, b})) directed: dict[frozenset, tuple[str, str]] = {} for s, roles in per_strecke.items(): if s not in nodes or nodes[s].element.is_circle: continue strecke_el = by_id.get(s) aus_lane = _kreisel_lane_node(roles["aus"], strecke_el, nodes) if roles.get("aus") else None ein_lane = _kreisel_lane_node(roles["ein"], strecke_el, nodes) if roles.get("ein") else None # Ruecklauf-Strecke (beide Enden am selben Kreisel): auf die zwei Bahnen verteilen if aus_lane and aus_lane == ein_lane: base = aus_lane.split("-")[0] lanes = [f"{base}-{L}" for L in (LANE_LEFT, LANE_RIGHT) if f"{base}-{L}" in nodes] if len(lanes) == 2: aus_lane, ein_lane = lanes warnings.append( f"Strecke {s}: Ein- und Ausschleusung am selben Kreisel {base} " f"(Ruecklauf) - Bahnen {lanes[0]}/{lanes[1]} angenommen" ) if aus_lane: # Kreisel -> Strecke directed[frozenset({aus_lane, s})] = (aus_lane, s, "Ausschleusung") if ein_lane: # Strecke -> Kreisel directed[frozenset({ein_lane, s})] = (s, ein_lane, "Einschleusung") # Fuer Strecken mit Transferelementen sind ALLEIN diese massgeblich fuer die # Kreisel<->Strecke-Verbindung. Jede weitere Kreisel<->Strecke-Kante aus den # rohen Nachbarn (Bounding-Box-Ueberlappung, z. B. ein VF, der zufaellig neben # einem dritten Kreisel liegt) ist ein Falschtreffer und wird verworfen. covered = set(per_strecke) result: list[Edge] = [] superseded: set[tuple[str, str]] = set() for e in edges: if frozenset({e.src, e.dst}) in directed: continue # wird gerichtet neu gesetzt if frozenset({e.src.split("-")[0], e.dst.split("-")[0]}) in weiche_resolved: superseded.add((e.src, e.dst)) continue # wird durch die Weiche-Kanten ersetzt ends = {e.src, e.dst} strecke_end = ends & covered others = ends - strecke_end if strecke_end and len(others) == 1: other = next(iter(others)) if other in nodes and nodes[other].kind in CIRCLE_KINDS: warnings.append( f"Kante {other} <-> {next(iter(strecke_end))}: Kreisel<->Strecke " f"ohne Transferelement - als Bounding-Box-Falschtreffer verworfen" ) continue result.append(e) for (src, dst, note) in directed.values(): result.append(Edge(src, dst, EDGE_FLOW, note)) result.extend(weiche_edges) # Die anfaengliche Hoehen-basierte Ausrichtung (_orient) kann eine # Kreisel<->Kreisel-Verbindung nie bestimmen und meldet sie immer als # "beidseitig gezeichnet" - das stimmt nicht mehr, sobald die Weiche sie # gerade durch obenstehende Kanten ersetzt hat. Die dadurch ueberholte # Meldung wird entfernt. if superseded: stale = { f"Verbindung {a} <-> {b}: Richtung nicht bestimmbar " f"(kein Transportobjekt beteiligt) - beidseitig gezeichnet" for src, dst in superseded for a, b in ((src, dst), (dst, src)) } warnings[:] = [w for w in warnings if w not in stale] # Kreisel<->Kreisel-Kanten, die eine Weiche verbinden, markieren (deckt vor # allem die Faelle ab, die mangels Drehrichtung nicht gerichtet aufgeloest # werden konnten - die aufgeloesten Weiche-Kanten tragen die Markierung # bereits). for e in result: if {e.src.split("-")[0], e.dst.split("-")[0]} in weiche_pairs: e.note = "Weiche" return result def _merge_station_symbols( elements: list[Element], warnings: list[str] | None = None ) -> set[str]: """ Generische Eingang/Ausgang-Symbole ermitteln, die deckungsgleich ueber einer konkreten BTMT-Station liegen (gleiche Rolle Quelle/Senke, Abstand <= STATION_MERGE_MM). Diese werden verschluckt: es bleibt nur die BTMT-Station als Knoten. Rueckgabe = TeileIds der verschluckten Symbole. Eingang gehoert zu Beladung (beide Quelle), Ausgang zu Entladung (beide Senke). Ein Symbol ohne passende nahe BTMT-Station bleibt ein eigener Knoten. warnings=None unterdrueckt die Meldungen (zweiter Aufruf im --track-Pfad, der sonst dieselben Hinweise ein zweites Mal ausgibt). """ swallowed: set[str] = set() btmt = [e for e in elements if e.is_station and e.kind not in STATION_SYMBOL_KINDS] for sym in elements: if sym.kind not in STATION_SYMBOL_KINDS or sym.x is None or sym.y is None: continue sym_is_entry = sym.kind in ENTRY_KINDS best, best_d = None, None for st in btmt: if st.x is None or st.y is None: continue if st.is_entry != sym_is_entry: # nur gleiche Rolle (Quelle/Senke) continue d = ((st.x - sym.x) ** 2 + (st.y - sym.y) ** 2) ** 0.5 if d <= STATION_MERGE_MM and (best_d is None or d < best_d): best, best_d = st, d if best is not None: swallowed.add(sym.teile_id) if warnings is not None: warnings.append( f"{sym.describe()}: deckungsgleich mit {best.kind} {best.teile_id} " f"({best_d:.0f} mm) - Symbol verschluckt, nur {best.kind} gezeichnet" ) return swallowed def build_graph(elements: list[Element], source: str, warnings: list[str]) -> Graph: # Generische Eingang/Ausgang-Symbole, die ueber einer BTMT-Station liegen, # verschlucken (nur die konkrete Station bleibt Knoten). swallowed = _merge_station_symbols(elements, warnings) if swallowed: elements = [e for e in elements if e.teile_id not in swallowed] by_id = {element.teile_id: element for element in elements} nodes = _build_nodes(elements, warnings) owners_of = _build_track_owners(elements, warnings) unassigned = _attach_devices(elements, nodes, by_id, owners_of, warnings) observed = _collect_links(elements, by_id, warnings) pairs: set[tuple[str, str]] = set() for key, lanes in sorted(observed.items(), key=lambda item: sorted(item[0])): base_a, base_b = sorted(key) for node_a in _endpoints(base_a, lanes[base_a], by_id, warnings, base_b): for node_b in _endpoints(base_b, lanes[base_b], by_id, warnings, base_a): pairs.add((node_a, node_b)) edges = _orient(sorted(pairs), nodes, warnings) # Kreisel: Richtung zwischen den beiden Bahnen aus der Drehrichtung # UZS (im Uhrzeigersinn) -> von rechts nach links, R -> L # GUZ (gegen den Uhrzeigersinn) -> von links nach rechts, L -> R # Fehlt die Angabe, bleibt der Umlauf beidseitig und wird gemeldet. for element in elements: if not element.is_circle: continue left = f"{element.teile_id}-{LANE_LEFT}" right = f"{element.teile_id}-{LANE_RIGHT}" rotation = element.rotation if rotation == ROTATION_CW: edges.append(Edge(right, left, EDGE_CIRCLE, ROTATION_CW)) elif rotation == ROTATION_CCW: edges.append(Edge(left, right, EDGE_CIRCLE, ROTATION_CCW)) else: warnings.append( f"{element.describe()}: Drehrichtung '{element.rotation or '-'}' " f"unbekannt (erwartet {ROTATION_CW} oder {ROTATION_CCW}) - " f"Umlauf beidseitig gezeichnet" ) edges.append(Edge(left, right, EDGE_CIRCLE, "unbestimmt")) edges.append(Edge(right, left, EDGE_CIRCLE, "unbestimmt")) # Ein-/Ausgabestationen (BTMT Be-/Entladung) als Quelle/Senke anbinden edges.extend(_attach_stations(elements, by_id, nodes, owners_of, warnings)) # Kreisel<->Strecke gerichtet aus den Transferelementen (massgeblich) edges = _apply_transfers(elements, by_id, nodes, edges, warnings) # Sackgassen und unerreichbare Bahnen melden. Sie entstehen, wenn ein # Kreisel nur in einer Richtung durchlaufen wird (Drehrichtung) und die # Gegenrichtung nicht ueber aeussere Verbindungen geschlossen ist. outgoing = {edge.src for edge in edges} incoming = {edge.dst for edge in edges} for node_id in sorted(nodes): node = nodes[node_id] if node.lane is None: continue devices = node.devices.get("Separator", []) detail = f" (Separatoren {', '.join(sorted(devices))})" if devices else "" if node_id not in outgoing: warnings.append( f"Bahn {node_id}{detail}: keine abgehende Verbindung - Sackgasse. " f"Drehrichtung {node.element.rotation or '?'} laesst das Material " f"nur in eine Richtung laufen" ) if node_id not in incoming: warnings.append( f"Bahn {node_id}{detail}: keine ankommende Verbindung - fuer das " f"Material nicht erreichbar" ) return Graph(nodes=nodes, edges=edges, warnings=warnings, unassigned=unassigned, source=source) # --------------------------------------------------------------------------- # Bahnzusammenhang-Graph (--track, siehe doc/track.md) # --------------------------------------------------------------------------- # # Feinerer Graph als build_graph(): Separator/Ausschleus/Einschleus/Weiche und # die Ein-/Ausgangspunkte (BTMT, Eingang/Ausgang) werden hier zu eigenen Knoten # (statt nur Kanten/Zuordnungen zu liefern wie im normalen Materialfluss). # # Quelle der Fein-Adjazenz ist die CSV-Spalte "TrackIds" (Element.track_ids): # jeder Container (Kreisel/Strecke/Gefaellestrecke) fuehrt die geordnete Liste # der auf ihm liegenden Bahnelemente. Der dxfmakros-Export berechnet diese # Reihenfolge bereits (Kreisel: Umlauf nach Winkel in Drehrichtung; Strecke/ # Gefaelle: Kette AS -> SEP.. -> ES in Baureihenfolge) - siehe # dxfmakros/doc/csv-export.md. build_track_graph zieht daraus je Container eine # gerichtete Kette; ein Element, das in zwei Listen steht (ein AS/ES in seiner # Strecke UND im Kreisel-Umlauf, eine Weiche in beiden Kreiseln), ist derselbe # Knoten und verkettet die Container automatisch. Die frueher noetige # Rekonstruktion der Reihenfolge aus der "Nachbarn"-Spalte (Rueckmeldungs-/ # Rollen-Heuristik) entfaellt damit. SYMBOL_ENTRY = "Eingang" SYMBOL_EXIT = "Ausgang" def _attach_track_stations( elements: list[Element], graph: Graph, track: TrackGraph, placeholder, add_edge ) -> None: """ BTMT Be-/Entladung an den Bahn-Platzhalter anbinden. Die Zielbahn wurde bereits von _attach_stations() im groben Graph bestimmt (naechster Kreisel + Eingangs-/Ausgangsbahn aus der Drehrichtung, siehe dort) - hier nur wiederverwendet statt neu berechnet. Eine Station, die schon ueber eine TrackIds-Kette (Kreisel-Umlauf/Strecke, Schritte 1+2 in build_track_graph) real mit einem anderen Bahnelement verbunden ist, braucht keine zusaetzliche Bruecke zur groben Kreisel-*Bahn* mehr - sonst haengt am sichtbar schon einsortierten Ein-/Ausgang ein ueberfluessiger "Bahn X"-Platzhalterknoten (Reihenfolge/Position offen) mit dran, obwohl die Reihenfolge laengst geklaert ist. Nur eine Station OHNE echte Kante (nicht Teil einer TrackIds-Kette) wird ueberhaupt an den Platzhalter gebrueckt; grobe Station<->Station-Kanten (ein Artefakt der Abstands- Anbindung im groben Graph) werden hier so oder so NICHT uebernommen, sonst entstuenden verwirrende Rueckkanten zwischen zwei BTMT-Punkten, die im Umlauf schon geordnet sind. """ already_linked = {edge.src for edge in track.edges} | {edge.dst for edge in track.edges} for element in elements: if not element.is_station: continue sid = element.teile_id if sid not in graph.nodes: continue track.nodes.setdefault(sid, TrackNode(sid, element.kind, element)) if sid in already_linked: continue for edge in graph.edges: partner = edge.dst if edge.src == sid else edge.src if edge.dst == sid else None if partner is None: continue partner_node = graph.nodes.get(partner) if partner_node is not None and partner_node.element.is_station: continue # Station<->Station-Kante des groben Graphen ueberspringen if edge.src == sid: add_edge(sid, placeholder(partner), "Station") else: add_edge(placeholder(partner), sid, "Station") def _attach_track_symbols( track: TrackGraph, add_edge, warnings: list[str] ) -> None: """ Offene Bahnenden ohne BTMT-Station als generisches Symbol Eingang/Ausgang markieren - aber anhand des FEINEN Graphen (track), nicht des groben. Ein echter System-Ein-/Ausgang ist ein reales Bahnelement (Separator, AS/ES, Station), das im --track-Graphen keinen Vorgaenger (Eingang) bzw. keinen Nachfolger (Ausgang) hat. Muss darum NACH allen Ketten- und Stations-Kanten laufen. Der grobe Graph taugt hier NICHT als Massstab: er kann eine Bahn als offenes Ende fuehren, obwohl ihre AS/ES-Enden im Kreisel-Umlauf laengst verkettet sind (die Richtung im groben Graph blieb nur 'unbestimmt') - genau das erzeugte frueher die vereinzelten Falsch-Eingaenge (z. B. Bahn 0001-0004 in HundM05). Platzhalter- und Symbol-Knoten selbst zaehlen nicht als Bahn. """ incoming = {e.dst for e in track.edges} outgoing = {e.src for e in track.edges} symbol_no = 0 for node_id in sorted(track.nodes): node = track.nodes[node_id] # Nur echte Bahnelemente betrachten: keine Platzhalter, keine schon # gesetzten Symbole, keine BTMT-Stationen (die sind selbst die Grenze). if node.placeholder or node.kind in (SYMBOL_ENTRY, SYMBOL_EXIT): continue if node.element is not None and node.element.is_station: continue if node_id not in incoming: symbol_no += 1 sym_id = f"SYM_EIN_{symbol_no:02d}" track.nodes[sym_id] = TrackNode(sym_id, SYMBOL_ENTRY) add_edge(sym_id, node_id, "Symbol") warnings.append( f"{node_id}: keine ankommende Verbindung und keine BTMT-Station - " f"im --track-Graph als Symbol {SYMBOL_ENTRY} markiert" ) if node_id not in outgoing: symbol_no += 1 sym_id = f"SYM_AUS_{symbol_no:02d}" track.nodes[sym_id] = TrackNode(sym_id, SYMBOL_EXIT) add_edge(node_id, sym_id, "Symbol") warnings.append( f"{node_id}: keine abgehende Verbindung und keine BTMT-Station - " f"im --track-Graph als Symbol {SYMBOL_EXIT} markiert" ) # Kanten-Notiz je Container-Art: eine Kette innerhalb einer Strecke/ # Gefaellestrecke ist eine Weiterfahrt auf derselben Bahn, ein Schritt im # Kreisel-Umlauf gehoert zur zirkulaeren Kreiselbahn. TRACK_NOTE_STRECKE = "Bahn" TRACK_NOTE_KREISEL = "Umlauf" def build_track_graph( elements: list[Element], by_id: dict[str, Element], graph: Graph, warnings: list[str] ) -> TrackGraph: """ Feiner Bahnzusammenhang-Graph fuer --track (siehe doc/track.md). Baut die Fein-Adjazenz direkt aus der CSV-Spalte "TrackIds" auf: jeder Container (Kreisel, Strecke, Gefaellestrecke) fuehrt dort die geordnete Liste der auf ihm liegenden Bahnelemente (Element.track_ids). Daraus wird je Container eine gerichtete Kette gezogen: * Kreisel -> zirkulaerer Umlauf in Drehrichtung (die Liste ist bereits nach Winkel in DREHRICHTUNG sortiert); aufeinanderfolgende Eintraege werden verbunden, das Ende schliesst an den Anfang zurueck (der Umlauf ist tatsaechlich geschlossen). * Strecke/ -> lineare Kette AS -> SEP.. -> ES; die Richtung kommt aus den Gefaelle Kettenenden (Ausschleus = Anfang, Einschleus = Ende), nicht aus der Listenreihenfolge. Elemente, die in zwei Containern gelistet sind (ein Ausschleus-/ Einschleuselement steht in seiner Strecke UND im Umlauf des angedockten Kreisels; eine Kreisel-Kreisel-Weiche steht in beiden Kreiseln), sind derselbe Knoten und verketten die Container automatisch - ohne die frueher noetige Rueckmeldungs-/Rollen-Heuristik aus der "Nachbarn"-Spalte. warnings wird mit build_graph() geteilt, damit Fein-Hinweise im selben Bericht erscheinen. """ # Deckungsgleiche Eingang/Ausgang-Symbole (wie im groben Graph) verschlucken: # aus der Element-Liste UND aus den TrackIds jedes Containers entfernen, sonst # laeuft der Kreisel-Umlauf durch zwei nahezu identische Knoten (Symbol + # BTMT-Station). warnings=None: die Meldung stand schon im groben Graph. swallowed = _merge_station_symbols(elements, warnings=None) if swallowed: elements = [ replace(e, track_ids=[t for t in e.track_ids if t not in swallowed]) for e in elements if e.teile_id not in swallowed ] by_id = {e.teile_id: e for e in elements} track = TrackGraph(source=graph.source) seen_edges: set[tuple[str, str]] = set() def add_edge(src: str | None, dst: str | None, note: str = "") -> None: # Dieselbe gerichtete Kante kann aus zwei Containern kommen (ein Paar, das # z. B. im Kreisel-Umlauf UND am Ende einer andockenden Strecke direkt # aufeinanderfolgt) - nur einmal ziehen, mit der zuerst gesehenen Notiz. if src is None or dst is None or src == dst or (src, dst) in seen_edges: return seen_edges.add((src, dst)) track.edges.append(TrackEdge(src, dst, note)) def placeholder(host_id: str | None, reason: str = "") -> str | None: if not host_id: return None if host_id not in track.nodes: coarse = graph.nodes.get(host_id) track.nodes[host_id] = TrackNode( host_id, "Bahn", coarse.element if coarse else None, placeholder=True, note=reason ) return host_id def ensure_node(tid: str) -> str | None: """Knoten fuer ein in TrackIds gelistetes Element sicherstellen.""" if tid in track.nodes: return tid el = by_id.get(tid) if el is None: warnings.append(f"TrackIds nennt unbekanntes Element {tid} - uebersprungen") return None track.nodes[tid] = TrackNode(tid, el.kind, el) return tid # 1. Strecke/Gefaellestrecke: lineare Kette AS -> SEP.. -> ES. for el in elements: if el.kind not in TRANSPORT_KINDS or not el.track_ids: continue chain = list(el.track_ids) # Richtung an den Kettenenden ausrichten: Einschleus vorne bedeutet, die # Liste laeuft entgegen dem Fluss - umdrehen. Ausschleus am Ende ebenso. first_el, last_el = by_id.get(chain[0]), by_id.get(chain[-1]) first_kind = first_el.kind if first_el else "" last_kind = last_el.kind if last_el else "" if first_kind == "Einschleus" or last_kind == "Ausschleus": chain = list(reversed(chain)) prev = None for tid in chain: node = ensure_node(tid) if node is None: continue if prev is not None: add_edge(prev, node, TRACK_NOTE_STRECKE) prev = node # 2. Kreisel: zirkulaerer Umlauf in Drehrichtung (Liste bereits sortiert). # Ohne bekannte Drehrichtung ist die Umlaufrichtung nicht gesichert - dann # wird die Kette offen (ohne Rueckschluss) gezogen und gemeldet. for el in elements: if not el.is_circle or not el.track_ids: continue chain = [n for tid in el.track_ids if (n := ensure_node(tid)) is not None] if len(chain) < 2: continue for a, b in zip(chain, chain[1:]): add_edge(a, b, TRACK_NOTE_KREISEL) if el.rotation in (ROTATION_CW, ROTATION_CCW): add_edge(chain[-1], chain[0], TRACK_NOTE_KREISEL) # Umlauf schliessen else: warnings.append( f"{el.describe()}: Drehrichtung '{el.rotation or '-'}' unbekannt - " f"Kreisel-Umlauf im --track-Graph nicht geschlossen" ) # 3. Verwaiste Bahnelemente (in keiner TrackIds-Liste eines Containers, z. B. # ein frei platzierter Separator ohne Kette): an ihren Zuordnung- # Platzhalter haengen statt sie unverbunden zu lassen. listed = {tid for el in elements for tid in el.track_ids} for el in elements: if el.kind not in TRACK_KINDS or el.teile_id in listed: continue track.nodes.setdefault(el.teile_id, TrackNode(el.teile_id, el.kind, el)) host = el.assignment if el.assignment and el.assignment != UNASSIGNED else None ph = placeholder(host, "nicht in einer TrackIds-Kette enthalten") if ph: add_edge(ph, el.teile_id, "unbestimmt") warnings.append( f"{el.describe()}: in keiner TrackIds-Kette enthalten - " f"Reihenfolge auf {host} ungeklaert" ) else: warnings.append( f"{el.describe()}: in keiner TrackIds-Kette und ohne Zuordnung - " f"im --track-Graph isoliert" ) # 4. Systemgrenzen: BTMT-Stationen (wiederverwendet) bzw. generisches # Eingang/Ausgang-Symbol an offenen Bahnenden ohne Station. _attach_track_stations(elements, graph, track, placeholder, add_edge) _attach_track_symbols(track, add_edge, warnings) # Letzte Sicherung: ein gelistetes Element ohne jede Kante sichtbar melden. touched = {e.src for e in track.edges} | {e.dst for e in track.edges} for tid in sorted(listed): el = by_id.get(tid) if el is not None and el.kind in TRACK_KINDS and tid not in touched: warnings.append( f"{el.describe()}: im --track-Graph ohne jede Kante - TrackIds widerspruechlich" ) # 5. Scanner bilden weiterhin keinen eigenen Knoten (siehe TRACK_KINDS), aber # ihre "Naechster Separator"-Angabe wird aufgeloest und die eigene # Zuordnung des Scanners auf dem betroffenen Separator vermerkt - # render_track_dot() zeichnet dafuer einen roten Punkt unten mittig ans # Separator-Rechteck, statt einen zusaetzlichen Knoten einzufuehren. for el in elements: if el.kind != "Scanner": continue sep_id = str(_merkmal(el.merkmale, "Naechster Separator") or "").strip() sep_node = track.nodes.get(sep_id) if sep_node is None or sep_node.kind != "Separator": if sep_id: warnings.append( f"{el.describe()}: 'Naechster Separator' {sep_id} ist im --track-Graph " f"kein Separator-Knoten - Scanner-Zuordnung nicht dargestellt" ) continue sep_node.scanner_zuordnung = el.assignment or sep_node.scanner_zuordnung return track # --------------------------------------------------------------------------- # DOT erzeugen # --------------------------------------------------------------------------- def _node_label(node: Node) -> str: element = node.element lines: list[str] = [] if element.is_station: role = "Materialeingang" if element.is_entry else "Materialausgang" lines.append(f"{element.name or element.kind} [{element.teile_id}]") lines.append(f"{element.kind} - {role}") if element.height is not None: lines.append(f"h = {element.height:.3f} m") return _dot_label(lines) if node.lane: name = _merkmal(element.merkmale, "Name") or element.name lines.append(f"{name} [{node.node_id}]") lines.append(f"Bahn {node.lane} ({LANE_NAMES[node.lane]})") else: lines.append(f"{element.name} [{element.teile_id}]") lines.append(element.kind) height = element.height if height is not None: lines.append(f"h = {height:.3f} m") if element.is_transport: uphill = drive_uphill(element)[0] arrow = {True: "aufwaerts", False: "abwaerts", None: "Richtung offen"}[uphill] if element.rise is not None: lines.append(f"{element.height_from:.0f} -> {element.height_to:.0f} mm " f"({element.rise:+.0f}) = {arrow}") elif element.drive_dir: lines.append(f"Antrieb: {element.drive_dir} = {arrow}") else: lines.append(arrow) for kind in DEVICE_KINDS: ids = node.devices.get(kind) if ids: lines.append(f"{kind} ({len(ids)}): {', '.join(sorted(ids))}") return _dot_label(lines) # Zeichnungsmassstab fuer --use-cords: Graphviz rechnet in Punkt, die Anlage in # Millimeter. 0.25 pt/mm (1:4) - wie in tro_flow.py, damit Material- und TRO-Graph # im selben Massstab lagerichtig sind. COORD_SCALE = 0.25 # Die beiden Bahnen (L/R) eines Kreisels teilen sich eine Position; im # lagerichtigen Layout leicht versetzen, damit sie nicht uebereinander liegen. LANE_NUDGE_MM = 350.0 def _node_xy(node: "Node") -> tuple[float, float] | None: """Anlagenkoordinate eines Knotens (Kreisel-Bahnen L/R leicht versetzt).""" el = node.element if el.x is None or el.y is None: return None x, y = el.x, el.y if node.lane == "L": y += LANE_NUDGE_MM elif node.lane == "R": y -= LANE_NUDGE_MM return (x, y) def _pos_attr(node: "Node", use_coords: bool) -> str: """pos-Attribut fuer 'neato -n', oder leer ohne --use-cords.""" if not use_coords: return "" xy = _node_xy(node) if xy is None: return "" return f', pos="{xy[0] * COORD_SCALE:.2f},{xy[1] * COORD_SCALE:.2f}"' def _flow_neighbours(node_id: str, edges: list[Edge]) -> tuple[str, str] | None: """ (Vorgaenger, Nachfolger) eines Knotens im Materialfluss. None, wenn der Knoten nicht genau einen ankommenden und einen abgehenden EDGE_FLOW hat - dann ist er kein einfacher Durchlaufknoten zwischen zwei festen Nachbarn. """ incoming = {e.src for e in edges if e.dst == node_id and e.kind == EDGE_FLOW} outgoing = {e.dst for e in edges if e.src == node_id and e.kind == EDGE_FLOW} if len(incoming) != 1 or len(outgoing) != 1: return None return next(iter(incoming)), next(iter(outgoing)) def _group_parallel_gefaelle(graph: Graph) -> list[list[Node]]: """ Parallele Gefaellestrecken zwischen denselben zwei Kreisel-Bahnen zu Gruppen zusammenfassen (z. B. eine Bank aus Schwerkraftlinien). Kriterium: gleicher Zulauf- und Ablauf-Knoten (jeweils eine Kreisel-Bahn) und gleiche Steigung (Hoehe_Von/Hoehe_Bis) - "gleich lang und gleiche Steigung" laesst sich aus dem Export nur ueber die Hoehenangabe pruefen, die Laenge ergibt sich daraus, dass beide Enden an denselben zwei Knoten haengen. Rein fuer die Darstellung; Graph-Knoten/-Kanten bleiben unveraendert. """ groups: dict[tuple, list[Node]] = {} for node_id in sorted(graph.nodes): node = graph.nodes[node_id] if node.kind != "Gefaellestrecke": continue neighbours = _flow_neighbours(node_id, graph.edges) if neighbours is None: continue before, after = neighbours if graph.nodes[before].kind != "Kreisel" or graph.nodes[after].kind != "Kreisel": continue element = node.element key = (before, after, element.height_from, element.height_to) groups.setdefault(key, []).append(node) return [members for members in groups.values() if len(members) >= MIN_GEFAELLE_GROUP] def _id_range(ids: list[str]) -> str: """Sortierte Ids als Bereich, z. B. ['0037', '0056'] -> '0037 - 0056'.""" ids = sorted(ids) return ids[0] if len(ids) == 1 else f"{ids[0]} - {ids[-1]}" def _gefaelle_group_label(members: list[Node]) -> str: """Sammel-Label einer Gefaellestrecken-Gruppe: Bereich der Linien-Ids, gemeinsame Steigung, zusammengefasste Separatoren/Scanner.""" element = members[0].element ids = sorted(m.node_id for m in members) lines = [ f"Gefaellestrecke {ids[0]} -> {ids[-1]} ({len(ids)}x)", "Gruppe paralleler Schwerkraftlinien", ] if element.height is not None: lines.append(f"h = {element.height:.3f} m") if element.rise is not None: uphill = drive_uphill(element)[0] arrow = {True: "aufwaerts", False: "abwaerts", None: "Richtung offen"}[uphill] lines.append(f"{element.height_from:.0f} -> {element.height_to:.0f} mm " f"({element.rise:+.0f}) = {arrow}") for kind in DEVICE_KINDS: ids_dev = sorted({d for m in members for d in m.devices.get(kind, [])}) if ids_dev: lines.append(f"{kind} ({len(ids_dev)}): {_id_range(ids_dev)}") return _dot_label(lines) def render_dot(graph: Graph, use_coords: bool = False) -> str: out: list[str] = [] add = out.append add("// Automatisch erzeugt von lib/material_flow.py - nicht manuell aendern.") add(f"// Quelle: {graph.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 Materialfluss {") add(' graph [rankdir=LR, splines=spline, nodesep=0.25, ranksep=1.6,') add(' fontname="Segoe UI", fontsize=11, labelloc="t",') add(f' label="Materialfluss - {_dot_escape(graph.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("") # Kreisel als Cluster mit den beiden Bahnen circles: dict[str, list[Node]] = {} plain: list[Node] = [] for node_id in sorted(graph.nodes): node = graph.nodes[node_id] if node.lane: circles.setdefault(node.element.teile_id, []).append(node) else: plain.append(node) for index, (teile_id, lanes) in enumerate(sorted(circles.items())): element = lanes[0].element name = _merkmal(element.merkmale, "Name") or element.name rot = element.rotation rot_text = (f" {rot} ({ROTATION_FLOW[rot]})" if rot in ROTATION_FLOW else f" Drehrichtung {rot or '?'}") add(f" subgraph cluster_kreisel_{index} {{") add(f' label="{_dot_escape(element.kind)} {_dot_escape(str(name))} ' f'[{teile_id}]{_dot_escape(rot_text)}";') add(' style="rounded,filled"; fillcolor="#eef2fa"; color="#2f5597";') add(' fontname="Segoe UI"; fontsize=10; fontcolor="#1f3864"; margin=12;') style = NODE_STYLES.get(element.kind, DEFAULT_NODE_STYLE) for node in sorted(lanes, key=lambda n: n.lane or ""): attrs = ", ".join(f'{key}="{value}"' for key, value in style.items()) add(f' "{node.node_id}" [label="{_node_label(node)}", {attrs}' f'{_pos_attr(node, use_coords)}];') add(" { rank=same; " + " ".join(f'"{n.node_id}";' for n in lanes) + " }") add(" }") add("") # Parallele Gefaellestrecken (gleicher Zu-/Ablauf, gleiche Steigung) zu # je einem Block zusammenfassen, damit eine Linienbank nicht als N # einzelne Kaesten gezeichnet wird. gefaelle_groups = _group_parallel_gefaelle(graph) group_of: dict[str, str] = {} for members in gefaelle_groups: group_id = f"grp_{members[0].node_id}" for member in members: group_of[member.node_id] = group_id grouped_ids = set(group_of) for node in plain: if node.node_id in grouped_ids: continue style = NODE_STYLES.get(node.kind, DEFAULT_NODE_STYLE) attrs = ", ".join(f'{key}="{value}"' for key, value in style.items()) add(f' "{node.node_id}" [label="{_node_label(node)}", {attrs}' f'{_pos_attr(node, use_coords)}];') for members in gefaelle_groups: group_id = f"grp_{members[0].node_id}" style = NODE_STYLES.get("Gefaellestrecke", DEFAULT_NODE_STYLE) attrs = ", ".join(f'{key}="{value}"' for key, value in style.items()) pos = "" if use_coords: xs = [xy[0] for m in members if (xy := _node_xy(m))] ys = [xy[1] for m in members if (xy := _node_xy(m))] if xs and ys: pos = (f', pos="{sum(xs) / len(xs) * COORD_SCALE:.2f},' f'{sum(ys) / len(ys) * COORD_SCALE:.2f}"') add(f' "{group_id}" [label="{_gefaelle_group_label(members)}", {attrs}, ' f'peripheries=2{pos}];') add("") seen_group_edges: set[tuple[str, str]] = set() for edge in graph.edges: src = group_of.get(edge.src, edge.src) dst = group_of.get(edge.dst, edge.dst) if edge.src in grouped_ids or edge.dst in grouped_ids: if src == dst or (src, dst) in seen_group_edges: continue seen_group_edges.add((src, dst)) if edge.kind == EDGE_CIRCLE: # Richtung kommt aus der Drehrichtung und steht als Label an der Kante label = f', xlabel="{_dot_escape(edge.note)}"' if edge.note else "" attrs = ('[color="#8ea9db", style=dashed, penwidth=1.2, constraint=false, ' 'arrowsize=0.9, fontcolor="#5b7fc7", ' f'tooltip="Kreisel-Umlauf {edge.note}"{label}]') elif edge.kind == EDGE_UNRESOLVED: weiche = ', xlabel="Weiche", fontcolor="#cc8800"' if edge.note == "Weiche" else "" attrs = ('[color="#bf8f00", style=dashed, dir=both, ' f'tooltip="Richtung {edge.note}"{weiche}]') elif edge.note == "Weiche": attrs = ('[color="#cc8800", penwidth=1.6, xlabel="Weiche", ' 'fontcolor="#cc8800", tooltip="Weiche (Kreisel-Uebergang)"]') else: attrs = "[]" add(f' "{src}" -> "{dst}" {attrs};') add("") legend_pos = "" if use_coords: xs = [xy[0] for n in graph.nodes.values() if (xy := _node_xy(n))] ys = [xy[1] for n in graph.nodes.values() if (xy := _node_xy(n))] if xs and ys: lx, ly = min(xs), min(ys) - 3000.0 # etwas unterhalb der Anlage legend_pos = (f' pos="{lx * COORD_SCALE:.2f},' f'{ly * COORD_SCALE:.2f}",') 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",{legend_pos} label=<' '' '' '' '' '' '' '' '" "
durchgezogenMaterialfluss
gestrichelt blauKreisel-Umlauf, Richtung aus Drehrichtung ' f'({ROTATION_CW} = {ROTATION_FLOW[ROTATION_CW]}, ' f'{ROTATION_CCW} = {ROTATION_FLOW[ROTATION_CCW]})
gestrichelt gelbRichtung nicht bestimmbar
doppelte UmrandungGruppe paralleler Gefaellestrecken (gleicher Zu-/Ablauf, ' "gleiche Steigung)
>];") add(" }") add("}") return "\n".join(out) + "\n" # --------------------------------------------------------------------------- # DOT des Bahnzusammenhang-Graphen erzeugen (--track) # --------------------------------------------------------------------------- TRACK_NODE_STYLES: dict[str, dict[str, str]] = { "Separator": dict(fillcolor="#fff2cc", fontcolor="#7f6000", color="#bf8f00"), # Scanner ist keine eigene TrackNode-Art (siehe TRACK_KINDS) und taucht im # --track-DOT nie unter diesem Schluessel auf - der Eintrag existiert nur, # damit er sich farblich vom DEFAULT_NODE_STYLE-Grau abheben kann, sobald # ein Konsument ihn braucht. "Scanner": dict(fillcolor="#d9d2e9", fontcolor="#351c75", color="#674ea7"), "Ausschleus": dict(fillcolor="#f9cb9c", fontcolor="#b45f06", color="#e69138"), "Einschleus": dict(fillcolor="#cfe2f3", fontcolor="#1155cc", color="#3d85c6"), "Weiche": dict(fillcolor="#ead1dc", fontcolor="#741b47", color="#a64d79"), "Beladung": dict(fillcolor="#d4f0d4", fontcolor="#1e5b1e", color="#338833"), "Entladung": dict(fillcolor="#f4d4d4", fontcolor="#7a1f1f", color="#cc3333"), SYMBOL_ENTRY: dict(fillcolor="#d4f0d4", fontcolor="#1e5b1e", color="#338833"), SYMBOL_EXIT: dict(fillcolor="#f4d4d4", fontcolor="#7a1f1f", color="#cc3333"), "Bahn": dict(fillcolor="#f2f2f2", fontcolor="#666666", color="#999999"), } TRACK_EDGE_ATTRS: dict[str, str] = { # Schritte innerhalb der TrackIds-Kette eines Containers (build_track_graph): "Bahn": '[color="#2f5597", tooltip="Weiterfahrt auf derselben Bahn (Strecke/Gefaelle)"]', "Umlauf": '[color="#8ea9db", tooltip="Kreisel-Umlauf (Reihenfolge in Drehrichtung)"]', "unbestimmt": '[color="#bf8f00", style=dashed, dir=both, tooltip="Richtung unbestimmt"]', "Symbol": '[color="#999999", style=dotted, tooltip="offenes Bahnende ohne BTMT-Station"]', "Station": '[color="#338833", tooltip="Anbindung BTMT Be-/Entladung"]', } def _track_node_label(node: TrackNode) -> str: if node.placeholder: lines = [f"Bahn {node.node_id}", "Reihenfolge/Position offen (Teil B)"] if node.note: lines.append(node.note) return _dot_label(lines) if node.element is None: return _dot_label([node.kind]) el = node.element lines = [f"{el.name or node.kind} [{node.node_id}]", node.kind] if node.note: lines.append(node.note) return _dot_label(lines) def _track_node_html_label(node: TrackNode) -> str: """HTML-like Label mit rotem Punkt unten mittig fuer einen Separator mit Scanner-Zuordnung (node.scanner_zuordnung, siehe build_track_graph Schritt 5) - der Scanner selbst bekommt weiterhin keinen eigenen Knoten.""" el = node.element lines = [f"{el.name or node.kind} [{node.node_id}]" if el else node.kind, node.kind] if node.note: lines.append(node.note) rows = "".join(f"{_xml_escape(line)}" for line in lines if line) dot_row = '●' return ( f'<' f'{rows}{dot_row}
>' ) def render_track_dot(track: TrackGraph) -> str: """DOT des Bahnzusammenhang-Graphen (--track); siehe build_track_graph() und doc/track.md. Gerendert ueber dieselben graph_to_svg/dot_to_svg wie render_dot().""" out: list[str] = [] add = out.append add("// Automatisch erzeugt von lib/material_flow.py (--track) - nicht manuell aendern.") add(f"// Quelle: {track.source}") add("digraph Bahnzusammenhang {") add(' graph [rankdir=LR, splines=spline, nodesep=0.2, ranksep=1.1,') add(' fontname="Segoe UI", fontsize=11, labelloc="t",') add(f' label="Bahnzusammenhang - {_dot_escape(track.source)}"];') add(' node [shape=box, style="rounded,filled", fontname="Segoe UI", fontsize=8];') add(' edge [fontname="Segoe UI", fontsize=7, color="#2f5597", penwidth=1.1, arrowsize=0.7];') add("") for node_id in sorted(track.nodes): node = track.nodes[node_id] style = TRACK_NODE_STYLES.get(node.kind, DEFAULT_NODE_STYLE) attrs = ", ".join(f'{key}="{value}"' for key, value in style.items()) extra = ', style="rounded,dashed,filled"' if node.placeholder else "" if node.kind == "Weiche": extra += ", shape=diamond" elif node.kind in (SYMBOL_ENTRY, SYMBOL_EXIT, "Beladung", "Entladung"): extra += ", shape=ellipse" if node.kind == "Separator" and node.scanner_zuordnung: label_attr = f"label={_track_node_html_label(node)}" else: label_attr = f'label="{_track_node_label(node)}"' add(f' "{node_id}" [{label_attr}, {attrs}{extra}];') add("") for edge in track.edges: attrs = TRACK_EDGE_ATTRS.get(edge.note, "[]") add(f' "{edge.src}" -> "{edge.dst}" {attrs};') add("") add(" subgraph cluster_legende {") add(' label="Legende"; style="rounded"; color="#a6a6a6";') add(' fontname="Segoe UI"; fontsize=10; fontcolor="#404040";') add(' legende [shape=plaintext, style="", fillcolor="none", label=<') add('') add('') add('' '') add('' '') add('' '') add('' '') add('' '') add('' '') add("
RauteWeiche (Kreuzung zweier Kreisel)
blauWeiterfahrt auf derselben Bahn (Strecke/Gefaelle-Kette)
hellblauKreisel-Umlauf (Reihenfolge in Drehrichtung)
gruenAnbindung BTMT Be-/Entladung
gestrichelter RandBahn-Platzhalter - Reihenfolge/Position offen (Teil B)
gestrichelt gelbRichtung unbestimmt (Element nicht in einer TrackIds-Kette)
●Separator hat einen zugeordneten Scanner (kein eigener Knoten)
>];") add(" }") add("}") return "\n".join(out) + "\n" # --------------------------------------------------------------------------- # connect.ini (Item-/Flussebene) erzeugen # --------------------------------------------------------------------------- def _connect_kind(edge: Edge) -> str: """Kantenart fuer connect.ini.""" if edge.kind == EDGE_CIRCLE: return "umlauf" if edge.note == "Weiche": return "weiche" if edge.kind == EDGE_UNRESOLVED: return "unbestimmt" if edge.note in ("Einschleusung", "Ausschleusung"): return edge.note.lower() return "normal" def render_connect_ini(graph: Graph, section: str = "Anlage", overrides=None) -> str: """ Materialfluss-Topologie als connect.ini (Item-/Flussebene). Zwischenstand fuer die TRO-Ableitung: Knoten = Flussobjekte (Kreisel-Bahnen, Strecke, Gefaellestrecke), Kanten = gerichteter Materialfluss. Von Hand editierbar. Aufbau angelehnt an doc/TRO_Katalog/connect.ini. `overrides` (aus cfg/tro_overrides.ini) wird als eigener Abschnitt `[.manual]` angehaengt. Die Datei hier wird bei jedem Lauf neu geschrieben - von Hand eingetragene Weichen waeren sonst weg. Der Abschnitt beschreibt bewusst nur, WAS vor Ort bestaetigt wurde; er erfindet keine Flusskanten, denn welche Kante zur Weiche gehoert, ist aus dem Layout nicht ableitbar. """ out: list[str] = [] add = out.append add("; ============================================================") add("; connect.ini - Materialfluss-Topologie (Item-/Flussebene)") add(f"; Automatisch erzeugt von lib/material_flow.py aus '{graph.source}'.") add("; Zwischenstand fuer die TRO-Ableitung; von Hand editierbar.") add(";") add("; [.nodes] = | ") add("; [.connections] cNNN = -> | | ") add("; kind: normal | umlauf | weiche | einschleusung | ausschleusung | unbestimmt") add("; [.externals] eNNN = EXTERN -> | (Zulauf/Beladung)") add("; eNNN = -> EXTERN | (Ablauf/Entladung)") add("; [.manual] vor Ort bestaetigte Angaben aus cfg/tro_overrides.ini") add("; ============================================================") add("") add(f"[{section}.nodes]") for node_id in sorted(graph.nodes): node = graph.nodes[node_id] el = node.element if node.lane: name = _merkmal(el.merkmale, "Name") or el.name comment = f"{name} (Bahn {node.lane})" else: comment = el.name or el.kind add(f"{node_id} = {node.kind} | {comment}") add("") add(f"[{section}.connections]") for i, edge in enumerate(sorted(graph.edges, key=lambda e: (e.src, e.dst)), start=1): add(f"c{i:03d} = {edge.src} -> {edge.dst} | | {_connect_kind(edge)}") add("") add(f"[{section}.externals]") stations = [n for n in graph.nodes.values() if n.element.is_station] if stations: for j, node in enumerate(sorted(stations, key=lambda n: n.node_id), start=1): if node.element.is_entry: add(f"e{j:03d} = EXTERN -> {node.node_id} | {node.element.name}") else: add(f"e{j:03d} = {node.node_id} -> EXTERN | {node.element.name}") else: add("; Noch keine BTMT Be-/Entladestation im Layout - Ein-/Ausgang offen.") add("; Kandidaten (offene Enden aus dem Materialfluss):") incoming = {e.dst for e in graph.edges} outgoing = {e.src for e in graph.edges} for node_id in sorted(graph.nodes): node = graph.nodes[node_id] if node.element.is_station: continue if node_id not in incoming: add(f"; Beladung? EXTERN -> {node_id} ({node.kind}, keine Zufuhr)") if node_id not in outgoing: add(f"; Entladung? {node_id} -> EXTERN ({node.kind}, kein Abfluss)") add("") if overrides is not None and (overrides.tros or overrides.splits or overrides.open_points): add(f"[{section}.manual]") add("; Vor Ort bestaetigt, im CSV-Export nicht enthalten.") add(f"; Quelle: {overrides.source}") add("; Diese Zeilen werden aus der Override-Datei erzeugt - nicht hier") add("; editieren, sonst sind sie beim naechsten Lauf wieder weg.") for section_key, override in sorted(overrides.tros.items()): if not override.type_name: continue wie = f"Separator {override.key}" if override.by_separator else override.key add(f"{section_key} = {override.type_name} | {wie} | {override.reason}") for spec in overrides.splits: add(f"{spec.name} = NEU {spec.type_name} " f"| Separator {', '.join(spec.separators)}" + (f" + Scanner {', '.join(spec.scanners)}" if spec.scanners else "") + (f" | Bahn {spec.host}" if spec.host else "") + f" | {spec.reason}") for point in overrides.open_points: add(f"{point.name} = OFFEN {point.expect or '?'} " f"| x={point.x:.2f} y={point.y:.2f} " f"| heute Teil von {point.belongs_to or '-'} " f"| {point.note}") add("") return "\n".join(out) + "\n" # --------------------------------------------------------------------------- # 2D-Draufsicht der Boundingboxen (--show-bbs) # --------------------------------------------------------------------------- # Farben je Objektart fuer die Boundingbox-Draufsicht (Fuellung, Rahmen, Text). # Eigene Palette statt NODE_STYLES, weil hier auch die Anbauteile (Separator, # Scanner, Ein-/Ausschleuselement, Weiche) eine eigene Farbe brauchen. BBS_STYLES: dict[str, dict[str, str]] = { "Kreisel": dict(fill="#2f5597", stroke="#1f3864", text="#ffffff"), "Gefaellestrecke": dict(fill="#fbe5d6", stroke="#c55a11", text="#833c00"), "Strecke": dict(fill="#e2f0d9", stroke="#548235", text="#375623"), "Beladung": dict(fill="#d4f0d4", stroke="#338833", text="#1e5b1e"), "Entladung": dict(fill="#f4d4d4", stroke="#cc3333", text="#7a1f1f"), "Separator": dict(fill="#fff2cc", stroke="#bf8f00", text="#7f6000"), "Scanner": dict(fill="#d9d2e9", stroke="#674ea7", text="#351c75"), "Einschleus": dict(fill="#cfe2f3", stroke="#3d85c6", text="#1155cc"), "Ausschleus": dict(fill="#f9cb9c", stroke="#e69138", text="#b45f06"), "Weiche": dict(fill="#ead1dc", stroke="#a64d79", text="#741b47"), } BBS_DEFAULT_STYLE = dict(fill="#f2f2f2", stroke="#808080", text="#404040") # Rand um die Anlage herum und minimale Schriftgroesse, jeweils in mm. BBS_MARGIN_MM = 500.0 BBS_MIN_FONT_MM = 40.0 # Darstellung der Anschlusspunkte K1-K4 (--show-kx): Groesse des Kreuzes und # der Beschriftung in mm, Verbindungslinie K1->K2->K3->K4 in Blau. KX_CROSS_MM = 120.0 KX_FONT_MM = 90.0 KX_CROSS_COLOR = "#c00000" KX_LINK_COLOR = "#1f5fd0" # Physisches Papierformat der SVG-Ausgabe: DIN A3, Quer- oder Hochformat je # nach Seitenverhaeltnis der Anlage. Die viewBox bleibt in Anlage-mm - nur # width/height (das gedruckte Format) werden auf A3 skaliert. DINA3_LANDSCAPE_MM = (420.0, 297.0) DINA3_PORTRAIT_MM = (297.0, 420.0) def _xml_escape(text: str) -> str: return ( text.replace("&", "&").replace("<", "<").replace(">", ">") .replace('"', """) ) def render_bbs_svg( elements: list[Element], source: str, warnings: list[str], show_kx: bool = False, ) -> str: """ 2D-Draufsicht der ganzen Szene: ein achsparalleles Rechteck je CSV-Zeile, aus 'Position' (Mittelpunkt) und 'Boundingbox' (Breite/Tiefe in X/Y - die Z-Ausdehnung ist fuer eine Draufsicht ohne Bedeutung), mit der Bezeichnung im Zentrum. Keine Rotation - reine Lageuebersicht auf Basis der Rohdaten, kein Ersatz fuer die CAD-Zeichnung. Anlage-Y waechst nach oben (wie im CSV/DXF), SVG-Y nach unten - deshalb wird Y beim Zeichnen gespiegelt, damit die Draufsicht nicht auf dem Kopf steht. Die viewBox bleibt in Anlage-mm (Koordinaten, Schrift, Strichstaerke bleiben so zueinander proportional); das gedruckte Format (width/height) wird auf DIN A3 (Quer- oder Hochformat je nach Seitenverhaeltnis) skaliert - sonst waere die SVG-Datei bei einer 25 m breiten Anlage 25 m breit. show_kx zeichnet zusaetzlich je Objekt die Anschlusspunkte K1-K4 (rotes Kreuz + kleine Beschriftung) und verbindet sie in der Reihenfolge K1->K2->K3->K4 mit einer blauen Linie. Ein Kreisel wird - anders als jedes andere Objekt - nicht als ein Rechteck, sondern als seine beiden Bahnen L/R gezeichnet: dieselbe massstabsgetreue Aufteilung (kurze Achse hälftig, Zuordnung-Mehrheits- entscheid fuer L/R), siehe _kreisel_lane_rects(). Die K1-K4-Anschlusspunkte (real am Kreisel als Ganzem haengend) werden trotzdem nur einmal gezeichnet, nicht je Bahn doppelt. """ items: list[tuple[Element, float, float, float, float, str | None]] = [] for element in elements: if element.position is None or element.bbox is None: warnings.append( f"{element.describe()}: keine Position/Boundingbox - in --show-bbs nicht gezeichnet" ) continue px, py = element.x, element.y bw, bd = element.bbox_x, element.bbox_y if not bw or not bd or bw <= 0 or bd <= 0: warnings.append( f"{element.describe()}: Boundingbox {bw}, {bd} mm ungueltig - in --show-bbs nicht gezeichnet" ) continue if element.is_circle: for lane, (x0, y0, x1, y1) in sorted(_kreisel_lane_rects(element, elements).items()): items.append((element, (x0 + x1) / 2, (y0 + y1) / 2, x1 - x0, y1 - y0, lane)) continue items.append((element, px, py, bw, bd, None)) if not items: raise ValueError("keine Objekte mit Position und Boundingbox gefunden") min_x = min(item[1] - item[3] / 2 for item in items) - BBS_MARGIN_MM max_x = max(item[1] + item[3] / 2 for item in items) + BBS_MARGIN_MM min_y = min(item[2] - item[4] / 2 for item in items) - BBS_MARGIN_MM max_y = max(item[2] + item[4] / 2 for item in items) + BBS_MARGIN_MM width, height = max_x - min_x, max_y - min_y # Groesste Objekte zuerst zeichnen, damit kleine Anbauteile (Separator, # Scanner, ...) sichtbar obenauf liegen statt unter einer Strecke/einem # Kreisel zu verschwinden. items.sort(key=lambda item: item[3] * item[4], reverse=True) page_w, page_h = DINA3_LANDSCAPE_MM if width >= height else DINA3_PORTRAIT_MM out: list[str] = [] add = out.append add('') add(f'') add(" ") add(f" ") add(f' ') for element, px, py, bw, bd, lane in items: style = BBS_STYLES.get(element.kind, BBS_DEFAULT_STYLE) x = (px - bw / 2) - min_x y = max_y - (py + bd / 2) name = f"{element.name or element.teile_id} Bahn {lane}" if lane else (element.name or element.teile_id) font_size = max(BBS_MIN_FONT_MM, min(bw, bd) * 0.28) add(f' ') add(f' {_xml_escape(name)}') # Anschlusspunkte K1-K4 zuletzt, damit sie ueber den Rechtecken liegen. if show_kx: half = KX_CROSS_MM / 2 seen_kx: set[str] = set() for element, *_rest in items: # Ein per Bahn L/R aufgeteilter Kreisel steht zweimal in items, # seine K1-K4 haengen aber am Kreisel als Ganzem - nur einmal zeichnen. if element.teile_id in seen_kx: continue seen_kx.add(element.teile_id) # Anlage- in viewBox-Koordinaten (X verschoben, Y gespiegelt). pts = [(label, kx - min_x, max_y - ky) for label, (kx, ky) in element.kpoints] # Verbindungslinie K1->K2->K3->K4 (nur bei mindestens zwei Punkten). for (_, x1, y1), (_, x2, y2) in zip(pts, pts[1:]): add(f' ') for label, sx, sy in pts: add(f' ') add(f' ') add(f' {label}') add("") return "\n".join(out) + "\n" # --------------------------------------------------------------------------- # Kreisel-Geometrie: Aufteilung in Bahn L/R # --------------------------------------------------------------------------- def _kreisel_lane_rects( kreisel: Element, fine: list[Element] ) -> dict[str, tuple[float, float, float, float]]: """ (x0, y0, x1, y1) je Bahn (LANE_LEFT/LANE_RIGHT) in Anlage-mm: die Kreisel- Boundingbox wird entlang ihrer kurzen Achse haelftig geteilt (die beiden Bahnen laufen parallel zur langen Achse). Welche Haelfte L bzw. R ist, wird per Mehrheitsentscheid aus der 'Zuordnung' (…-L/…-R) der uebergebenen Feinelemente abgeleitet; ohne verwertbare Zuordnung faellt die Haelfte mit der kleineren Koordinate auf LANE_RIGHT zurueck (willkuerliche, aber stabile Konvention, wie zuvor in data/gen_chain_svg.py). """ x, y, bw, bh = kreisel.x, kreisel.y, kreisel.bbox_x, kreisel.bbox_y x0, x1 = x - bw / 2, x + bw / 2 y0, y1 = y - bh / 2, y + bh / 2 horizontal = bw >= bh # Bahnen laufen entlang X, Haelften entlang Y gestapelt if horizontal: mid = (y0 + y1) / 2 half_a, half_b = (x0, y0, x1, mid), (x0, mid, x1, y1) else: mid = (x0 + x1) / 2 half_a, half_b = (x0, y0, mid, y1), (mid, y0, x1, y1) votes: dict[str, list[float]] = {LANE_LEFT: [], LANE_RIGHT: []} for el in fine: base, _, lane = el.assignment.partition("-") if base != kreisel.teile_id or lane not in (LANE_LEFT, LANE_RIGHT): continue coord = el.y if horizontal else el.x if coord is not None: votes[lane].append(coord) def avg(values: list[float]) -> float | None: return sum(values) / len(values) if values else None avg_l, avg_r = avg(votes[LANE_LEFT]), avg(votes[LANE_RIGHT]) a_mid = (half_a[1] + half_a[3]) / 2 if horizontal else (half_a[0] + half_a[2]) / 2 if avg_l is None and avg_r is None: lane_a = LANE_RIGHT else: dist_l = abs(avg_l - a_mid) if avg_l is not None else float("inf") dist_r = abs(avg_r - a_mid) if avg_r is not None else float("inf") lane_a = LANE_LEFT if dist_l < dist_r else LANE_RIGHT lane_b = LANE_RIGHT if lane_a == LANE_LEFT else LANE_LEFT return {lane_a: half_a, lane_b: half_b} # --------------------------------------------------------------------------- # Chain-/Lageplan-SVG der groben Objekte erzeugen (--chain-graph) # --------------------------------------------------------------------------- # # Schritt 1 des neu aufgebauten Materialfluss-Bildes: nur die groben Objekte - # Kreisel (je Bahn L/R) und Strecke/Gefaellestrecke (VF/GF) - als massstabs- # getreue Rechtecke an ihrer echten Position/Boundingbox, verbunden durch die # bereits von build_graph() aufgeloesten Materialfluss-Kanten (graph.edges). # Keine Anbauteile (Separator, Scanner, Weiche, ...) und keine Ein-/ # Ausgabestationen - die kommen in einem spaeteren Schritt dazu. CHAIN_MAX_W = 1400.0 CHAIN_MAX_H = 850.0 CHAIN_MARGIN = 70.0 CHAIN_TITLE_H = 60.0 CHAIN_EDGE_COLOR = { EDGE_FLOW: "#2f5597", EDGE_CIRCLE: "#8ea9db", EDGE_UNRESOLVED: "#bf8f00", } def _chain_marker_id(color: str) -> str: return "chainArrow_" + color.lstrip("#") def render_chain_svg(elements: list[Element], graph: Graph, source: str) -> str: """ Massstabsgetreuer Chain-/Lageplan der groben Objekte (--chain-graph): jeder Kreisel als seine beiden Bahnen L/R (_kreisel_lane_rects(), Zuordnung aus allen Elementen fuer den Mehrheitsentscheid - wie --show-bbs) und jede Strecke/Gefaellestrecke als ein Rechteck an ihrer echten Position/ Boundingbox. Kanten kommen direkt aus dem schon aufgeloesten groben Materialfluss-Graphen (graph.edges) - eine Kante, deren Knoten hier nicht gezeichnet wird (z. B. eine BTMT-Station), wird schlicht uebersprungen. """ rects: dict[str, tuple[float, float, float, float]] = {} lane_info: dict[str, tuple[Element, str]] = {} strecke_labels: dict[str, Element] = {} for element in elements: if element.position is None or element.bbox is None: continue if element.is_circle: for lane, rect in _kreisel_lane_rects(element, elements).items(): node_id = f"{element.teile_id}-{lane}" rects[node_id] = rect lane_info[node_id] = (element, lane) elif element.is_transport: rects[element.teile_id] = ( element.x - element.bbox_x / 2, element.y - element.bbox_y / 2, element.x + element.bbox_x / 2, element.y + element.bbox_y / 2, ) strecke_labels[element.teile_id] = element if not rects: raise ValueError("keine Kreisel/Strecken mit Position/Boundingbox gefunden") all_x = [v for r in rects.values() for v in (r[0], r[2])] all_y = [v for r in rects.values() for v in (r[1], r[3])] min_x, max_x = min(all_x), max(all_x) min_y, max_y = min(all_y), max(all_y) real_w, real_h = max(max_x - min_x, 1.0), max(max_y - min_y, 1.0) scale = min(CHAIN_MAX_W / real_w, CHAIN_MAX_H / real_h) def sx(rx: float) -> float: return CHAIN_MARGIN + (rx - min_x) * scale def sy(ry: float) -> float: return CHAIN_TITLE_H + (max_y - ry) * scale draw_w, draw_h = real_w * scale, real_h * scale canvas_w = CHAIN_MARGIN * 2 + draw_w canvas_h = CHAIN_TITLE_H + draw_h + CHAIN_MARGIN def center(node_id: str) -> tuple[float, float]: x0, y0, x1, y1 = rects[node_id] return ((x0 + x1) / 2, (y0 + y1) / 2) def anchor(node_id: str, other: tuple[float, float]) -> tuple[float, float]: """Punkt auf dem Rand von node_id, der am naechsten zu 'other' liegt - auch wenn 'other' (der Mittelpunkt des Partners) selbst innerhalb des Rechtecks liegt (schmale Kreisel-Bahn ueber die ganze Anlage).""" x0, y0, x1, y1 = rects[node_id] ox, oy = other cx, cy = min(max(ox, x0), x1), min(max(oy, y0), y1) if (cx, cy) != (ox, oy): return (cx, cy) d_left, d_right = ox - x0, x1 - ox d_bottom, d_top = oy - y0, y1 - oy nearest = min(d_left, d_right, d_bottom, d_top) if nearest == d_left: return (x0, oy) if nearest == d_right: return (x1, oy) if nearest == d_bottom: return (ox, y0) return (ox, y1) edges = [e for e in graph.edges if e.src in rects and e.dst in rects] # Mindestlaenge eines Pfeils in Bildschirm-Pixeln: zwei Kreisel-Bahnen # grenzen direkt aneinander (siehe _kreisel_lane_rects) - ihr Umlauf-Pfeil # laeuft dann exakt auf der gemeinsamen Grenze und wuerde ohne Nachschub # unsichtbar lang Null sein. MIN_ARROW_PX = 14.0 edge_svg: list[str] = [] for edge in edges: x1, y1 = anchor(edge.src, center(edge.dst)) x2, y2 = anchor(edge.dst, center(edge.src)) px1, py1, px2, py2 = sx(x1), sy(y1), sx(x2), sy(y2) if (px2 - px1) ** 2 + (py2 - py1) ** 2 < (MIN_ARROW_PX / 2) ** 2: src_cx, src_cy = center(edge.src) dst_cx, dst_cy = center(edge.dst) dx, dy = sx(dst_cx) - sx(src_cx), sy(dst_cy) - sy(src_cy) length = max((dx ** 2 + dy ** 2) ** 0.5, 1e-6) ux, uy = dx / length * (MIN_ARROW_PX / 2), dy / length * (MIN_ARROW_PX / 2) mx, my = (px1 + px2) / 2, (py1 + py2) / 2 px1, py1, px2, py2 = mx - ux, my - uy, mx + ux, my + uy color = CHAIN_EDGE_COLOR.get(edge.kind, CHAIN_EDGE_COLOR[EDGE_FLOW]) dash = ' stroke-dasharray="4 3"' if edge.kind == EDGE_UNRESOLVED or edge.note == "unbestimmt" else "" title = f"{edge.src} -> {edge.dst}" + (f" ({_xml_escape(edge.note)})" if edge.note else "") edge_svg.append( f'{title}' ) box_svg: list[str] = [] for node_id, (x0, y0, x1, y1) in rects.items(): if node_id in lane_info: kreisel, lane = lane_info[node_id] style = NODE_STYLES.get(kreisel.kind, DEFAULT_NODE_STYLE) label = f"{kreisel.name or kreisel.teile_id} Bahn {lane}" else: strecke = strecke_labels[node_id] style = NODE_STYLES.get(strecke.kind, DEFAULT_NODE_STYLE) label = f"{strecke.name or node_id} [{node_id}]" box_svg.append( f'' ) box_svg.append( f'{_xml_escape(label)}' ) heading = f"{Path(source).stem} - Materialfluss (grobe Objekte)" out: list[str] = [] add = out.append add('') add(f'') add(f' ') add(f'') add(f' {_xml_escape(heading)}') add(f' ') add(' ') for color in sorted(set(CHAIN_EDGE_COLOR.values())): add(f' ' f'') add(' ') add(f' {_xml_escape(heading)}') add(f' ' f'Massstabsgetreu • Quelle: {_xml_escape(source)}') out.extend(box_svg) out.extend(edge_svg) add("") return "\n".join(out) + "\n" # --------------------------------------------------------------------------- # SVG erzeugen # --------------------------------------------------------------------------- def find_dot() -> str | None: env = os.environ.get("GRAPHVIZ_DOT") if env: return env if Path(env).is_file() else shutil.which(env) return shutil.which("dot") def find_engine(engine: str) -> str | None: """Layout-Programm von Graphviz suchen ('dot', 'neato', ...).""" if engine == "dot": return find_dot() return shutil.which(engine) def graph_to_svg( dot_file: Path, svg_file: Path, engine: str = "dot", no_op: int = 0 ) -> None: """ SVG aus der .dot-Datei erzeugen. engine Layout-Programm ('dot' fuer die berechnete Anordnung, 'neato' fuer vorgegebene Koordinaten) no_op nur fuer neato: 1 oder 2 setzt den No-op-Schalter (-n / -n2), dann werden die pos-Attribute der Knoten als Position uebernommen und nur die Kanten berechnet. Alle Knoten brauchen dann ein pos. """ executable = find_engine(engine) if not executable: raise RuntimeError( f"Graphviz '{engine}' nicht gefunden. Bitte Graphviz installieren " f"(https://graphviz.org/download/)" + (" oder GRAPHVIZ_DOT auf die dot-Programmdatei setzen." if engine == "dot" else ".") ) command = [executable] if no_op: command.append(f"-n{no_op}") command += ["-Tsvg", str(dot_file), "-o", str(svg_file)] result = subprocess.run(command, capture_output=True, text=True) if result.returncode != 0: raise RuntimeError( f"'{engine}' ist mit Code {result.returncode} fehlgeschlagen:\n" f"{(result.stderr or result.stdout).strip()}" ) # 'neato -n' gibt harmlose Warnungen auf stderr aus (z. B. sich beruehrende # Knoten -> gerade Kanten). Nur dann scheitern, wenn wirklich keine SVG # entstanden ist; sonst ist die Datei gueltig und die Warnung unkritisch. if not svg_file.is_file() or svg_file.stat().st_size == 0: raise RuntimeError( f"'{engine}' hat keine SVG erzeugt: {result.stderr.strip() or '(keine Meldung)'}" ) def dot_to_svg(dot_file: Path, svg_file: Path) -> None: """SVG mit der berechneten Anordnung von 'dot' erzeugen.""" graph_to_svg(dot_file, svg_file, engine="dot") # --------------------------------------------------------------------------- # Plausibilitaet # --------------------------------------------------------------------------- def check_elements(graph: Graph, elements: list[Element]) -> list[str]: """ Merkmal-Stueckzahlen gegen die tatsaechlich vorhandenen Anbauteile pruefen. Rein objektbezogen - alles, was TROs betrifft, liegt in lib/tro_flow.py. """ findings: list[str] = [] counted: dict[str, dict[str, int]] = {} for node_id, node in graph.nodes.items(): for kind, ids in node.devices.items(): counted.setdefault(node_id, {})[kind] = len(ids) # Widerspruch zwischen Hoehenangabe und Antriebfahrtrichtung for element in elements: if not element.is_transport: continue rise, drive = element.rise, element.drive_dir.lower() if rise is None or rise == 0 or not drive: continue stated_up = drive.startswith("auf") if stated_up != (rise > 0): findings.append( f"{element.kind} {element.teile_id} '{element.name}': " f"Antriebfahrtrichtung '{element.drive_dir}' passt nicht zu " f"Hoehe_Von {element.height_from:.0f} -> Hoehe_Bis " f"{element.height_to:.0f} mm ({rise:+.0f}) - die Hoehenangabe gilt" ) for element in elements: if not (element.is_transport or element.is_circle): continue node_ids = ( [element.teile_id] if element.is_transport else [f"{element.teile_id}-{LANE_LEFT}", f"{element.teile_id}-{LANE_RIGHT}"] ) for label, keys in ( ("Separator", ("Anzahl_Separator", "Anzahl der Separatoren")), ("Scanner", ("Anzahl_Scanner", "Anzahl der Scanner")), ): expected = _as_float(_merkmal(element.merkmale, *keys)) if expected is None: continue actual = sum(counted.get(n, {}).get(label, 0) for n in node_ids) if int(expected) != actual: findings.append( f"{element.kind} {element.teile_id} '{element.name}': Merkmal " f"'{keys[0] if keys[0] in element.merkmale else keys[1]}' = " f"{int(expected)}, tatsaechlich {actual} {label}" ) findings.sort() return findings # --------------------------------------------------------------------------- # Dokumentation (Markdown) # --------------------------------------------------------------------------- GENERATED_MARKER = "" def render_doc( graph: Graph, findings: list[str], csv_file: Path, dot_file: Path, svg_file: Path | None, ) -> str: """Dokumentation des Materialflusses (nur Objekte, keine TROs).""" out: list[str] = [GENERATED_MARKER] add = out.append add(f"# Materialfluss - {csv_file.name}") add("") add(f"**Erzeugt:** {datetime.now().strftime('%Y-%m-%d %H:%M')} von " f"`lib/material_flow.py` ") add(f"**Quelle:** `{csv_file}`") add("") add("> Materialfluss der mechanischen Objekte. Die daraus abgeleiteten " "Steuerungsobjekte (TROs) dokumentiert `lib/tro_flow.py --doc`.") add("") kinds: dict[str, int] = {} for node in graph.nodes.values(): kinds[node.kind] = kinds.get(node.kind, 0) + 1 devices: dict[str, int] = {} for node in graph.nodes.values(): for kind, ids in node.devices.items(): devices[kind] = devices.get(kind, 0) + len(ids) flow = sum(1 for e in graph.edges if e.kind == EDGE_FLOW) circle = sum(1 for e in graph.edges if e.kind == EDGE_CIRCLE) open_dir = sum(1 for e in graph.edges if e.kind == EDGE_UNRESOLVED) add("## 1. Kurzfassung") add("") add("| | |") add("|---|---|") add(f"| Flussknoten | {len(graph.nodes)} " f"({', '.join(f'{k}: {v}' for k, v in sorted(kinds.items()))}) |") add(f"| Flusskanten | {flow} gerichtet, {circle} Kreisel-Umlauf, " f"{open_dir} Richtung unbestimmt |") add(f"| Anbauteile | " f"{', '.join(f'{k}: {v}' for k, v in sorted(devices.items())) or '-'} |") add(f"| Plausibilitaetshinweise | {len(findings)} |") add("") add(f"Graph: [`{dot_file.name}`]({dot_file.name})" + (f" - [`{svg_file.name}`]({svg_file.name})" if svg_file else "")) add("") add("Ein **Kreisel** ist ein Umlauf und wird als zwei Knoten modelliert - eine " "Bahn je Fahrtrichtung (`-L` links, `-R` rechts), die zu einem Kreis " "verbunden sind.") add("") add("## 2. Anlagenobjekte") add("") add("| Knoten | Art | Bezeichnung | Planquadrat | Hoehe [m] | Separatoren | Scanner |") add("|---|---|---|---|---|---|---|") for node_id in sorted(graph.nodes): node = graph.nodes[node_id] height = f"{node.height:.3f}" if node.height is not None else "-" name = node.element.name logical = _merkmal(node.element.merkmale, "Name") if logical and str(logical).strip() and str(logical).strip() != name: name = f"{name} / {str(logical).strip()}" add(f"| `{node_id}` | {node.kind} | {name}" f"{f' (Bahn {node.lane})' if node.lane else ''} | " f"{node.element.plan_square or '-'} | {height} | " f"{', '.join(sorted(node.devices.get('Separator', []))) or '-'} | " f"{', '.join(sorted(node.devices.get('Scanner', []))) or '-'} |") add("") add("Bezeichnung: `Bezeichnung` aus dem CSV; wo das Merkmal `Name` abweicht, " "steht es dahinter (die beiden Nummerierungen laufen nicht zwangslaeufig " "gleich).") add("") add("Hoehe: bei Kreiseln aus dem Merkmal `Höhe in m`, bei Gefaellestrecke und " "Strecke aus `Montagehoehe_m` - unveraendert aus dem CSV uebernommen. Fuer " "die Richtungsableitung werden nur die Hoehen der **Nachbarn** verglichen, " "nie die des Transportobjekts selbst.") add("") if graph.unassigned: add("**Anbauteile ohne Zuordnung**") add("") for element in graph.unassigned: add(f"- {element.kind} {element.teile_id} '{element.name}' " f"(Zuordnung: `{element.assignment or '-'}`)") add("") add("## 3. Verbindungen") add("") add("| von | nach | Art |") add("|---|---|---|") kind_text = {EDGE_FLOW: "Materialfluss", EDGE_CIRCLE: "Kreisel-Umlauf", EDGE_UNRESOLVED: "Richtung unbestimmt"} for edge in graph.edges: note = f" ({edge.note})" if edge.note else "" add(f"| `{edge.src}` | `{edge.dst}` | " f"{kind_text.get(edge.kind, edge.kind)}{note} |") add("") add(f"Die Richtung innerhalb eines Kreisels kommt aus dem Merkmal " f"`Drehrichtung`: **{ROTATION_CW}** (im Uhrzeigersinn) laesst das Material " f"{ROTATION_FLOW[ROTATION_CW]} laufen, **{ROTATION_CCW}** (gegen den " f"Uhrzeigersinn) {ROTATION_FLOW[ROTATION_CCW]}.") add("") add("## 4. Plausibilitaetshinweise") add("") if findings: add("Abweichungen zwischen den Merkmalen des CSV-Exports und den " "tatsaechlich vorhandenen Objekten.") add("") for finding in findings: add(f"- {finding}") else: add("_Keine Abweichungen gefunden._") add("") add("## 5. Warnungen des Laufs") add("") if graph.warnings: for warning in graph.warnings: add(f"- {warning}") else: add("_Keine._") add("") return "\n".join(out) + "\n" # --------------------------------------------------------------------------- # Bericht # --------------------------------------------------------------------------- def report( graph: Graph, findings: list[str], csv_file: Path, dot_file: Path, svg_file: Path | None, doc_file: Path | None, stale_svg: Path | None = None, bbs_file: Path | None = None, track_dot_file: Path | None = None, track_svg_file: Path | None = None, chain_file: Path | None = None, ) -> None: kinds: dict[str, int] = {} for node in graph.nodes.values(): kinds[node.kind] = kinds.get(node.kind, 0) + 1 flow = sum(1 for e in graph.edges if e.kind == EDGE_FLOW) circle = sum(1 for e in graph.edges if e.kind == EDGE_CIRCLE) open_dir = sum(1 for e in graph.edges if e.kind == EDGE_UNRESOLVED) devices = 0 for node in graph.nodes.values(): devices += sum(len(ids) for ids in node.devices.values()) print("") print("================================================================") print("MATERIALFLUSS") print("================================================================") print(f"Eingabe = {csv_file}") print(f"Knoten = {len(graph.nodes)}" + (f" ({', '.join(f'{k}: {v}' for k, v in sorted(kinds.items()))})" if kinds else "")) print(f"Kanten = {len(graph.edges)} (Fluss: {flow}, " f"Kreisel-Umlauf: {circle}, unbestimmt: {open_dir})") print(f"Anbauteile = {devices} an Knoten, " f"{len(graph.unassigned)} ohne Zuordnung") 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"BB-Draufsicht = {bbs_file if bbs_file else '- (ohne --show-bbs)'}") print(f"Bahnzusammenhang = {track_dot_file if track_dot_file else '- (ohne --track)'}") if track_dot_file: print(f" SVG = {track_svg_file if track_svg_file else '- (ohne --tosvg)'}") print(f"Chain-Graph = {chain_file if chain_file else '- (ohne --chain-graph)'}") print("================================================================") if stale_svg is not None: print("") print(f"HINWEIS: {stale_svg}") print(" stammt aus einem frueheren Lauf und passt nicht zur neuen " "DOT-Datei.") print(" Mit --tosvg neu erzeugen.") if graph.unassigned: print("") print(f"Ohne Zuordnung ({len(graph.unassigned)}):") for element in graph.unassigned: print(f" - {element.kind} {element.teile_id} '{element.name}'") if findings: print("") print(f"Plausibilitaetshinweise ({len(findings)}):") for finding in findings: print(f" ? {finding}") if graph.warnings: print("") print(f"Warnungen ({len(graph.warnings)}):") for warning in graph.warnings: print(f" ! {warning}") print("") print("TRO-Liste und TRO-Flussdiagramm: bin/tro_flow.bat " f"--file {csv_file.name} --tosvg --doc") print("") # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser( prog="material_flow", description="Erzeugt aus einem CSV-Export der Anlagenobjekte einen " "gerichteten Materialfluss-Graphen der mechanischen Objekte " "(DOT, optional SVG). Die TROs ermittelt tro_flow.py.", formatter_class=argparse.RawDescriptionHelpFormatter, epilog="Beispiele:\n" " material_flow.bat --file mubea.csv\n" " material_flow.bat --file mubea.csv --tosvg\n" " material_flow.bat --file mubea.csv --tosvg --doc\n" " material_flow.bat --file HundM05.csv --show-bbs\n" " material_flow.bat --file mubea.csv --track --tosvg\n" " material_flow.bat --file HundM05.csv --chain-graph\n" "\n" "Exit-Codes:\n" " 0 ok\n" " 1 Eingabe- oder Aufruffehler\n" " 2 SVG konnte nicht erzeugt werden (Graphviz)\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 eine SVG-Datei erzeugen " "(benoetigt Graphviz 'dot').", ) parser.add_argument( "--use-cords", "--use-coords", dest="use_cords", action="store_true", help="Jeden Knoten auf seine Anlagenkoordinate setzen und das SVG von " f"Graphviz 'neato -n' erzeugen statt von 'dot' (lagerichtig, " f"{COORD_SCALE} pt/mm). Nur mit --tosvg wirksam.", ) parser.add_argument( "--doc", action="store_true", help="Dokumentation des Materialflusses als Markdown in %%SKEL_RESULTS%% " "erzeugen (Anlagenobjekte, Verbindungen, Pruefungen).", ) parser.add_argument( "--connect", action="store_true", help="Materialfluss-Topologie zusaetzlich als connect.ini (Item-/Flussebene) " "nach %%SKEL_RESULTS%% schreiben (Knoten, gerichtete Verbindungen, " "Ein-/Ausgang-Kandidaten) - Zwischenstand fuer die TRO-Ableitung.", ) parser.add_argument( "--show-bbs", dest="show_bbs", action="store_true", help="2D-Draufsicht der ganzen Szene als _bbs.svg nach %%SKEL_RESULTS%% " "schreiben: ein achsparalleles Rechteck je CSV-Zeile aus 'Position' " "(Mittelpunkt) und 'Boundingbox' (Breite/Tiefe), mit der Bezeichnung " "im Zentrum, skaliert auf DIN A3. Braucht kein Graphviz.", ) parser.add_argument( "--show-kx", dest="show_kx", action="store_true", help="Nur zusammen mit --show-bbs: je Objekt die Anschlusspunkte K1-K4 als " "kleines Kreuz mit Beschriftung einzeichnen und in der Reihenfolge " "K1->K2->K3->K4 mit einer blauen Linie verbinden.", ) parser.add_argument( "--track", action="store_true", help="Zusaetzlich einen feineren Bahnzusammenhang-Graph erzeugen " "(_track.dot, mit --tosvg auch _track.svg): Separator, " "Ausschleus-/Einschleuselement und Weiche werden zu eigenen Knoten " "(Verzweigung/Zusammenfuehrung/Kreuzung), statt nur wie im normalen " "Materialfluss Kanten/Zuordnungen an Kreisel/Strecke/Gefaellestrecke " "zu liefern. Siehe doc/track.md. Aendert den normalen Materialfluss-" "Output nicht.", ) parser.add_argument( "--chain-graph", dest="chain_graph", action="store_true", help="Massstabsgetreuen Chain-/Lageplan der groben Objekte als " "_chain.svg nach %%SKEL_RESULTS%% schreiben: jeder Kreisel " "(je Bahn L/R) und jede Strecke/Gefaellestrecke (VF/GF) als " "Rechteck an ihrer echten Position/Boundingbox, verbunden durch " "die bereits aufgeloesten Materialfluss-Kanten. Schritt 1 - noch " "ohne Anbauteile (Separator, Scanner, Weiche, ...) oder Ein-/" "Ausgabestationen. Braucht kein Graphviz.", ) return parser.parse_args(argv) def main(argv: list[str] | None = None) -> int: args = parse_args(argv) csv_file = resolve_input(args.file) if csv_file.suffix.lower() != ".csv": print(f"FEHLER: --file erwartet eine CSV-Datei, keine '{csv_file.suffix}'-Datei: " f"{csv_file}", file=sys.stderr) return 1 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 stem = csv_file.stem if args.show_kx and not args.show_bbs: print("FEHLER: --show-kx ist nur zusammen mit --show-bbs moeglich.", file=sys.stderr) return 1 bbs_file = None if args.show_bbs: bbs_target = results / f"{stem}_bbs.svg" try: bbs_target.write_text( render_bbs_svg(elements, csv_file.name, warnings, show_kx=args.show_kx), encoding="utf-8", ) except ValueError as exc: print(f"FEHLER: --show-bbs: {exc}", file=sys.stderr) return 1 bbs_file = bbs_target graph = build_graph(elements, csv_file.name, warnings) if not graph.nodes: print(f"FEHLER: {csv_file} enthaelt keine Flussobjekte " f"({', '.join(TRANSPORT_KINDS + CIRCLE_KINDS)}).", file=sys.stderr) return 1 findings = check_elements(graph, elements) dot_file = results / f"{stem}_material_flow.dot" svg_file = results / f"{stem}_material_flow.svg" doc_target = results / f"{stem}_material_flow.md" # --use-cords braucht an jedem Knoten eine Koordinate if args.use_cords: without = [n.node_id for n in graph.nodes.values() if _node_xy(n) is None] if without: print(f"FEHLER: --use-cords nicht moeglich, {len(without)} Knoten ohne " f"Koordinate: {', '.join(without)}", file=sys.stderr) print(" Ohne pos an jedem Knoten kann 'neato -n' nicht zeichnen.", file=sys.stderr) return 1 existing_svg = svg_file if svg_file.is_file() else None dot_file.write_text(render_dot(graph, use_coords=args.use_cords), encoding="utf-8") svg_error: str | None = None if args.tosvg: try: if args.use_cords: graph_to_svg(dot_file, svg_file, engine="neato", no_op=1) else: 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(graph, findings, csv_file, dot_file, created_svg), encoding="utf-8", ) doc_file = doc_target connect_file = None if args.connect: # Spaet importiert: tro_overrides zieht tro_catalog nach, das # material_flow sonst nicht braucht. from tro_overrides import load_overrides overrides = load_overrides(csv_file.name, env_dir("SKEL_CFG", "cfg")) connect_target = results / f"{stem}_connect.ini" connect_target.write_text( render_connect_ini(graph, section=stem, overrides=overrides), encoding="utf-8", ) connect_file = connect_target by_id = {element.teile_id: element for element in elements} track: TrackGraph | None = None track_dot_file = track_svg_file = None if args.track: track = build_track_graph(elements, by_id, graph, warnings) track_dot_target = results / f"{stem}_track.dot" track_dot_target.write_text(render_track_dot(track), encoding="utf-8") track_dot_file = track_dot_target if args.tosvg: track_svg_target = results / f"{stem}_track.svg" try: dot_to_svg(track_dot_file, track_svg_target) track_svg_file = track_svg_target except RuntimeError as exc: print(f"FEHLER: Bahnzusammenhang-SVG nicht erzeugt: {exc}", file=sys.stderr) chain_file = None if args.chain_graph: try: svg = render_chain_svg(elements, graph, csv_file.name) except ValueError as exc: print(f"WARNUNG: Chain-Graph: {exc} - nicht erzeugt.", file=sys.stderr) else: chain_target = results / f"{stem}_chain.svg" chain_target.write_text(svg, encoding="utf-8") chain_file = chain_target report( graph, findings, csv_file, dot_file, created_svg, doc_file, stale_svg=None if args.tosvg and svg_error is None else existing_svg, bbs_file=bbs_file, track_dot_file=track_dot_file, track_svg_file=track_svg_file, chain_file=chain_file, ) if svg_error: print(f"FEHLER: SVG nicht erzeugt: {svg_error}", file=sys.stderr) return 2 return 0 if __name__ == "__main__": sys.exit(main())