m.stangl
90bfbd89b6
export_graph.py behandelte BTMT-Stationen bisher wie ein generisches Durchlaufelement mit zwei Achs-Enden - dadurch entstanden mehrere, teils widerspruechliche Kanten und die Rolle blieb "Element" statt "Eingang"/"Ausgang". BTMT hat aber physisch nur EINEN Anschluss zum Foerdersystem (sitzt als dritter Zweig einer Weiche auf der Kreisel- Durchgangsbahn, der Kreisel bleibt dabei ein einzelner Verteiler-Knoten): Beladung speist fest Material ein (kein Vorgaenger im Fluss), Entladung nimmt es fest raus (kein Nachfolger). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
633 lines
24 KiB
Python
633 lines
24 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
export_graph.py - Baut aus dem CSV-Export (EXPORTCSV) einen Materialfluss-Graph.
|
|
|
|
Aufruf:
|
|
python export_graph.py <export.csv> [<ausgabe-basispfad>]
|
|
|
|
Erzeugt neben <basispfad>.json (vollstaendiges Graph-Modell) auch
|
|
<basispfad>.dot (GraphViz) und <basispfad>.mmd (Mermaid) zur Visualisierung.
|
|
|
|
KNOTEN = ein Anlagenelement je CSV-Zeile (Kreisel, Gefaellestrecke, Strecke,
|
|
BTMT Beladung/Entladung als Ein-/Ausgang des Graphen).
|
|
KANTEN = eine Materialuebergabe zwischen zwei Elementen.
|
|
|
|
Warum nicht die CSV-Spalte "Nachbarn"?
|
|
-------------------------------------
|
|
Die Spalte ist bewusst KEINE Flussinformation (siehe export_neighbors.py):
|
|
* Sie ist ungerichtet - der Materialfluss braucht aber eine Richtung.
|
|
* Strecken-Teilearten werden dort absichtlich NICHT gegeneinander
|
|
getestet, nur gegen Kreisel/Eckrad.
|
|
* Sie beruht auf Bounding-Box-Ueberschneidung im Grundriss. Ein langer
|
|
Kreisel ueberdeckt dadurch praktisch die gesamte Anlage und bekommt
|
|
Nachbarn, mit denen er keinen Uebergabepunkt teilt.
|
|
|
|
Dieses Modul arbeitet stattdessen mit ANSCHLUSSPUNKTEN (Ports): jedes
|
|
Element hat entlang seiner Laengsachse zwei Enden. Eine Kante entsteht nur,
|
|
wenn das Ende eines Elements im Grundriss auf der Flaeche eines anderen
|
|
Elements liegt (Toleranz siehe PORT_TOLERANZ_MM). Das ergibt genau die
|
|
realen Uebergabestellen statt flaechiger Ueberdeckung.
|
|
|
|
Richtungsermittlung
|
|
-------------------
|
|
Sobald der Export die Spalten K1/K2 (KS_EIN/KS_AUS) fuellt, ist die Richtung
|
|
exakt: K1 = Eingang, K2 = Ausgang, Kante immer K2 -> K1. Solange diese
|
|
Spalten leer sind (aktueller Stand, siehe csv:get-k-kos-strings in
|
|
Lisp/export.lsp), wird die Richtung heuristisch abgeleitet:
|
|
|
|
* Gefaellestrecke: Schwerkraft - Fluss vom hoeher zum tiefer liegenden
|
|
Anschluss. Die Endhoehen werden aus der Hoehe des jeweils dort
|
|
angeschlossenen Partners bestimmt.
|
|
* Strecke/Foerderer: Merkmal "Antriebfahrtrichtung" ("Auf" = aufwaerts,
|
|
"Ab" = abwaerts) gegen die Partnerhoehen.
|
|
* Kreisel: Verteilknoten. Richtung ergibt sich aus den angeschlossenen
|
|
Elementen, nicht aus dem Kreisel selbst.
|
|
* BTMT Beladung/Entladung: FEST gerichtet (Beladung -> Partner, Partner ->
|
|
Entladung), unabhaengig von Hoehe/Antrieb - siehe naechster Abschnitt.
|
|
|
|
Jede Kante traegt daher ein Feld "richtung_quelle": "ks" (exakt, aus K1/K2),
|
|
"gefaelle", "antrieb", "btmt" oder "unbestimmt".
|
|
|
|
Ein-/Ausgaenge des Graphen: BTMT Beladung/Entladung
|
|
----------------------------------------------------
|
|
BTMT Beladung/Entladung sind die Ein- bzw. Ausgaenge des gesamten
|
|
Materialfluss-Graphen - eine Beladestation speist NEUES Material ein (kein
|
|
Vorgaenger im Fluss), eine Entladestation nimmt es endgueltig raus (kein
|
|
Nachfolger). Physisch sitzen sie AUF der Durchgangsbahn eines Kreisels
|
|
(der Kreisel laeuft an dieser Stelle durch UND die BTMT-Station zweigt dort
|
|
zusaetzlich ab/ein - wie eine Weiche mit einem fixen dritten Zweig). Der
|
|
Kreisel bleibt im Graphmodell trotzdem EIN Knoten (Rolle "Verteiler", keine
|
|
Aufteilung in Segmente); BTMT wird als eigener Knoten mit GENAU EINEM
|
|
Anschlusspunkt (Elementmitte statt zwei Achs-Enden, siehe _enden) und einer
|
|
fest gerichteten Kante zum Kreisel/Partner dargestellt (Rolle "Eingang"
|
|
bzw. "Ausgang", siehe _rolle/BTMT_ARTEN).
|
|
"""
|
|
|
|
import csv
|
|
import json
|
|
import math
|
|
import os
|
|
import sys
|
|
|
|
# Base64-Alphabet von csv:kos-encode / csv:trans-encode (Lisp/export.lsp)
|
|
B64_CHARS = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"
|
|
B64_INDEX = {c: i for i, c in enumerate(B64_CHARS)}
|
|
KOS_MAX = 8388607 # Saettigungsgrenze der 24-Bit-Fixed-Point-Werte
|
|
KOS_FAKTOR = 1000.0 # csv:kos-encode (Insertpoint)
|
|
TRANS_FAKTOR = 10.0 # csv:trans-encode (K1-K4)
|
|
|
|
# Ein Anschluss gilt als verbunden, wenn das Elementende im Grundriss
|
|
# hoechstens so weit von der Partnerflaeche entfernt liegt (mm).
|
|
PORT_TOLERANZ_MM = 400.0
|
|
|
|
KREISEL_ARTEN = {"ILS 2.0 Kreisel", "ILS 2.0 Eckrad"}
|
|
GEFAELLE_ARTEN = {"ILS 2.0 Gefaellestrecke"}
|
|
STRECKEN_ARTEN = {"ILS 2.0 Strecke", "ILS 2.0 Strecke - Modul"}
|
|
|
|
# Ein-/Ausgaenge des Materialfluss-Graphen: BTMT Beladung speist Material
|
|
# EIN (Graph-Quelle, kein Vorgaenger im Fluss), BTMT Entladung nimmt es
|
|
# AUS (Graph-Senke, kein Nachfolger). Anders als Kreisel/Strecken haben
|
|
# beide nur EINEN Anschluss zum Foerdersystem, keine zwei Enden einer
|
|
# Durchlaufachse - siehe _enden().
|
|
BTMT_BELADUNG_ARTEN = {"ILS 2.0 BTMT Beladung"}
|
|
BTMT_ENTLADUNG_ARTEN = {"ILS 2.0 BTMT Entladung"}
|
|
BTMT_ARTEN = BTMT_BELADUNG_ARTEN | BTMT_ENTLADUNG_ARTEN
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Dekodierung der Base64-Koordinatenstrings
|
|
# ------------------------------------------------------------------
|
|
|
|
def _decode_ints(text):
|
|
"""Base64-String in 24-Bit-Integer je 4 Zeichen zurueckrechnen."""
|
|
werte = []
|
|
for start in range(0, len(text) - 3, 4):
|
|
wert = 0
|
|
for zeichen in text[start:start + 4]:
|
|
if zeichen not in B64_INDEX:
|
|
return []
|
|
wert = (wert << 6) | B64_INDEX[zeichen]
|
|
if wert >= 8388608: # Zweierkomplement (24 Bit)
|
|
wert -= 16777216
|
|
werte.append(wert)
|
|
return werte
|
|
|
|
|
|
def decode_insertpoint(text):
|
|
"""Insertpoint (24 Zeichen) -> dict mit Position, Drehung, Saettigung.
|
|
|
|
Die Drehung steckt als Halbwinkel-Quaternion (nur qz ist belegt, reine
|
|
Z-Drehung, siehe csv:z-angle-to-quat). Rueckgabe None bei ungueltigem
|
|
String.
|
|
"""
|
|
werte = _decode_ints((text or "").strip())
|
|
if len(werte) < 6:
|
|
return None
|
|
x, y, z, _qx, _qy, qz = werte[:6]
|
|
qz_norm = max(-1.0, min(1.0, qz / float(KOS_MAX)))
|
|
winkel = math.degrees(2.0 * math.asin(min(1.0, abs(qz_norm))))
|
|
if qz_norm < 0:
|
|
winkel = -winkel
|
|
return {
|
|
"x": x / KOS_FAKTOR,
|
|
"y": y / KOS_FAKTOR,
|
|
"z": z / KOS_FAKTOR,
|
|
"drehung_grad": round(winkel, 4),
|
|
# Achsen, deren Wert an der Fixed-Point-Grenze haengt und damit
|
|
# unbrauchbar ist (siehe Modul-Hinweis in report_saturation).
|
|
"gesaettigt": [name for name, wert in (("x", x), ("y", y), ("z", z))
|
|
if abs(wert) >= KOS_MAX],
|
|
}
|
|
|
|
|
|
def decode_trans(text):
|
|
"""K1-K4 (12 Zeichen, nur Position) -> (x, y, z) in mm, oder None."""
|
|
werte = _decode_ints((text or "").strip())
|
|
if len(werte) < 3:
|
|
return None
|
|
return tuple(w / TRANS_FAKTOR for w in werte[:3])
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# CSV einlesen
|
|
# ------------------------------------------------------------------
|
|
|
|
def _zahlen(text):
|
|
"""'1, 2, 3' -> [1.0, 2.0, 3.0]; leerer Text -> []."""
|
|
text = (text or "").strip()
|
|
if not text:
|
|
return []
|
|
try:
|
|
return [float(teil) for teil in text.split(",")]
|
|
except ValueError:
|
|
return []
|
|
|
|
|
|
def load_elements(csv_pfad):
|
|
"""CSV-Zeilen in Element-dicts umwandeln (Geometrie bereits ausgewertet)."""
|
|
with open(csv_pfad, encoding="utf-8-sig", newline="") as fh:
|
|
zeilen = list(csv.DictReader(fh, delimiter=";"))
|
|
|
|
elemente = []
|
|
for zeile in zeilen:
|
|
art = (zeile.get("TeileArt") or "").strip()
|
|
# Die synthetische Summenzeile (Omniflo Sum) ist kein Bauteil.
|
|
if art.endswith("Sum"):
|
|
continue
|
|
mitte = _zahlen(zeile.get("Position"))
|
|
masse = _zahlen(zeile.get("Boundingbox"))
|
|
if len(mitte) < 3 or len(masse) < 3:
|
|
continue
|
|
|
|
try:
|
|
merkmale = json.loads(zeile.get("Merkmale") or "{}")
|
|
except (ValueError, TypeError):
|
|
merkmale = {}
|
|
|
|
element = {
|
|
"nr": (zeile.get("Elementnummer") or "").strip(),
|
|
"teileid": (zeile.get("TeileId") or "").strip(),
|
|
"teileart": art,
|
|
"bezeichnung": (zeile.get("Bezeichnung") or "").strip(),
|
|
"planquadrat": (zeile.get("Planquadrat") or "").strip(),
|
|
"mitte": {"x": mitte[0], "y": mitte[1], "z": mitte[2]},
|
|
"masse": {"dx": masse[0], "dy": masse[1], "dz": masse[2]},
|
|
"merkmale": merkmale,
|
|
"insertpoint": decode_insertpoint(zeile.get("Insertpoint")),
|
|
"ks_ein": decode_trans(zeile.get("K1")),
|
|
"ks_aus": decode_trans(zeile.get("K2")),
|
|
}
|
|
element["footprint"] = _footprint(element)
|
|
element["z_bereich"] = (mitte[2] - masse[2] / 2.0, mitte[2] + masse[2] / 2.0)
|
|
element["achse"] = "x" if masse[0] >= masse[1] else "y"
|
|
element["enden"] = _enden(element)
|
|
elemente.append(element)
|
|
return elemente
|
|
|
|
|
|
def _footprint(element):
|
|
"""Achsparallele Grundflaeche (minx, maxx, miny, maxy) in mm.
|
|
|
|
Die Boundingbox aus dem Export ist bereits WCS-achsparallel (sie kommt
|
|
von vla-getboundingbox), die Drehung ist darin also enthalten - es muss
|
|
hier nichts mehr rotiert werden.
|
|
"""
|
|
m, d = element["mitte"], element["masse"]
|
|
return (m["x"] - d["dx"] / 2.0, m["x"] + d["dx"] / 2.0,
|
|
m["y"] - d["dy"] / 2.0, m["y"] + d["dy"] / 2.0)
|
|
|
|
|
|
def _enden(element):
|
|
"""Anschlusspunkte eines Elements im Grundriss.
|
|
|
|
BTMT Beladung/Entladung haben nur EINEN Anschluss zum Foerdersystem
|
|
(Elementmitte) - anders als eine Durchlaufstrecke keine zwei Enden
|
|
einer Laengsachse. Mit der generischen Zwei-Enden-Logik wuerden sie
|
|
faelschlich als Durchlaufelement mit zwei Uebergabestellen behandelt
|
|
(siehe derive_flow/BTMT_ARTEN).
|
|
"""
|
|
if element["teileart"] in BTMT_ARTEN:
|
|
m = element["mitte"]
|
|
return {"a": (m["x"], m["y"])}
|
|
minx, maxx, miny, maxy = element["footprint"]
|
|
m = element["mitte"]
|
|
if element["achse"] == "x":
|
|
return {"a": (minx, m["y"]), "b": (maxx, m["y"])}
|
|
return {"a": (m["x"], miny), "b": (m["x"], maxy)}
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Verbindungen ermitteln
|
|
# ------------------------------------------------------------------
|
|
|
|
def _abstand_zu_flaeche(punkt, footprint):
|
|
"""Kuerzester Abstand eines Punktes zu einem Rechteck (0 = innerhalb)."""
|
|
px, py = punkt
|
|
minx, maxx, miny, maxy = footprint
|
|
dx = max(minx - px, 0.0, px - maxx)
|
|
dy = max(miny - py, 0.0, py - maxy)
|
|
return math.hypot(dx, dy)
|
|
|
|
|
|
def find_connections(elemente, toleranz_mm=PORT_TOLERANZ_MM):
|
|
"""Ungerichtete Anschluesse: Ende von A liegt auf der Flaeche von B.
|
|
|
|
Rueckgabe: Liste von dicts {"a": idx, "a_port": "a"|"b", "b": idx,
|
|
"abstand": mm}.
|
|
|
|
Auch Kreisel werden als Ausgangspunkt geprueft: ein Kreisel kann mit
|
|
seinem Bahnende an einen anderen Kreisel anschliessen (Stichbahn). Das
|
|
ist unkritisch, weil hier nur ENDPUNKTE gegen Flaechen getestet werden -
|
|
die flaechige Ueberdeckung eines 20 m langen Kreisels, die die Spalte
|
|
"Nachbarn" verfaelscht, spielt dabei keine Rolle.
|
|
"""
|
|
anschluesse = []
|
|
for idx_a, el_a in enumerate(elemente):
|
|
for port, punkt in el_a["enden"].items():
|
|
beste = None
|
|
for idx_b, el_b in enumerate(elemente):
|
|
if idx_a == idx_b:
|
|
continue
|
|
abstand = _abstand_zu_flaeche(punkt, el_b["footprint"])
|
|
if abstand > toleranz_mm:
|
|
continue
|
|
if beste is None or abstand < beste["abstand"]:
|
|
beste = {"a": idx_a, "a_port": port, "b": idx_b,
|
|
"abstand": round(abstand, 2)}
|
|
if beste:
|
|
anschluesse.append(beste)
|
|
return anschluesse
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Richtung ableiten
|
|
# ------------------------------------------------------------------
|
|
|
|
def _partner_hoehe(elemente, idx):
|
|
"""Referenzhoehe eines Elements (Mitte der Z-Ausdehnung, in mm)."""
|
|
return elemente[idx]["mitte"]["z"]
|
|
|
|
|
|
def derive_flow(elemente, anschluesse):
|
|
"""Aus den ungerichteten Anschluessen gerichtete Flusskanten machen.
|
|
|
|
Ein Streckenelement (Gefaellestrecke/Foerderer) hat zwei Anschluesse.
|
|
Die Richtung wird ueber die Hoehe der beiden Partner bestimmt:
|
|
* Gefaellestrecke -> vom hoeheren zum tieferen Partner (Schwerkraft).
|
|
* Foerderer mit Antriebfahrtrichtung "Auf" -> vom tieferen zum hoeheren,
|
|
"Ab" -> umgekehrt.
|
|
Liegen K1/K2 vor, gewinnt diese exakte Angabe.
|
|
|
|
BTMT Beladung/Entladung sind die Ein-/Ausgaenge des gesamten Graphen
|
|
(siehe BTMT_ARTEN/_enden): ihr einziger Anschluss wird FEST gerichtet
|
|
(Beladung -> Partner, Partner -> Entladung), unabhaengig von Hoehe/
|
|
Antriebfahrtrichtung - eine Beladestation speist immer ein, eine
|
|
Entladestation nimmt immer aus, nie umgekehrt.
|
|
"""
|
|
# Anschluesse je Streckenelement sammeln
|
|
je_element = {}
|
|
for anschluss in anschluesse:
|
|
je_element.setdefault(anschluss["a"], []).append(anschluss)
|
|
|
|
kanten = []
|
|
for idx, liste in sorted(je_element.items()):
|
|
el = elemente[idx]
|
|
art = el["teileart"]
|
|
|
|
if art in BTMT_BELADUNG_ARTEN:
|
|
for anschluss in liste:
|
|
kanten.append(_kante(idx, anschluss["b"], "btmt", anschluss["abstand"]))
|
|
continue
|
|
if art in BTMT_ENTLADUNG_ARTEN:
|
|
for anschluss in liste:
|
|
kanten.append(_kante(anschluss["b"], idx, "btmt", anschluss["abstand"]))
|
|
continue
|
|
|
|
# Exakte Richtung, sobald der Export K1/K2 liefert
|
|
if el["ks_ein"] and el["ks_aus"]:
|
|
kanten.extend(_kanten_aus_ks(elemente, idx, liste))
|
|
continue
|
|
|
|
if len(liste) < 2:
|
|
# Nur ein Anschluss -> Richtung nicht bestimmbar, aber die
|
|
# Verbindung selbst ist real und darf nicht verloren gehen.
|
|
for anschluss in liste:
|
|
kanten.append(_kante(idx, anschluss["b"], "unbestimmt",
|
|
anschluss["abstand"], gerichtet=False))
|
|
continue
|
|
|
|
# Zwei (oder mehr) Anschluesse: nach Partnerhoehe sortieren
|
|
liste = sorted(liste, key=lambda a: _partner_hoehe(elemente, a["b"]))
|
|
tief, hoch = liste[0], liste[-1]
|
|
h_tief = _partner_hoehe(elemente, tief["b"])
|
|
h_hoch = _partner_hoehe(elemente, hoch["b"])
|
|
|
|
richtung_quelle = "unbestimmt"
|
|
von, nach = None, None
|
|
|
|
if art in GEFAELLE_ARTEN and h_hoch > h_tief:
|
|
von, nach = hoch, tief # Schwerkraft: hoch -> tief
|
|
richtung_quelle = "gefaelle"
|
|
elif art in STRECKEN_ARTEN:
|
|
fahrt = (el["merkmale"].get("Antriebfahrtrichtung") or "").strip().lower()
|
|
if fahrt.startswith("auf") and h_hoch > h_tief:
|
|
von, nach = tief, hoch
|
|
richtung_quelle = "antrieb"
|
|
elif fahrt.startswith("ab") and h_hoch > h_tief:
|
|
von, nach = hoch, tief
|
|
richtung_quelle = "antrieb"
|
|
|
|
if von is None:
|
|
# Richtung unklar -> beide Anschluesse ungerichtet ausgeben
|
|
for anschluss in liste:
|
|
kanten.append(_kante(idx, anschluss["b"], "unbestimmt",
|
|
anschluss["abstand"], gerichtet=False))
|
|
continue
|
|
|
|
# Fluss laeuft Partner -> Element -> Partner
|
|
kanten.append(_kante(von["b"], idx, richtung_quelle, von["abstand"]))
|
|
kanten.append(_kante(idx, nach["b"], richtung_quelle, nach["abstand"]))
|
|
|
|
return kanten
|
|
|
|
|
|
def _kanten_aus_ks(elemente, idx, liste):
|
|
"""Kanten aus den exakten KS_EIN/KS_AUS-Positionen (Spalten K1/K2)."""
|
|
el = elemente[idx]
|
|
ein, aus = el["ks_ein"][:2], el["ks_aus"][:2]
|
|
kanten = []
|
|
for anschluss in liste:
|
|
partner = elemente[anschluss["b"]]
|
|
d_ein = _abstand_zu_flaeche(ein, partner["footprint"])
|
|
d_aus = _abstand_zu_flaeche(aus, partner["footprint"])
|
|
if d_ein <= d_aus:
|
|
kanten.append(_kante(anschluss["b"], idx, "ks", anschluss["abstand"]))
|
|
else:
|
|
kanten.append(_kante(idx, anschluss["b"], "ks", anschluss["abstand"]))
|
|
return kanten
|
|
|
|
|
|
def _kante(von, nach, quelle, abstand, gerichtet=True):
|
|
return {"von": von, "nach": nach, "richtung_quelle": quelle,
|
|
"abstand_mm": abstand, "gerichtet": gerichtet}
|
|
|
|
|
|
def _dedupe(kanten):
|
|
"""Doppelte Kanten entfernen.
|
|
|
|
Dieselbe Uebergabestelle wird von beiden Seiten gefunden (z.B. Kreisel A
|
|
findet das Ende von Kreisel B und umgekehrt). Gerichtete Kanten werden
|
|
nach (von, nach) dedupliziert, ungerichtete nach dem unsortierten
|
|
Knotenpaar - sonst erscheint eine Stichbahn zweimal. Von mehreren
|
|
Treffern gewinnt der mit dem kleinsten Abstand.
|
|
"""
|
|
beste = {}
|
|
for kante in kanten:
|
|
if kante["gerichtet"]:
|
|
schluessel = ("g", kante["von"], kante["nach"])
|
|
else:
|
|
schluessel = ("u",) + tuple(sorted((kante["von"], kante["nach"])))
|
|
vorhanden = beste.get(schluessel)
|
|
if vorhanden is None or kante["abstand_mm"] < vorhanden["abstand_mm"]:
|
|
beste[schluessel] = kante
|
|
return list(beste.values())
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Ausgabe
|
|
# ------------------------------------------------------------------
|
|
|
|
def _label(el):
|
|
name = el["bezeichnung"] or el["teileart"]
|
|
return f"{name} [{el['teileid']}]"
|
|
|
|
|
|
def build_graph(elemente, toleranz_mm=PORT_TOLERANZ_MM):
|
|
"""Vollstaendiges Graph-Modell (Knoten + Kanten + Kennzahlen)."""
|
|
anschluesse = find_connections(elemente, toleranz_mm)
|
|
kanten = derive_flow(elemente, anschluesse)
|
|
|
|
kanten = _dedupe(kanten)
|
|
|
|
knoten = []
|
|
for idx, el in enumerate(elemente):
|
|
knoten.append({
|
|
"id": idx,
|
|
"teileid": el["teileid"],
|
|
"teileart": el["teileart"],
|
|
"bezeichnung": el["bezeichnung"],
|
|
"planquadrat": el["planquadrat"],
|
|
"rolle": _rolle(el),
|
|
"position": el["mitte"],
|
|
"masse": el["masse"],
|
|
"hoehe_z": el["mitte"]["z"],
|
|
"merkmale": el["merkmale"],
|
|
})
|
|
|
|
eingang = {}
|
|
ausgang = {}
|
|
offen = {}
|
|
for kante in kanten:
|
|
if kante["gerichtet"]:
|
|
ausgang[kante["von"]] = ausgang.get(kante["von"], 0) + 1
|
|
eingang[kante["nach"]] = eingang.get(kante["nach"], 0) + 1
|
|
else:
|
|
# Ungerichtete Verbindung zaehlt bei beiden Knoten, damit ein nur
|
|
# ungerichtet angebundenes Element nicht wie isoliert aussieht.
|
|
for seite in (kante["von"], kante["nach"]):
|
|
offen[seite] = offen.get(seite, 0) + 1
|
|
for n in knoten:
|
|
n["grad_ein"] = eingang.get(n["id"], 0)
|
|
n["grad_aus"] = ausgang.get(n["id"], 0)
|
|
n["grad_offen"] = offen.get(n["id"], 0)
|
|
|
|
return {
|
|
"quelle": "EXPORTCSV",
|
|
"toleranz_mm": toleranz_mm,
|
|
"knoten": knoten,
|
|
"kanten": kanten,
|
|
"kennzahlen": {
|
|
"knoten": len(knoten),
|
|
"kanten": len(kanten),
|
|
"gerichtet": sum(1 for k in kanten if k["gerichtet"]),
|
|
"unbestimmt": sum(1 for k in kanten if not k["gerichtet"]),
|
|
},
|
|
}
|
|
|
|
|
|
def _rolle(el):
|
|
"""Funktion des Elements im Materialfluss."""
|
|
if el["teileart"] in KREISEL_ARTEN:
|
|
return "Verteiler"
|
|
if el["teileart"] in GEFAELLE_ARTEN:
|
|
return "Schwerkraftstrecke"
|
|
if el["teileart"] in STRECKEN_ARTEN:
|
|
return "Foerderer"
|
|
if el["teileart"] in BTMT_BELADUNG_ARTEN:
|
|
return "Eingang"
|
|
if el["teileart"] in BTMT_ENTLADUNG_ARTEN:
|
|
return "Ausgang"
|
|
return "Element"
|
|
|
|
|
|
STIL = {
|
|
"Verteiler": ("box3d", "#1f6feb"),
|
|
"Schwerkraftstrecke": ("box", "#8250df"),
|
|
"Foerderer": ("box", "#1a7f37"),
|
|
"Eingang": ("invhouse", "#bf8700"),
|
|
"Ausgang": ("house", "#bf8700"),
|
|
"Element": ("ellipse", "#57606a"),
|
|
}
|
|
|
|
|
|
def to_dot(graph):
|
|
"""GraphViz-DOT. Kanten ohne bestimmte Richtung werden gestrichelt."""
|
|
zeilen = ["digraph Materialfluss {", " rankdir=LR;",
|
|
' node [style="filled,rounded" fontname="Segoe UI" fontsize=10 fillcolor="#f6f8fa"];',
|
|
' edge [fontname="Segoe UI" fontsize=8];']
|
|
for n in graph["knoten"]:
|
|
form, farbe = STIL[n["rolle"]]
|
|
beschriftung = f"{n['bezeichnung']}\\n{n['teileart']}\\nz={n['hoehe_z']:.0f} mm"
|
|
zeilen.append(f' n{n["id"]} [label="{beschriftung}" shape={form} color="{farbe}"];')
|
|
for k in graph["kanten"]:
|
|
if k["gerichtet"]:
|
|
zeilen.append(f' n{k["von"]} -> n{k["nach"]} '
|
|
f'[label="{k["richtung_quelle"]}"];')
|
|
else:
|
|
zeilen.append(f' n{k["von"]} -> n{k["nach"]} '
|
|
f'[dir=none style=dashed color="#8c959f"];')
|
|
zeilen.append("}")
|
|
return "\n".join(zeilen)
|
|
|
|
|
|
def to_mermaid(graph):
|
|
"""Mermaid-Flowchart (fuer Doku/Markdown)."""
|
|
zeilen = ["flowchart LR"]
|
|
for n in graph["knoten"]:
|
|
text = f"{n['bezeichnung']}<br/>z={n['hoehe_z']:.0f}"
|
|
if n["rolle"] == "Verteiler":
|
|
zeilen.append(f' n{n["id"]}[[{text}]]')
|
|
else:
|
|
zeilen.append(f' n{n["id"]}[{text}]')
|
|
for k in graph["kanten"]:
|
|
pfeil = "-->" if k["gerichtet"] else "---"
|
|
zeilen.append(f' n{k["von"]} {pfeil} n{k["nach"]}')
|
|
return "\n".join(zeilen)
|
|
|
|
|
|
def report_saturation(elemente):
|
|
"""Warnt, wenn Insertpoint-Werte an der Fixed-Point-Grenze haengen.
|
|
|
|
csv:kos-encode in Lisp/export.lsp rechnet mit Faktor 1000, der 24-Bit-
|
|
Wertebereich endet damit bei +-8388.607 Zeichnungseinheiten. Bei einer
|
|
Zeichnung in mm ist bei +-8.39 m Schluss - jede groessere Koordinate wird
|
|
stillschweigend abgeschnitten. Betroffene Insertpoints sind unbrauchbar.
|
|
"""
|
|
betroffen = [el for el in elemente
|
|
if el["insertpoint"] and el["insertpoint"]["gesaettigt"]]
|
|
return betroffen
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Kommandozeile
|
|
# ------------------------------------------------------------------
|
|
|
|
def _flusspfade(graph):
|
|
"""Zusammengefasste Flusspfade: je Verteiler die abgehenden Strecken."""
|
|
nach_id = {n["id"]: n for n in graph["knoten"]}
|
|
gruppen = {}
|
|
for kante in graph["kanten"]:
|
|
if not kante["gerichtet"]:
|
|
continue
|
|
von, nach = nach_id[kante["von"]], nach_id[kante["nach"]]
|
|
if von["rolle"] == "Verteiler":
|
|
gruppen.setdefault((von["bezeichnung"], nach["teileart"]), []).append(nach)
|
|
return gruppen
|
|
|
|
|
|
def main(argv):
|
|
if len(argv) < 2:
|
|
print(__doc__)
|
|
return 2
|
|
|
|
csv_pfad = argv[1]
|
|
basis = argv[2] if len(argv) > 2 else os.path.join(
|
|
os.path.dirname(os.path.abspath(csv_pfad)), "export_graph")
|
|
|
|
elemente = load_elements(csv_pfad)
|
|
if not elemente:
|
|
print(f"[export_graph] Keine Elemente in {csv_pfad} gefunden.")
|
|
return 1
|
|
print(f"[export_graph] {len(elemente)} Elemente aus {csv_pfad} geladen.")
|
|
|
|
gesaettigt = report_saturation(elemente)
|
|
if gesaettigt:
|
|
print(f"[export_graph] WARNUNG: {len(gesaettigt)} von {len(elemente)} "
|
|
f"Insertpoint-Werten sind an der Fixed-Point-Grenze abgeschnitten "
|
|
f"(+-{KOS_MAX / KOS_FAKTOR:.3f} Einheiten).")
|
|
print("[export_graph] Betroffene Achsen werden nicht verwendet; "
|
|
"der Graph nutzt Position/Boundingbox.")
|
|
|
|
ohne_ks = [el for el in elemente if not (el["ks_ein"] and el["ks_aus"])]
|
|
if ohne_ks:
|
|
print(f"[export_graph] HINWEIS: {len(ohne_ks)} Elemente ohne K1/K2 "
|
|
f"(KS_EIN/KS_AUS) - Richtung wird heuristisch abgeleitet.")
|
|
|
|
graph = build_graph(elemente)
|
|
kz = graph["kennzahlen"]
|
|
print(f"[export_graph] Graph: {kz['knoten']} Knoten, {kz['kanten']} Kanten "
|
|
f"({kz['gerichtet']} gerichtet, {kz['unbestimmt']} ohne Richtung).")
|
|
|
|
with open(basis + ".json", "w", encoding="utf-8") as fh:
|
|
json.dump(graph, fh, indent=2, ensure_ascii=False)
|
|
with open(basis + ".dot", "w", encoding="utf-8") as fh:
|
|
fh.write(to_dot(graph))
|
|
with open(basis + ".mmd", "w", encoding="utf-8") as fh:
|
|
fh.write(to_mermaid(graph))
|
|
for endung in (".json", ".dot", ".mmd"):
|
|
print(f"[export_graph] geschrieben: {basis}{endung}")
|
|
|
|
# Kurzuebersicht des ermittelten Flusses
|
|
print("\n=== Materialfluss (zusammengefasst) ===")
|
|
for (verteiler, art), ziele in sorted(_flusspfade(graph).items()):
|
|
print(f" {verteiler} --> {len(ziele)}x {art}")
|
|
print("\n=== Verteiler ===")
|
|
for n in graph["knoten"]:
|
|
if n["rolle"] != "Verteiler":
|
|
continue
|
|
print(f" {n['bezeichnung']:<12} z={n['hoehe_z']:7.1f} mm "
|
|
f"ein={n['grad_ein']:>2} aus={n['grad_aus']:>2} "
|
|
f"ohne Richtung={n['grad_offen']:>2}")
|
|
|
|
isoliert = [n for n in graph["knoten"]
|
|
if not (n["grad_ein"] or n["grad_aus"] or n["grad_offen"])]
|
|
if isoliert:
|
|
print(f"\n[export_graph] WARNUNG: {len(isoliert)} Element(e) ohne jede "
|
|
f"Verbindung: " + ", ".join(n["bezeichnung"] for n in isoliert))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main(sys.argv))
|