[FIX] Materialfluss: Kreisel als geschlossener Umlauf

Jeder Kreisel bekam nur EINE Bahnkante aus der Drehrichtung (UZS R->L,
GUZ L->R). Die Gegenbahn hatte damit keinen Abgang bzw. Zugang, und der
Sackgassen-Check meldete an fast jedem Kreisel "Sackgasse"/"nicht
erreichbar", obwohl die Anlage durchlaufbar ist.

- build_graph: immer L->R UND R->L; die Gegenkante traegt
  CIRCLE_RETURN_NOTE und wird in render_dot nicht gezeichnet
- Sackgassen-Pruefung ersetzt durch Erreichbarkeit im Gesamtgraphen
  (von Beladung erreichbar / Entladung erreichbar); unbestimmte Kanten
  zaehlen in beide Richtungen
- Weiche: Merkmal "Beruehrung" (Kreis/Kreis bzw. Tangente) hat Vorrang
  vor der Geometrie, die am Kappenuebergang per Rundung entschied
- Test test_kreisel_umlauf_graph.py mit Fixture test_system_export.csv

TRO-Graph (nach Hosts) fuer dematic/hundm_05/mubea/showroom unveraendert,
HundM05 komplett unveraendert.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Samer Ayadi
2026-10-15 15:40:53 +02:00
co-authored by Claude Opus 5.5
parent 550bdeb464
commit c4122b4497
3 changed files with 243 additions and 28 deletions
+95 -28
View File
@@ -186,6 +186,10 @@ NEIGHBOUR_RE = re.compile(r"^(?P<base>[^\s-]+)(?:-(?P<lane>[LR]))?$")
EDGE_FLOW = "flow"
EDGE_CIRCLE = "circle"
EDGE_UNRESOLVED = "unresolved"
# Notiz der Gegenkante eines Kreisel-Umlaufs (EDGE_CIRCLE entgegen der
# Hauptkante, siehe build_graph) - fuer den Fluss gleichwertig, wird aber
# nicht gezeichnet.
CIRCLE_RETURN_NOTE = "Rueckfuehrung"
# Darstellung je Objektart
NODE_STYLES = {
@@ -1330,7 +1334,21 @@ def _apply_transfers(
touch_a = _kreisel_touch_lane(el_a, point, elements) if el_a and point else None
touch_b = _kreisel_touch_lane(el_b, point, elements) if el_b and point else None
if touch_a is not None and touch_b is not None and (not touch_a[1] or not touch_b[1]):
# Beruehrart aus dem Export (dxfmakros, Merkmal "Beruehrung") hat
# Vorrang vor der Geometrie: ein Kreis/Kreis-Punkt liegt genau auf dem
# Uebergang Endkappe -> Gerade, dort entschied in _kreisel_touch_lane
# die Rundung der CSV-Werte (test_system: 4891.79 gegen Schwelle
# 4891.79) und machte die Weiche faelschlich "unbestimmt".
beruehrung = str(_merkmal(weiche_el.merkmale, "Beruehrung") or "").strip()
if beruehrung == "Kreis/Kreis":
laengsseite = False
elif "Tangente" in beruehrung:
laengsseite = True
else:
laengsseite = (touch_a is not None and touch_b is not None
and (not touch_a[1] or not touch_b[1]))
if laengsseite and touch_a is not None and touch_b is not None:
# Mindestens eine Seite liegt an der Laengsseite: Bahn direkt aus
# der Geometrie, Richtung bleibt unbestimmt (ohne Schaltzustand
# der Weiche ist dort kein Eingang/Ausgang ableitbar).
@@ -1473,6 +1491,63 @@ def _merge_station_symbols(
return swallowed
def _reachability_warnings(nodes: dict[str, Node], edges: list[Edge]) -> list[str]:
"""Bahnen, die von keiner Beladung aus erreichbar sind bzw. von denen
aus keine Entladung erreichbar ist.
Ein Kreisel ist ein geschlossener Umlauf (siehe build_graph), eine Bahn
hat darum lokal immer Zu- und Abgang - ob das Material sie erreicht bzw.
wieder verlassen kann, ist eine Frage des ganzen Graphen. Kanten mit
unbestimmter Richtung (EDGE_UNRESOLVED, z. B. eine Weiche an der
Laengsseite) gelten in beide Richtungen als begehbar. Ohne Be- bzw.
Entladestation im Layout entfaellt die jeweilige Pruefung.
"""
succ: dict[str, set[str]] = {}
pred: dict[str, set[str]] = {}
for edge in edges:
succ.setdefault(edge.src, set()).add(edge.dst)
pred.setdefault(edge.dst, set()).add(edge.src)
if edge.kind == EDGE_UNRESOLVED:
succ.setdefault(edge.dst, set()).add(edge.src)
pred.setdefault(edge.src, set()).add(edge.dst)
def reach(starts: list[str], step: dict[str, set[str]]) -> set[str]:
seen = set(starts)
stack = list(starts)
while stack:
for nxt in step.get(stack.pop(), ()):
if nxt not in seen:
seen.add(nxt)
stack.append(nxt)
return seen
entries = sorted(n for n, node in nodes.items()
if node.element.is_station and node.element.is_entry)
exits = sorted(n for n, node in nodes.items()
if node.element.is_station and node.element.is_exit)
from_entry = reach(entries, succ) if entries else None
to_exit = reach(exits, pred) if exits else None
result: list[str] = []
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 from_entry is not None and node_id not in from_entry:
result.append(
f"Bahn {node_id}{detail}: von keiner Beladung ({', '.join(entries)}) "
f"aus erreichbar"
)
if to_exit is not None and node_id not in to_exit:
result.append(
f"Bahn {node_id}{detail}: keine Entladung ({', '.join(exits)}) "
f"erreichbar - Sackgasse"
)
return result
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).
@@ -1496,9 +1571,16 @@ def build_graph(elements: list[Element], source: str, warnings: list[str]) -> Gr
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
# Kreisel: geschlossener Umlauf - das Ende von L laeuft an der einen
# Endkappe in den Anfang von R, das Ende von R an der anderen in den
# Anfang von L. Darum IMMER zwei Kanten L -> R und R -> L. Die Drehrichtung
# legt nur fest, welche davon die "Hauptkante" mit dem Kuerzel ist:
# UZS (im Uhrzeigersinn) -> Hauptkante R -> L
# GUZ (gegen den Uhrzeigersinn) -> Hauptkante L -> R
# Die Gegenkante traegt CIRCLE_RETURN_NOTE und wird in render_dot nicht
# gezeichnet (der Umlauf-Pfeil steht fuer beide). Frueher fehlte sie: jeder
# Kreisel hatte dann eine Bahn ohne Abgang und eine ohne Zugang, und die
# Pruefung unten meldete an fast jedem Kreisel Sackgasse + unerreichbar.
# Fehlt die Angabe, bleibt der Umlauf beidseitig und wird gemeldet.
for element in elements:
if not element.is_circle:
@@ -1508,8 +1590,10 @@ def build_graph(elements: list[Element], source: str, warnings: list[str]) -> Gr
rotation = element.rotation
if rotation == ROTATION_CW:
edges.append(Edge(right, left, EDGE_CIRCLE, ROTATION_CW))
edges.append(Edge(left, right, EDGE_CIRCLE, CIRCLE_RETURN_NOTE))
elif rotation == ROTATION_CCW:
edges.append(Edge(left, right, EDGE_CIRCLE, ROTATION_CCW))
edges.append(Edge(right, left, EDGE_CIRCLE, CIRCLE_RETURN_NOTE))
else:
warnings.append(
f"{element.describe()}: Drehrichtung '{element.rotation or '-'}' "
@@ -1525,28 +1609,7 @@ def build_graph(elements: list[Element], source: str, warnings: list[str]) -> Gr
# Kreisel<->Strecke gerichtet aus den Transferelementen (massgeblich)
edges = _apply_transfers(elements, by_id, nodes, edges, warnings)
# 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"
)
warnings.extend(_reachability_warnings(nodes, edges))
return Graph(nodes=nodes, edges=edges, warnings=warnings, unassigned=unassigned, source=source)
@@ -2078,6 +2141,8 @@ def render_dot(graph: Graph, use_coords: bool = False) -> str:
seen_group_edges: set[tuple[str, str]] = set()
for edge in graph.edges:
if edge.kind == EDGE_CIRCLE and edge.note == CIRCLE_RETURN_NOTE:
continue # Umlauf-Pfeil der Hauptkante steht fuer beide
src = group_of.get(edge.src, edge.src)
dst = group_of.get(edge.dst, edge.dst)
if edge.src in grouped_ids or edge.dst in grouped_ids:
@@ -3499,7 +3564,8 @@ def render_doc(
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)
circle = sum(1 for e in graph.edges
if e.kind == EDGE_CIRCLE and e.note != CIRCLE_RETURN_NOTE)
open_dir = sum(1 for e in graph.edges if e.kind == EDGE_UNRESOLVED)
add("## 1. Kurzfassung")
@@ -3622,7 +3688,8 @@ def report(
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)
circle = sum(1 for e in graph.edges
if e.kind == EDGE_CIRCLE and e.note != CIRCLE_RETURN_NOTE)
open_dir = sum(1 for e in graph.edges if e.kind == EDGE_UNRESOLVED)
devices = 0