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())