#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ export_graph.py - Baut aus dem CSV-Export (EXPORTCSV) einen Materialfluss-Graph. Aufruf: python export_graph.py [] Erzeugt neben .json (vollstaendiges Graph-Modell) auch .dot (GraphViz) und .mmd (Mermaid) zur Visualisierung. KNOTEN = ein Anlagenelement je CSV-Zeile (Kreisel, Gefaellestrecke, Strecke). 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. Jede Kante traegt daher ein Feld "richtung_quelle": "ks" (exakt, aus K1/K2), "gefaelle", "antrieb" oder "unbestimmt". """ 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"} # ------------------------------------------------------------------ # 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): """Die beiden Enden der Laengsachse als Anschlusspunkte im Grundriss.""" 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. """ # 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"] # 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" return "Element" STIL = { "Verteiler": ("box3d", "#1f6feb"), "Schwerkraftstrecke": ("box", "#8250df"), "Foerderer": ("box", "#1a7f37"), "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']}
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))