From 4454af51651712c3d7bb22fcbcd0c2837aa9952b Mon Sep 17 00:00:00 2001 From: Samer Ayadi Date: Thu, 30 Jul 2026 13:13:01 +0200 Subject: [PATCH] Add material flow generator for the mechanical layout Reads the CSV export of the plant objects (ILS 2.0) from SKEL_DATA and builds a directed graph of the material flow, written as Graphviz DOT and optionally SVG into SKEL_RESULTS. Node model: one node per Gefaellestrecke and Strecke; a Kreisel becomes two nodes (-L / -R) wired into a circle, since it is a carousel with one lane per travel direction. Separators and scanners are no nodes of their own; they are shown on the label of the object they are assigned to. Edge direction is not in the CSV - the Nachbarn column is symmetric. It is derived from geometry instead: a Gefaellestrecke runs downhill away from its highest neighbour, a driven Strecke follows Antriebfahrtrichtung ("Auf" uphill, "Ab" downhill). Connections that stay ambiguous, e.g. between two Kreisel at the same height, are drawn as bidirectional edges and reported. The lane suffix of a Kreisel is always stated in the *partner's* row, so both viewpoints of a link are merged before orienting it. Switches: --file name of the CSV in SKEL_DATA, or a full path --tosvg render the DOT via Graphviz --doc write a Markdown documentation of the plant objects Co-Authored-By: Claude Opus 5 (1M context) --- bin/material_flow.bat | 30 ++ bin/material_flow.sh | 31 ++ lib/material_flow.py | 1079 +++++++++++++++++++++++++++++++++++++++++ 3 files changed, 1140 insertions(+) create mode 100644 bin/material_flow.bat create mode 100644 bin/material_flow.sh create mode 100644 lib/material_flow.py diff --git a/bin/material_flow.bat b/bin/material_flow.bat new file mode 100644 index 0000000..3f77368 --- /dev/null +++ b/bin/material_flow.bat @@ -0,0 +1,30 @@ +@echo off +REM ================================================================ +REM SPS_SKEL - Materialfluss-Graph erzeugen +REM ================================================================ +REM Aufruf: +REM bin\material_flow.bat --file export.csv +REM bin\material_flow.bat --file export.csv --tosvg +REM +REM Die CSV-Datei wird in %SKEL_DATA% gesucht, DOT und SVG werden +REM in %SKEL_RESULTS% abgelegt. +REM ================================================================ + +setlocal + +REM Umgebung setzen (SKEL_DATA, SKEL_RESULTS, SKEL_LIB, PYTHONPATH ...) +call "%~dp0setenv.bat" >nul + +REM Virtuelle Umgebung nutzen, falls vorhanden +if exist "%SPS_SKEL%\.venv\Scripts\activate.bat" ( + call "%SPS_SKEL%\.venv\Scripts\activate.bat" +) + +REM Python-Interpreter bestimmen (py-Launcher bevorzugt, sonst python) +set "PY=py" +py --version >nul 2>&1 || set "PY=python" + +"%PY%" "%SKEL_LIB%\material_flow.py" %* +set "RC=%ERRORLEVEL%" + +endlocal & exit /b %RC% diff --git a/bin/material_flow.sh b/bin/material_flow.sh new file mode 100644 index 0000000..d457e51 --- /dev/null +++ b/bin/material_flow.sh @@ -0,0 +1,31 @@ +#!/usr/bin/env bash +# ================================================================ +# SPS_SKEL - Materialfluss-Graph erzeugen +# ================================================================ +# Aufruf: +# bash bin/material_flow.sh --file export.csv +# bash bin/material_flow.sh --file export.csv --tosvg +# +# Die CSV-Datei wird in $SKEL_DATA gesucht, DOT und SVG werden +# in $SKEL_RESULTS abgelegt. +# ================================================================ + +set -eu + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" + +# Umgebung setzen (SKEL_DATA, SKEL_RESULTS, SKEL_LIB, PYTHONPATH ...) +# shellcheck source=/dev/null +. "$SCRIPT_DIR/setenv.sh" >/dev/null + +# Virtuelle Umgebung nutzen, falls vorhanden +if [ -f "$SPS_SKEL/.venv/bin/activate" ]; then + # shellcheck source=/dev/null + . "$SPS_SKEL/.venv/bin/activate" +fi + +# Python-Interpreter bestimmen +PY="python3" +command -v python3 >/dev/null 2>&1 || PY="python" + +exec "$PY" "$SKEL_LIB/material_flow.py" "$@" diff --git a/lib/material_flow.py b/lib/material_flow.py new file mode 100644 index 0000000..a93465a --- /dev/null +++ b/lib/material_flow.py @@ -0,0 +1,1079 @@ +# -*- 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 als + Kreis miteinander verbunden sind + (L -> R und R -> L) +* Separator / Scanner -> kein eigener Knoten; sie werden + ueber "Zuordnung" am Label des + uebergeordneten Knotens angezeigt + +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 +from datetime import datetime +from pathlib import Path + +# --------------------------------------------------------------------------- +# Konstanten +# --------------------------------------------------------------------------- + +CSV_DELIMITER = ";" +CSV_ENCODING = "utf-8-sig" +TEILEART_PREFIX = "ILS 2.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") + +LANE_LEFT = "L" +LANE_RIGHT = "R" +LANE_NAMES = {LANE_LEFT: "links", LANE_RIGHT: "rechts"} + +UNASSIGNED = "nicht zugeordnet" + +# "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"), +} +DEFAULT_NODE_STYLE = dict(fillcolor="#f2f2f2", fontcolor="#404040", color="#808080") + + +# --------------------------------------------------------------------------- +# 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]] + merkmale: dict + row: int + position: tuple[float, float, float] | None = None + + @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 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 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 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 = "" + + +# --------------------------------------------------------------------------- +# 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 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 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 + + 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 ""), + merkmale=merkmale, + row=offset, + position=parse_position(row.get("Position") or ""), + ) + + 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 not element.is_device: + warnings.append( + f"{element.describe()}: unbekannte TeileArt '{element.teile_art}' - kein Knoten erzeugt" + ) + return nodes + + +def _attach_devices( + elements: list[Element], nodes: dict[str, Node], warnings: list[str] +) -> list[Element]: + """Separatoren/Scanner ueber 'Zuordnung' an ihren Knoten haengen.""" + unassigned: list[Element] = [] + for element in elements: + if not element.is_device: + continue + target = element.assignment + if not target or target == UNASSIGNED: + unassigned.append(element) + continue + if target not in nodes: + warnings.append( + f"{element.describe()}: Zuordnung '{target}' zeigt auf keinen Knoten - nicht dargestellt" + ) + unassigned.append(element) + continue + nodes[target].devices.setdefault(element.kind, []).append(element.teile_id) + 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 _flow_uphill(element: Element, warnings: list[str]) -> bool | None: + """ + True = Fluss laeuft zum hoechsten Nachbarn hin (bergauf) + False = Fluss laeuft vom hoechsten Nachbarn weg (bergab) + None = nicht bestimmbar + """ + if element.kind == "Gefaellestrecke": + return False + direction = element.drive_dir.lower() + if direction.startswith("auf"): + return True + if direction.startswith("ab"): + return False + warnings.append( + f"{element.describe()}: Antriebfahrtrichtung fehlt oder unbekannt " + f"('{element.drive_dir}') - Richtung nicht bestimmbar" + ) + return None + + +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 build_graph(elements: list[Element], source: str, warnings: list[str]) -> Graph: + by_id = {element.teile_id: element for element in elements} + nodes = _build_nodes(elements, warnings) + unassigned = _attach_devices(elements, nodes, 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: die beiden Bahnen bilden den Kreis + for element in elements: + if not element.is_circle: + continue + left = f"{element.teile_id}-{LANE_LEFT}" + right = f"{element.teile_id}-{LANE_RIGHT}" + edges.append(Edge(left, right, EDGE_CIRCLE)) + edges.append(Edge(right, left, EDGE_CIRCLE)) + + return Graph(nodes=nodes, edges=edges, warnings=warnings, unassigned=unassigned, source=source) + + +# --------------------------------------------------------------------------- +# DOT erzeugen +# --------------------------------------------------------------------------- + + +def _node_label(node: Node) -> str: + element = node.element + lines: list[str] = [] + + 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.kind == "Strecke" and element.drive_dir: + lines.append(f"Antrieb: {element.drive_dir}") + + 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) + + +def render_dot(graph: Graph) -> str: + out: list[str] = [] + add = out.append + + add("// Automatisch erzeugt von lib/material_flow.py - nicht manuell aendern.") + add(f"// Quelle: {graph.source}") + 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 + add(f" subgraph cluster_kreisel_{index} {{") + add(f' label="{_dot_escape(element.kind)} {_dot_escape(str(name))} [{teile_id}]";') + 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}];') + add(" { rank=same; " + " ".join(f'"{n.node_id}";' for n in lanes) + " }") + add(" }") + add("") + + for node in plain: + 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}];') + add("") + + for edge in graph.edges: + if edge.kind == EDGE_CIRCLE: + attrs = ('[color="#8ea9db", style=dashed, penwidth=1.0, constraint=false, ' + 'arrowsize=0.6, tooltip="Kreisel-Umlauf"]') + elif edge.kind == EDGE_UNRESOLVED: + attrs = ('[color="#bf8f00", style=dashed, dir=both, ' + f'tooltip="Richtung {edge.note}"]') + else: + attrs = "[]" + 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=<' + '' + '' + '' + '' + '' + '' + "
durchgezogenMaterialfluss
gestrichelt blauKreisel-Umlauf (L / R)
gestrichelt gelbRichtung nicht bestimmbar
>];") + add(" }") + 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 dot_to_svg(dot_file: Path, svg_file: Path) -> None: + """SVG aus der .dot-Datei erzeugen (Graphviz).""" + executable = find_dot() + if not executable: + raise RuntimeError( + "Graphviz 'dot' nicht gefunden. Bitte Graphviz installieren " + "(https://graphviz.org/download/) oder GRAPHVIZ_DOT auf die " + "dot-Programmdatei setzen." + ) + + result = subprocess.run( + [executable, "-Tsvg", str(dot_file), "-o", str(svg_file)], + capture_output=True, + text=True, + ) + if result.returncode != 0: + raise RuntimeError( + f"'dot' ist mit Code {result.returncode} fehlgeschlagen:\n" + f"{(result.stderr or result.stdout).strip()}" + ) + + +# --------------------------------------------------------------------------- +# 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) + + 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: + add(f"| `{edge.src}` | `{edge.dst}` | {kind_text.get(edge.kind, edge.kind)} |") + 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, +) -> 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("================================================================") + + 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" + "\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( + "--doc", + action="store_true", + help="Dokumentation des Materialflusses als Markdown in %%SKEL_RESULTS%% " + "erzeugen (Anlagenobjekte, Verbindungen, Pruefungen).", + ) + return parser.parse_args(argv) + + +def main(argv: list[str] | None = None) -> int: + args = parse_args(argv) + + csv_file = resolve_input(args.file) + if not csv_file.is_file(): + print(f"FEHLER: Eingabedatei nicht gefunden: {csv_file}", file=sys.stderr) + print(f" Suchpfad fuer Dateinamen: {env_dir('SKEL_DATA', 'data')}", + file=sys.stderr) + return 1 + + results = env_dir("SKEL_RESULTS", "results") + results.mkdir(parents=True, exist_ok=True) + + warnings: list[str] = [] + try: + elements = read_elements(csv_file, warnings) + except (OSError, ValueError, csv.Error) as exc: + print(f"FEHLER: {csv_file} konnte nicht gelesen werden: {exc}", file=sys.stderr) + return 1 + + if not elements: + print(f"FEHLER: {csv_file} enthaelt keine Objekte.", file=sys.stderr) + return 1 + + graph = build_graph(elements, csv_file.name, warnings) + if not graph.nodes: + print(f"FEHLER: {csv_file} enthaelt keine Flussobjekte " + f"({', '.join(TRANSPORT_KINDS + CIRCLE_KINDS)}).", file=sys.stderr) + return 1 + + findings = check_elements(graph, elements) + + stem = csv_file.stem + dot_file = results / f"{stem}_material_flow.dot" + svg_file = results / f"{stem}_material_flow.svg" + doc_target = results / f"{stem}_material_flow.md" + + existing_svg = svg_file if svg_file.is_file() else None + dot_file.write_text(render_dot(graph), encoding="utf-8") + + svg_error: str | None = None + if args.tosvg: + try: + dot_to_svg(dot_file, svg_file) + except RuntimeError as exc: + svg_error = str(exc) + created_svg = svg_file if args.tosvg and svg_error is None else None + + doc_file = None + if args.doc: + doc_target.write_text( + render_doc(graph, findings, csv_file, dot_file, created_svg), + encoding="utf-8", + ) + doc_file = doc_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, + ) + + if svg_error: + print(f"FEHLER: SVG nicht erzeugt: {svg_error}", file=sys.stderr) + return 2 + return 0 + + +if __name__ == "__main__": + sys.exit(main())