# -*- 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" 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" # 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"), } 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 rotation(self) -> str: """Drehrichtung eines Kreisels: UZS (im Uhrzeigersinn) oder GUZ.""" return str(_merkmal(self.merkmale, "Drehrichtung") or "").strip().upper() @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: 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")) # 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) # --------------------------------------------------------------------------- # 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 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}];') 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: # 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: 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, Richtung aus Drehrichtung ' f'({ROTATION_CW} = {ROTATION_FLOW[ROTATION_CW]}, ' f'{ROTATION_CCW} = {ROTATION_FLOW[ROTATION_CCW]})
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 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 meldet fehlende Positionen nur als Warnung auf stderr if result.stderr.strip(): raise RuntimeError(f"'{engine}' meldet: {result.stderr.strip()}") 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) 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, ) -> 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())