mistangl/sps_skel
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Materialfluss: VF/GF-Richtung, Stations-Bahn und Eingang/Ausgang-Merge

Browse Source 
- _apply_transfers liest die andockende Strecke auch aus dem Merkmal
  "Foerderer"/"Förderer" (nicht nur "Strecke"), damit VF/GF gerichtet
  Kreisel->AS->..->ES->Kreisel gezeichnet werden (Ruecklauf = zwei Pfeile
  auf die beiden Kreisel-Bahnen) statt "unbestimmt".

- _attach_stations haengt BTMT/Eingang/Ausgang an die Kreisel-Bahn, in
  deren TrackIds-Umlauf sie liegen (Bahn aus der Drehrichtung); Nachbarn,
  die keine Flussobjekte sind, werden uebersprungen - keine
  Station<->Station-Kanten mehr.

- Eingang/Ausgang sind die generischen Symbole fuer Beladung/Entladung.
  Liegt ein Symbol deckungsgleich (<= STATION_MERGE_MM = 600 mm) ueber
  einer BTMT-Station gleicher Rolle, wird es verschluckt
  (_merge_station_symbols) - nur ein Knoten, in grobem Graph UND --track
  (dort auch aus den TrackIds gestrichen). Einzelne Symbole bleiben eigener
  gruener/roter Knoten.

- drive_uphill: die Hoehe bestimmt die Richtung NUR bei einer reinen
  Gefaellestrecke (Schwerkraft, bergab). Ein Foerderer/gemischte Strecke
  kann hoch oder runter foerdern - dort zaehlt nur eine eindeutige
  Antriebfahrtrichtung, sonst kommt die Richtung aus AS/ES.

Doku in doc/Python_Scripts.md und doc/track.md nachgezogen.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Michael Stangl
2026-09-17 12:54:44 +02:00
co-authored by Claude Opus 4.8
parent 516f30278b
commit 117cd8914b
3 changed files with 222 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/material_flow.py
+181 -56
View File
@@ -58,7 +58,7 @@ import re
import shutil
import subprocess
import sys
from dataclasses import dataclass, field
from dataclasses import dataclass, field, replace
from datetime import datetime
from pathlib import Path
@@ -86,22 +86,35 @@ CIRCLE_KINDS = ("Kreisel",)
# Objektarten, die per "Zuordnung" an einem Knoten haengen (kein eigener Knoten)
DEVICE_KINDS = ("Separator", "Scanner")
# Ein-/Ausgabestationen (BTMT Be-/Entladung) = Materialquelle/-senke, d.h. die
# Ein-/Ausgangspunkte des Materialflusses. Sie werden in sps_skel erkannt;
# dxfmakros exportiert sie derzeit NICHT (Blockmuster fehlt), die Erkennung
# greift also, sobald eine passende Zeile im CSV auftaucht - erkannt an
# TeileArt/Bezeichnung: "Beladung"/"BTMT-Beladung" -> Eingang, "Entladung"/
# "SC_Entladung"/"Abwurf" -> Ausgang.
ENTRY_KINDS = ("Beladung",) # BTMT-Beladung -> Materialeingang (Quelle)
EXIT_KINDS = ("Entladung",) # SC_Entladung -> Materialausgang (Senke)
# Ein-/Ausgabestationen = Materialquelle/-senke, d.h. die Ein-/Ausgangspunkte
# des Materialflusses. Es gibt sie in zwei Auspraegungen:
# * konkrete BTMT-Station: "Beladung" (Quelle) / "Entladung" (Senke)
# * generisches Symbol: "Eingang" (Quelle) / "Ausgang" (Senke) - der
# allgemeinere Bezeichner fuer dasselbe. Liegt ein Symbol nahezu deckungs-
# gleich ueber der konkreten BTMT-Station (haeufig, gleiche x/y), wird nur
# ein Knoten gezeichnet: die BTMT-Station gewinnt, das Symbol wird
# verschluckt (siehe _merge_station_symbols / STATION_MERGE_MM).
ENTRY_KINDS = ("Beladung", "Eingang") # -> Materialeingang (Quelle)
EXIT_KINDS = ("Entladung", "Ausgang") # -> Materialausgang (Senke)
STATION_KINDS = ENTRY_KINDS + EXIT_KINDS
# Die generischen Symbole (verschluckbar zugunsten der konkreten BTMT-Station).
STATION_SYMBOL_KINDS = ("Eingang", "Ausgang")
# Schluesselwoerter zur Erkennung der Stationen (Kleinschreibung, Teilstring)
# Schluesselwoerter zur Erkennung der Stationen (Kleinschreibung, Teilstring).
# "eingang"/"ausgang" MUESSEN nach den spezifischeren Mustern kommen, damit z.B.
# "BTMT-Beladung" als Beladung (nicht Eingang) klassifiziert wird.
STATION_PATTERNS = (
("Beladung", ("beladung", "beladestation", "btmt-beladung", "btmt beladung")),
("Entladung", ("entladung", "entladestation", "sc_entladung", "abwurf")),
("Eingang", ("eingang",)),
("Ausgang", ("ausgang",)),
)
# Max. Abstand (mm), ab dem ein generisches Eingang/Ausgang-Symbol als
# "deckungsgleich" mit einer BTMT-Station gilt und verschluckt wird. Die
# beobachteten Paare liegen bei 256-431 mm, echte getrennte Stationen >1000 mm.
STATION_MERGE_MM = 600.0
# Transfer-/Weichenelemente aus dem erweiterten Export. Sie bilden KEINE eigenen
# Knoten, liefern aber die *gerichtete* Verbindung zwischen den Transportobjekten:
# Ausschleuselement = Kreisel -> Strecke (am Streckenanfang, Material verlaesst Kreisel)
@@ -157,6 +170,10 @@ NODE_STYLES = {
"Strecke": dict(fillcolor="#e2f0d9", fontcolor="#375623", color="#548235"),
"Beladung": dict(fillcolor="#d4f0d4", fontcolor="#1e5b1e", color="#338833"),
"Entladung": dict(fillcolor="#f4d4d4", fontcolor="#7a1f1f", color="#cc3333"),
# Generische Symbole - gleiche Farbe wie die konkreten BTMT-Stationen
# (gruen = Eingang/Quelle, rot = Ausgang/Senke).
"Eingang": dict(fillcolor="#d4f0d4", fontcolor="#1e5b1e", color="#338833"),
"Ausgang": dict(fillcolor="#f4d4d4", fontcolor="#7a1f1f", color="#cc3333"),
}
DEFAULT_NODE_STYLE = dict(fillcolor="#f2f2f2", fontcolor="#404040", color="#808080")
@@ -694,35 +711,43 @@ def drive_uphill(element: Element) -> tuple[bool | None, str]:
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
Die Hoehe bestimmt die Richtung NUR bei einer reinen Gefaellestrecke - dort
laeuft der Fluss per Definition bergab (Schwerkraft). Ein Foerderer (VF)
kann bauartbedingt nach oben ODER unten foerdern; eine Strecke ist oft eine
gemischte GF/VF-Kette. Fuer beide ist die Hoehendifferenz KEIN Richtungs-
indikator - ihre Richtung kommt aus den Ein-/Ausschleuselementen
(_apply_transfers, AS -> ES) bzw. einer eindeutigen Antriebfahrtrichtung,
sonst bleibt sie unbestimmt.
Der zweite Rueckgabewert nennt die verwendete Quelle fuer Bericht und
Zeichnung.
Bewertet wird daher:
1. Gefaellestrecke: Hoehe_Von/Bis (bergab), sonst per Definition bergab
2. Antriebfahrtrichtung "Auf"/"Ab" (nur wenn eindeutig, nicht bei
gemischten Listen wie "horizontal,horizontal,Auf")
3. sonst None (unbestimmt)
"""
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":
rise = element.rise
if rise is not None and rise != 0:
direction = "Auf" if rise > 0 else "Ab"
return rise > 0, (
f"Gefaellestrecke Hoehe_Von {element.height_from:.0f} mm -> Hoehe_Bis "
f"{element.height_to:.0f} mm ({direction}, {rise:+.0f} mm)"
)
return False, "Gefaellestrecke ohne Hoehenangabe - laeuft bergab"
# Foerderer/Strecke: Hoehe sagt NICHTS ueber die Richtung. Nur eine
# eindeutige Antriebfahrtrichtung (genau "Auf" bzw. "Ab", keine gemischte
# Liste) zaehlt.
drive = element.drive_dir.strip().lower()
if drive == "auf":
return True, f"Antriebfahrtrichtung '{element.drive_dir}' (nach oben)"
if drive == "ab":
return False, f"Antriebfahrtrichtung '{element.drive_dir}' (nach unten)"
return None, (
f"weder Hoehe_Von/Bis noch Antriebfahrtrichtung verwertbar "
f"(Antrieb '{element.drive_dir}', Hoehen "
f"{element.height_from}/{element.height_to})"
f"Foerderer/Strecke - Hoehe ist kein Richtungsindikator und "
f"Antriebfahrtrichtung '{element.drive_dir or '-'}' nicht eindeutig "
f"(Richtung kommt sonst aus AS/ES)"
)
@@ -840,21 +865,34 @@ def _nearest_kreisel(station: Element, elements: list[Element]) -> Element | Non
return best
def _kreisel_containing(station_id: str, elements: list[Element]) -> Element | None:
"""Kreisel, in dessen TrackIds-Umlauf die Station liegt (oder None)."""
for el in elements:
if el.is_circle and station_id in el.track_ids:
return el
return None
def _attach_stations(
elements: list[Element], by_id: dict[str, Element], nodes: dict[str, Node],
warnings: list[str]
) -> list[Edge]:
"""
Ein-/Ausgabestationen (BTMT Be-/Entladung) mit dem Fluss verbinden.
Ein-/Ausgabestationen (BTMT Be-/Entladung, Eingang/Ausgang) mit dem Fluss
verbinden.
Eingang (Beladung) ist Quelle: Kante Station -> Ziel.
Ausgang (Entladung) ist Senke: Kante Ziel -> Station.
Ziel kommt aus den 'Nachbarn' des Exports; fehlen sie (dxfmakros exportiert
fuer BTMT derzeit keine), haengen beide baulich direkt am naechstgelegenen
Kreisel, auf dessen Eingangs- bzw. Ausgangsbahn (siehe _weiche_lane_roles):
UZS -> Eingang Bahn R / Ausgang Bahn L, GUZ -> Eingang Bahn L / Ausgang
Bahn R. Nur wenn kein Kreisel mit bekannter Drehrichtung gefunden wird,
greift ersatzweise die naechstgelegene Bahn/Strecke per Abstand.
Eingang (Beladung) ist Quelle: Kante Station -> Kreisel-Bahn.
Ausgang (Entladung) ist Senke: Kante Kreisel-Bahn -> Station.
Die Zielbahn kommt in dieser Reihenfolge:
1. Der Kreisel, in dessen TrackIds-Umlauf die Station liegt (verlaesslich,
siehe Element.track_ids) - Eingangs-/Ausgangsbahn aus der Drehrichtung
(_weiche_lane_roles: UZS -> Eingang R / Ausgang L, GUZ umgekehrt).
2. Ersatzweise: ein Flussobjekt aus den 'Nachbarn' (Separator/Station als
Nachbar werden bewusst uebersprungen - sonst haengen zwei BTMT-Punkte
aneinander statt an der Bahn).
3. Ersatzweise: der naechstgelegene Kreisel per Abstand, sonst der
naechste Flussknoten.
"""
edges: list[Edge] = []
for element in elements:
@@ -863,38 +901,66 @@ def _attach_stations(
sid = element.teile_id
if sid not in nodes:
continue
partners: list[str] = []
for base, lane in element.neighbours:
if lane and f"{base}-{lane}" in nodes:
partners.append(f"{base}-{lane}")
elif base in nodes:
partners.append(base)
else:
partners += [f"{base}-{L}" for L in (LANE_LEFT, LANE_RIGHT)
if f"{base}-{L}" in nodes]
# 1. Kreisel aus dem TrackIds-Umlauf + Bahn aus der Drehrichtung.
container = _kreisel_containing(sid, elements)
if container is not None:
roles = _weiche_lane_roles(container.teile_id, by_id, nodes)
if roles is not None:
eingang, ausgang = roles
lane, role = ((eingang, "Eingangsbahn") if element.is_entry
else (ausgang, "Ausgangsbahn"))
partners = [lane]
warnings.append(
f"{element.describe()}: im Umlauf von Kreisel "
f"{container.teile_id} - an Bahn {lane.split('-')[-1]} "
f"angebunden ({role}, Drehrichtung {container.rotation})"
)
# 2. Nachbarn, aber nur echte Flussobjekte (kein Separator/Station).
if not partners:
for base, lane in element.neighbours:
partner_el = by_id.get(base)
if partner_el is None or not (partner_el.is_transport or partner_el.is_circle):
continue
if lane and f"{base}-{lane}" in nodes:
partners.append(f"{base}-{lane}")
elif base in nodes:
partners.append(base)
else:
partners += [f"{base}-{L}" for L in (LANE_LEFT, LANE_RIGHT)
if f"{base}-{L}" in nodes]
# 3. Naechster Kreisel per Abstand (Bahn aus Drehrichtung), sonst
# naechster Flussknoten.
if not partners and (element.is_entry or element.is_exit):
kreisel = _nearest_kreisel(element, elements)
roles = _weiche_lane_roles(kreisel.teile_id, by_id, nodes) if kreisel else None
if roles:
eingang, ausgang = roles
lane, role = (eingang, "Eingangsbahn") if element.is_entry else (ausgang, "Ausgangsbahn")
lane, role = ((eingang, "Eingangsbahn") if element.is_entry
else (ausgang, "Ausgangsbahn"))
partners = [lane]
warnings.append(
f"{element.describe()}: keine Nachbarn im Export - "
f"an Kreisel {kreisel.teile_id} Bahn {lane.split('-')[-1]} "
f"angebunden ({role}, Drehrichtung {kreisel.rotation})"
f"{element.describe()}: nicht in einem TrackIds-Umlauf - "
f"an naechsten Kreisel {kreisel.teile_id} Bahn "
f"{lane.split('-')[-1]} angebunden ({role}, Drehrichtung "
f"{kreisel.rotation})"
)
if not partners:
nearest = _nearest_node(element, nodes)
if nearest:
partners = [nearest]
warnings.append(
f"{element.describe()}: keine Nachbarn im Export - "
f"{element.describe()}: kein Umlauf/Flussnachbar - "
f"naechstes Objekt '{nearest}' per Abstand verbunden"
)
else:
warnings.append(f"{element.describe()}: kein Ziel gefunden - unverbunden")
continue
for partner in partners:
if element.is_entry:
edges.append(Edge(sid, partner, EDGE_FLOW))
@@ -963,7 +1029,9 @@ def _apply_transfers(
for el in elements:
if el.kind in ("Ausschleus", "Einschleus"):
k = str(_merkmal(el.merkmale, "Kreisel") or "").strip()
s = str(_merkmal(el.merkmale, "Strecke") or "").strip()
# Der Export benennt die andockende Strecke je nach Bauart als
# "Strecke" (VF) oder "Foerderer"/"Förderer" (GF/Streckengruppe).
s = str(_merkmal(el.merkmale, "Strecke", "Foerderer", "Förderer") or "").strip()
if k and s:
role = "aus" if el.kind == "Ausschleus" else "ein"
per_strecke.setdefault(s, {})[role] = k
@@ -1067,7 +1135,52 @@ def _apply_transfers(
return result
def _merge_station_symbols(
elements: list[Element], warnings: list[str] | None = None
) -> set[str]:
"""
Generische Eingang/Ausgang-Symbole ermitteln, die deckungsgleich ueber einer
konkreten BTMT-Station liegen (gleiche Rolle Quelle/Senke, Abstand
<= STATION_MERGE_MM). Diese werden verschluckt: es bleibt nur die
BTMT-Station als Knoten. Rueckgabe = TeileIds der verschluckten Symbole.
Eingang gehoert zu Beladung (beide Quelle), Ausgang zu Entladung (beide
Senke). Ein Symbol ohne passende nahe BTMT-Station bleibt ein eigener Knoten.
warnings=None unterdrueckt die Meldungen (zweiter Aufruf im --track-Pfad,
der sonst dieselben Hinweise ein zweites Mal ausgibt).
"""
swallowed: set[str] = set()
btmt = [e for e in elements if e.is_station and e.kind not in STATION_SYMBOL_KINDS]
for sym in elements:
if sym.kind not in STATION_SYMBOL_KINDS or sym.x is None or sym.y is None:
continue
sym_is_entry = sym.kind in ENTRY_KINDS
best, best_d = None, None
for st in btmt:
if st.x is None or st.y is None:
continue
if st.is_entry != sym_is_entry: # nur gleiche Rolle (Quelle/Senke)
continue
d = ((st.x - sym.x) ** 2 + (st.y - sym.y) ** 2) ** 0.5
if d <= STATION_MERGE_MM and (best_d is None or d < best_d):
best, best_d = st, d
if best is not None:
swallowed.add(sym.teile_id)
if warnings is not None:
warnings.append(
f"{sym.describe()}: deckungsgleich mit {best.kind} {best.teile_id} "
f"({best_d:.0f} mm) - Symbol verschluckt, nur {best.kind} gezeichnet"
)
return swallowed
def build_graph(elements: list[Element], source: str, warnings: list[str]) -> Graph:
# Generische Eingang/Ausgang-Symbole, die ueber einer BTMT-Station liegen,
# verschlucken (nur die konkrete Station bleibt Knoten).
swallowed = _merge_station_symbols(elements, warnings)
if swallowed:
elements = [e for e in elements if e.teile_id not in swallowed]
by_id = {element.teile_id: element for element in elements}
nodes = _build_nodes(elements, warnings)
unassigned = _attach_devices(elements, nodes, warnings)
@@ -1279,6 +1392,18 @@ def build_track_graph(
warnings wird mit build_graph() geteilt, damit Fein-Hinweise im selben
Bericht erscheinen.
"""
# Deckungsgleiche Eingang/Ausgang-Symbole (wie im groben Graph) verschlucken:
# aus der Element-Liste UND aus den TrackIds jedes Containers entfernen, sonst
# laeuft der Kreisel-Umlauf durch zwei nahezu identische Knoten (Symbol +
# BTMT-Station). warnings=None: die Meldung stand schon im groben Graph.
swallowed = _merge_station_symbols(elements, warnings=None)
if swallowed:
elements = [
replace(e, track_ids=[t for t in e.track_ids if t not in swallowed])
for e in elements if e.teile_id not in swallowed
]
by_id = {e.teile_id: e for e in elements}
track = TrackGraph(source=graph.source)
seen_edges: set[tuple[str, str]] = set()