s.ayadi
5dcaa8b8fa
The new export fills Hoehe_Von_mm / Hoehe_Bis_mm, so the direction of a transport
element no longer has to be guessed from its neighbours. drive_uphill() now
decides in this order:
1. Hoehe_Von_mm -> Hoehe_Bis_mm: the rise of the element itself. Only the sign
is read, since the values are absolute on a Gefaellestrecke (1941 -> 1466)
but relative on a Strecke (0 -> 474).
2. Antriebfahrtrichtung "Auf" (up) or "Ab" (down).
3. A Gefaellestrecke without either runs downhill by definition.
Previously a Gefaellestrecke was assumed downhill outright and everything else
depended on Antriebfahrtrichtung alone; now both are backed by the height data
and the fallback order is explicit.
The node label states what was read and what follows from it, e.g.
"0 -> 474 mm (+474) = aufwaerts", so the arrow can be checked against the export
without opening the CSV.
Adds a plausibility check for the case where Antriebfahrtrichtung and the height
change disagree. The height wins and the conflict is reported, rather than one of
the two silently deciding.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
1250 lines
45 KiB
Python
1250 lines
45 KiB
Python
# -*- 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)
|
|
-------------------------------
|
|
<name>_material_flow.dot immer
|
|
<name>_material_flow.svg mit --tosvg (Graphviz)
|
|
<name>_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<base>[^\s-]+)(?:-(?P<lane>[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 height_from(self) -> float | None:
|
|
"""Hoehe am Anfang des Objekts in mm (Merkmal Hoehe_Von_mm)."""
|
|
return _as_float(_merkmal(self.merkmale, "Hoehe_Von_mm", "Höhe_Von_mm"))
|
|
|
|
@property
|
|
def height_to(self) -> float | None:
|
|
"""Hoehe am Ende des Objekts in mm (Merkmal Hoehe_Bis_mm)."""
|
|
return _as_float(_merkmal(self.merkmale, "Hoehe_Bis_mm", "Höhe_Bis_mm"))
|
|
|
|
@property
|
|
def rise(self) -> float | None:
|
|
"""
|
|
Hoehenaenderung in Foerderrichtung in mm.
|
|
|
|
Positiv = das Objekt foerdert nach oben, negativ = nach unten. Die Werte
|
|
koennen absolut sein (Gefaellestrecke: 1941 -> 1466) oder relativ zum
|
|
Objekt (Strecke: 0 -> 474); nur das Vorzeichen wird ausgewertet.
|
|
"""
|
|
start, end = self.height_from, self.height_to
|
|
if start is None or end is None:
|
|
return None
|
|
return end - start
|
|
|
|
@property
|
|
def assignment(self) -> str:
|
|
return str(_merkmal(self.merkmale, "Zuordnung") or "").strip()
|
|
|
|
def describe(self) -> str:
|
|
return f"Zeile {self.row} ({self.kind} {self.teile_id} '{self.name}')"
|
|
|
|
|
|
@dataclass
|
|
class Node:
|
|
"""Ein Knoten des Materialfluss-Graphen."""
|
|
|
|
node_id: str
|
|
kind: str
|
|
element: Element
|
|
lane: str | None = None
|
|
devices: dict[str, list[str]] = field(default_factory=dict)
|
|
|
|
@property
|
|
def height(self) -> float | None:
|
|
return self.element.height
|
|
|
|
|
|
@dataclass
|
|
class Edge:
|
|
src: str
|
|
dst: str
|
|
kind: str
|
|
note: str = ""
|
|
|
|
|
|
@dataclass
|
|
class Graph:
|
|
nodes: dict[str, Node] = field(default_factory=dict)
|
|
edges: list[Edge] = field(default_factory=list)
|
|
warnings: list[str] = field(default_factory=list)
|
|
unassigned: list[Element] = field(default_factory=list)
|
|
source: str = ""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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 drive_uphill(element: Element) -> tuple[bool | None, str]:
|
|
"""
|
|
Foerdert das Objekt nach oben?
|
|
|
|
True = Fluss laeuft zum hoechsten Nachbarn hin (bergauf)
|
|
False = Fluss laeuft vom hoechsten Nachbarn weg (bergab)
|
|
None = nicht bestimmbar
|
|
|
|
Bewertet wird in dieser Reihenfolge:
|
|
1. Hoehe_Von_mm / Hoehe_Bis_mm - die Hoehenaenderung des Objekts selbst
|
|
2. Antriebfahrtrichtung "Auf" (nach oben) bzw. "Ab" (nach unten)
|
|
3. Gefaellestrecke ohne Angaben laeuft per Definition bergab
|
|
|
|
Der zweite Rueckgabewert nennt die verwendete Quelle fuer Bericht und
|
|
Zeichnung.
|
|
"""
|
|
rise = element.rise
|
|
if rise is not None and rise != 0:
|
|
direction = "Auf" if rise > 0 else "Ab"
|
|
return rise > 0, (
|
|
f"Hoehe_Von {element.height_from:.0f} mm -> Hoehe_Bis "
|
|
f"{element.height_to:.0f} mm ({direction}, {rise:+.0f} mm)"
|
|
)
|
|
|
|
drive = element.drive_dir.lower()
|
|
if drive.startswith("auf"):
|
|
return True, f"Antriebfahrtrichtung '{element.drive_dir}' (nach oben)"
|
|
if drive.startswith("ab"):
|
|
return False, f"Antriebfahrtrichtung '{element.drive_dir}' (nach unten)"
|
|
|
|
if element.kind == "Gefaellestrecke":
|
|
return False, "Gefaellestrecke ohne Hoehenangabe - laeuft bergab"
|
|
|
|
return None, (
|
|
f"weder Hoehe_Von/Bis noch Antriebfahrtrichtung verwertbar "
|
|
f"(Antrieb '{element.drive_dir}', Hoehen "
|
|
f"{element.height_from}/{element.height_to})"
|
|
)
|
|
|
|
|
|
def _flow_uphill(element: Element, warnings: list[str]) -> bool | None:
|
|
"""Wie drive_uphill, meldet aber unbestimmbare Faelle als Warnung."""
|
|
uphill, reason = drive_uphill(element)
|
|
if uphill is None:
|
|
warnings.append(f"{element.describe()}: {reason} - Richtung nicht bestimmbar")
|
|
return uphill
|
|
|
|
|
|
def _orient(
|
|
pairs: list[tuple[str, str]], nodes: dict[str, Node], warnings: list[str]
|
|
) -> list[Edge]:
|
|
"""Flussrichtung je Verbindung aus Hoehe und Antriebsrichtung ableiten."""
|
|
incident: dict[str, list[int]] = {}
|
|
for index, (left, right) in enumerate(pairs):
|
|
incident.setdefault(left, []).append(index)
|
|
incident.setdefault(right, []).append(index)
|
|
|
|
# Vorschlaege je Verbindung, gesammelt aus der Sicht der Transportobjekte
|
|
proposals: dict[int, set[tuple[str, str]]] = {index: set() for index in range(len(pairs))}
|
|
|
|
for node_id, node in sorted(nodes.items()):
|
|
if node.kind not in TRANSPORT_KINDS:
|
|
continue
|
|
|
|
indices = incident.get(node_id, [])
|
|
if not indices:
|
|
warnings.append(
|
|
f"{node.element.describe()}: keine Nachbarn - Knoten bleibt unverbunden"
|
|
)
|
|
continue
|
|
|
|
partners = {}
|
|
for index in indices:
|
|
left, right = pairs[index]
|
|
partners[index] = right if left == node_id else left
|
|
|
|
heights = {index: nodes[partner].height for index, partner in partners.items()}
|
|
missing = [partners[i] for i, h in heights.items() if h is None]
|
|
if missing:
|
|
warnings.append(
|
|
f"{node.element.describe()}: Hoehe von {', '.join(sorted(missing))} unbekannt - "
|
|
f"Richtung nicht bestimmbar"
|
|
)
|
|
continue
|
|
|
|
uphill = _flow_uphill(node.element, warnings)
|
|
if uphill is None:
|
|
continue
|
|
|
|
highest = max(heights.values())
|
|
if highest == min(heights.values()):
|
|
warnings.append(
|
|
f"{node.element.describe()}: alle Nachbarn auf gleicher Hoehe "
|
|
f"({highest:.3f} m) - Richtung nicht bestimmbar"
|
|
)
|
|
continue
|
|
|
|
for index, partner in partners.items():
|
|
at_top = heights[index] == highest
|
|
if uphill == at_top:
|
|
proposals[index].add((node_id, partner))
|
|
else:
|
|
proposals[index].add((partner, node_id))
|
|
|
|
edges: list[Edge] = []
|
|
for index, (left, right) in enumerate(pairs):
|
|
found = proposals[index]
|
|
if len(found) == 1:
|
|
src, dst = next(iter(found))
|
|
edges.append(Edge(src, dst, EDGE_FLOW))
|
|
elif len(found) > 1:
|
|
warnings.append(
|
|
f"Verbindung {left} <-> {right}: widerspruechliche Richtungen "
|
|
f"({', '.join(f'{s} -> {d}' for s, d in sorted(found))}) - beidseitig gezeichnet"
|
|
)
|
|
edges.append(Edge(left, right, EDGE_UNRESOLVED, "widerspruechlich"))
|
|
else:
|
|
warnings.append(
|
|
f"Verbindung {left} <-> {right}: Richtung nicht bestimmbar "
|
|
f"(kein Transportobjekt beteiligt) - beidseitig gezeichnet"
|
|
)
|
|
edges.append(Edge(left, right, EDGE_UNRESOLVED, "unbestimmt"))
|
|
|
|
return edges
|
|
|
|
|
|
def 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.is_transport:
|
|
uphill = drive_uphill(element)[0]
|
|
arrow = {True: "aufwaerts", False: "abwaerts", None: "Richtung offen"}[uphill]
|
|
if element.rise is not None:
|
|
lines.append(f"{element.height_from:.0f} -> {element.height_to:.0f} mm "
|
|
f"({element.rise:+.0f}) = {arrow}")
|
|
elif element.drive_dir:
|
|
lines.append(f"Antrieb: {element.drive_dir} = {arrow}")
|
|
else:
|
|
lines.append(arrow)
|
|
|
|
for kind in DEVICE_KINDS:
|
|
ids = node.devices.get(kind)
|
|
if ids:
|
|
lines.append(f"{kind} ({len(ids)}): {', '.join(sorted(ids))}")
|
|
|
|
return _dot_label(lines)
|
|
|
|
|
|
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=<'
|
|
'<table border="0" cellborder="0" cellspacing="2" cellpadding="2">'
|
|
'<tr><td align="left">durchgezogen</td><td align="left">Materialfluss</td></tr>'
|
|
'<tr><td align="left"><font color="#8ea9db">gestrichelt blau</font></td>'
|
|
'<td align="left">Kreisel-Umlauf, Richtung aus Drehrichtung '
|
|
f'({ROTATION_CW} = {ROTATION_FLOW[ROTATION_CW]}, '
|
|
f'{ROTATION_CCW} = {ROTATION_FLOW[ROTATION_CCW]})</td></tr>'
|
|
'<tr><td align="left"><font color="#bf8f00">gestrichelt gelb</font></td>'
|
|
'<td align="left">Richtung nicht bestimmbar</td></tr>'
|
|
"</table>>];")
|
|
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)
|
|
|
|
# Widerspruch zwischen Hoehenangabe und Antriebfahrtrichtung
|
|
for element in elements:
|
|
if not element.is_transport:
|
|
continue
|
|
rise, drive = element.rise, element.drive_dir.lower()
|
|
if rise is None or rise == 0 or not drive:
|
|
continue
|
|
stated_up = drive.startswith("auf")
|
|
if stated_up != (rise > 0):
|
|
findings.append(
|
|
f"{element.kind} {element.teile_id} '{element.name}': "
|
|
f"Antriebfahrtrichtung '{element.drive_dir}' passt nicht zu "
|
|
f"Hoehe_Von {element.height_from:.0f} -> Hoehe_Bis "
|
|
f"{element.height_to:.0f} mm ({rise:+.0f}) - die Hoehenangabe gilt"
|
|
)
|
|
|
|
for element in elements:
|
|
if not (element.is_transport or element.is_circle):
|
|
continue
|
|
node_ids = (
|
|
[element.teile_id]
|
|
if element.is_transport
|
|
else [f"{element.teile_id}-{LANE_LEFT}", f"{element.teile_id}-{LANE_RIGHT}"]
|
|
)
|
|
for label, keys in (
|
|
("Separator", ("Anzahl_Separator", "Anzahl der Separatoren")),
|
|
("Scanner", ("Anzahl_Scanner", "Anzahl der Scanner")),
|
|
):
|
|
expected = _as_float(_merkmal(element.merkmale, *keys))
|
|
if expected is None:
|
|
continue
|
|
actual = sum(counted.get(n, {}).get(label, 0) for n in node_ids)
|
|
if int(expected) != actual:
|
|
findings.append(
|
|
f"{element.kind} {element.teile_id} '{element.name}': Merkmal "
|
|
f"'{keys[0] if keys[0] in element.merkmale else keys[1]}' = "
|
|
f"{int(expected)}, tatsaechlich {actual} {label}"
|
|
)
|
|
findings.sort()
|
|
return findings
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Dokumentation (Markdown)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
GENERATED_MARKER = "<!-- erzeugt von lib/material_flow.py -->"
|
|
|
|
|
|
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())
|