tro_flow: Speicherbloecke je Steuer-Separator, Ring-Schluss bis zum ersten Kettenglied, Weichen nur in Laufrichtung

- _speicher_bloecke: Schwerkraftlinien nach Umlaufreihenfolge in Bloecke schneiden (Separator dazwischen = neuer Block); der Weichen-Separator vor dem Block ist Steuer-Separator des PinStore_Auto statt eigener 1Sep1Swi (Dematic: 3 Speicher 10/7/10 statt einem mit 27)
- Umlauf schliessen bis zum ersten Element mit TRO, nicht nur ueber Stationen hinweg (Dematic Kreisel2 beginnt mit Einschleuselement: TRO01 -> Speicher fehlte; Mubea TRO06 -> TRO10 statt Fallback TRO06 -> TRO18)
- _weichen_nach_umlauf: Separator steuert ein Ausschleuselement nur als letzter Separator davor in Laufrichtung (Dematic 0047/0046 waren ueber den Ring-Schluss bzw. hinter dem Block "benachbart")
- _kreisel_bahn: Kreisel-Separator mit Gefaellestrecke als Zuordnung auf die Kreisel-Bahn haengen (Dematic 0046 zaehlte als Speicherlinie)
- HundM05 unveraendert, Mubea 26 Kanten (eine korrigiert)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Samer Ayadi
2026-10-06 16:11:25 +02:00
co-authored by Claude Opus 5.5
parent e0defad011
commit a741dd0157
+186 -9
View File
@@ -290,6 +290,149 @@ def _group_lanes(lanes: list[str], graph: Graph) -> list[list[str]]:
return [group["lanes"] for group in groups]
def _kreisel_bahn(sep: Element, elements: list[Element], graph: Graph) -> str | None:
"""Kreisel-Bahn-Knoten eines Separators, der in der Umlaufkette
(Element.track_ids) eines Kreisels steht - sonst None.
dxfmakros traegt bei Kreisel-Separatoren als "Zuordnung" oft die naechste
Strecke statt der Bahn ein (Dematic/Arvato: 0043 -> VF_5, 0048 -> VF_4,
0046 -> Gefaellestrecke 0037; nur 0038/0047 stehen auf "0002-L"/"0003-R").
Bei einer Gefaellestrecke zaehlt er dann faelschlich als Speicherlinien-
Separator (Aufrufer in derive_tros wendet es nur dort an).
Die Bahn ergibt sich aus der Seite der Kreisel-Achse ("Drehung"): links
der Achsrichtung L, rechts R - dieselbe Konvention wie die Zuordnungen,
die der Export richtig setzt (0038 unterhalb von Kreisel2 mit Drehung 180
-> L, 0047 oberhalb von Kreisel3 -> R).
"""
for k in elements:
if k.kind != "Kreisel" or sep.teile_id not in k.track_ids:
continue
drehung = _as_float(_merkmal(k.merkmale, "Drehung"))
if drehung is None or None in (sep.x, sep.y, k.x, k.y):
return None
rad = math.radians(drehung)
kreuz = math.cos(rad) * (sep.y - k.y) - math.sin(rad) * (sep.x - k.x)
bahn = f"{k.teile_id}-{LANE_LEFT if kreuz > 0 else LANE_RIGHT}"
return bahn if bahn in graph.nodes else None
return None
def _weichen_nach_umlauf(
switch_hits: dict[str, tuple[Element, float]],
elements: list[Element],
findings: list[str],
) -> dict[str, tuple[Element, float]]:
"""Weichen-Separatoren verwerfen, die in Laufrichtung NACH ihrem
Ausschleuselement liegen.
_find_switch_separators paart ueber die CSV-Nachbarn - die kennen aber
keine Laufrichtung, und am Ring ist auch das erste mit dem letzten
Element der TrackIds-Liste benachbart. Ein Separator steuert ein
Ausschleuselement nur, wenn er der LETZTE Separator davor ist (Element.
track_ids des Kreisels ist in Laufrichtung sortiert, bei gesetzter
Drehrichtung ein geschlossener Ring). Dematic/Arvato (Review 2026-10-06):
0047 sitzt auf Kreisel3 am Kopf der Gegenbahn und war ueber den Ring-
Schluss mit dem Ausschleuselement 0057 von VF_1 "benachbart" (TRO03 als
Weiche in VF_1), 0046 liegt hinter dem letzten Ausschleuselement von
Speicherblock C. Beide sind gewoehnliche Kreisel-Separatoren (Schritt 3).
"""
by_id = {e.teile_id: e for e in elements}
kreisel = [e for e in elements if e.kind == "Kreisel" and e.track_ids]
result: dict[str, tuple[Element, float]] = {}
for sep_id, (transfer, distance) in switch_hits.items():
k = next((k for k in kreisel
if transfer.teile_id in k.track_ids and sep_id in k.track_ids), None)
if k is None:
result[sep_id] = (transfer, distance)
continue
ids = list(k.track_ids)
i = ids.index(transfer.teile_id)
ring = k.rotation in (ROTATION_CW, ROTATION_CCW)
davor = ids[:i][::-1] + (ids[i + 1:][::-1] if ring else [])
vorgaenger = next((t for t in davor
if (el := by_id.get(t)) and el.kind == "Separator"), None)
if vorgaenger == sep_id:
result[sep_id] = (transfer, distance)
else:
findings.append(
f"Separator {sep_id}: CSV-Nachbar von Ausschleuselement "
f"{transfer.teile_id}, liegt in Laufrichtung aber nicht davor "
f"(letzter Separator davor: {vorgaenger or '-'}) - keine Weiche, "
f"gewoehnlicher Kreisel-Separator")
return result
def _speicher_bloecke(
lanes: list[str],
elements: list[Element],
switch_hits: dict[str, tuple[Element, float]],
) -> tuple[dict[str, tuple[str, int]], dict[str, tuple[str, int]]]:
"""Schwerkraftlinien je Kreisel in Speicherbloecke schneiden.
_group_lanes fasst alle Linien mit gemeinsamem Zu-/Ablauf zusammen - an
Dematic/Arvato haengen aber alle 27 Linien an denselben zwei Kreiseln und
bilden doch DREI Bloecke (10/7/10), je von einem eigenen Separator auf der
Kreisel-Bahn gesteuert, dazwischen kehren VF_1..VF_3 zurueck (Review
2026-10-06). Grenze ist darum die Umlaufreihenfolge (Element.track_ids):
die Ausschleuselemente der Linien folgen dort lueckenlos aufeinander, ein
Separator dazwischen beginnt einen neuen Block. Mubea Kreisel3 (20
Ausschleuselemente ohne Separator dazwischen) bleibt ein Block.
Steuer-Separator eines Blocks ist der Weichen-Separator (switch_hits,
_find_switch_separators), dessen Ausschleuselement zu einer Linie des
Blocks gehoert: er steht auf der Kreisel-Bahn direkt davor und hat das
erste Ausschleuselement des Blocks als CSV-Nachbarn (Dematic 0040 ->
0083). Er gehoert in den PinStore_Auto-TRO (Katalog: 1x Separator +
Scanner + Encoder + Speicherlinien), nicht in einen eigenen 1Sep1Swi -
das Ausschleuselement einer Speicherlinie ist keine Weiche einer Strecke.
Mubea 0044 (TRO05) ist durch ein Einschleuselement vom Block getrennt,
kein Nachbar eines Ausschleuselements und bleibt deshalb eigener 1Sep.
Rueckgabe: (Linie -> Block, Steuer-Separator -> Block). Linien ohne
Ausschleuselement in einer Kreisel-Kette fehlen in der ersten Zuordnung.
"""
lane_set = set(lanes)
by_id = {e.teile_id: e for e in elements}
as_lane: dict[str, str] = {}
for e in elements:
if e.kind != "Ausschleus":
continue
strecke = str(_merkmal(e.merkmale, "Strecke", "Förderer") or "")
if strecke in lane_set:
as_lane[e.teile_id] = strecke
lane_block: dict[str, tuple[str, int]] = {}
as_block: dict[str, tuple[str, int]] = {}
for kreisel in (e for e in elements if e.kind == "Kreisel"):
ids = list(kreisel.track_ids)
if not any(t in as_lane for t in ids):
continue
# Am ersten Separator beginnen, damit ein Block nicht ueber das
# Listenende hinweg in zwei Teile zerfaellt (die Kette ist ein Ring).
start = next((i for i, t in enumerate(ids)
if (el := by_id.get(t)) and el.kind == "Separator"), 0)
ids = ids[start:] + ids[:start]
nummer, offen = 0, False
for t in ids:
el = by_id.get(t)
if el is not None and el.kind == "Separator":
offen = False
elif t in as_lane:
if not offen:
nummer += 1
offen = True
block = (kreisel.teile_id, nummer)
as_block[t] = block
lane_block.setdefault(as_lane[t], block)
steuer: dict[str, tuple[str, int]] = {}
for sep_id, (transfer, _d) in switch_hits.items():
block = as_block.get(transfer.teile_id)
if block is not None:
steuer[sep_id] = block
return lane_block, steuer
def _switch_exits(hosts: list[str], graph: Graph) -> list[str]:
"""Abgehende Flusskanten eines TRO-Hosts (ohne Kreisel-Umlauf)."""
return sorted(
@@ -587,7 +730,8 @@ def derive_tros(
# unten (3b) ihre eigene 1Sep1Swi-Uebergabestelle und duerfen nicht zugleich
# als gewoehnlicher Separator ihres Zuordnung-Hosts (Kreisel-Bahn oder
# Strecke - beides kommt vor) mitgezaehlt werden.
switch_hits = _find_switch_separators(switch_candidates, elements)
switch_hits = _weichen_nach_umlauf(
_find_switch_separators(switch_candidates, elements), elements, findings)
# Separatoren an einer echten ILS Weiche (siehe _find_weiche_separators) -
# wie switch_hits vor der Host-Aufloesung herausgenommen, Schritt 3c.
weiche_hits = _find_weiche_separators(
@@ -601,6 +745,17 @@ def derive_tros(
if sep.teile_id in switch_hits or sep.teile_id in weiche_hits:
continue
host = resolve_host(sep, findings)
# Steht er in der Umlaufkette eines Kreisels, zeigt die Zuordnung aber
# auf eine Gefaellestrecke, ist das nur die naechstgelegene Linie
# (siehe _kreisel_bahn) - sonst zaehlte er als Speicherlinien-
# Separator (Dematic 0046 in Speicherblock C). Bewusst NUR fuer
# Gefaellestrecken: Kreisel-Separatoren mit einer VF-Strecke als
# Zuordnung bleiben wie bisher (HundM05 0018 laut Review 2026-09-21
# Teil seiner Strecke; ein Umhaengen nummerierte HundM05/Mubea um).
if host is not None and graph.nodes[host].kind == "Gefaellestrecke":
bahn = _kreisel_bahn(sep, elements, graph)
if bahn is not None:
host = bahn
if host is not None:
sep_host[sep.teile_id] = host
@@ -661,20 +816,37 @@ def derive_tros(
lanes = sorted(
{h for h in sep_host.values() if graph.nodes[h].kind == "Gefaellestrecke"}
)
# Speicherbloecke nach Umlaufreihenfolge und Steuer-Separator schneiden
# (siehe _speicher_bloecke). Steuer-Separatoren eines gebildeten Speichers
# bekommen in 3b keinen eigenen 1Sep1Swi mehr.
lane_block, steuer_block = _speicher_bloecke(lanes, elements, switch_hits)
speicher_steuer: set[str] = set()
teilgruppen: list[list[str]] = []
for group in _group_lanes(lanes, graph):
nach_block: dict = {}
for lane in group:
nach_block.setdefault(lane_block.get(lane), []).append(lane)
teilgruppen.extend(sorted(nach_block.values(), key=lambda g: g[0]))
for group in teilgruppen:
seps = seps_on(group)
if len(group) >= MIN_STORE_LANES:
bloecke = {lane_block.get(lane) for lane in group} - {None}
steuer = sorted(s for s, b in steuer_block.items() if b in bloecke)
speicher_steuer.update(steuer)
alle_seps = steuer + seps
tros.append(
Tro(
tro_id="",
type_name="PinStore_Auto",
label=f"Speicher ({len(group)} Schwerkraftlinien)",
hosts=group,
separators=seps,
scanners=scanners_for(seps),
separators=alle_seps,
scanners=scanners_for(alle_seps),
reason=(
f"{len(group)} Gefaellestrecken mit je eigenem Separator, "
f"gemeinsamer Zu-/Ablauf - Muster eines Linienspeichers"
+ (f"; Steuer-Separator {', '.join(steuer)} auf der "
f"Kreisel-Bahn davor" if steuer else "")
),
confidence="hoch",
)
@@ -857,6 +1029,8 @@ def derive_tros(
# von selbst ein, ohne einen weiteren synthetischen Host zu erfinden.
as_switch_tro: dict[str, Tro] = {} # Ausschleuselement-Id -> 1Sep1Swi
for sep_id, (transfer, distance) in sorted(switch_hits.items()):
if sep_id in speicher_steuer:
continue # Teil seines Speichers (Schritt 1)
strecke_id = str(_merkmal(transfer.merkmale, "Strecke", "Förderer") or "")
ende = str(_merkmal(transfer.merkmale, "Streckenende") or "")
own_scanners = sorted(sep_scanners.get(sep_id, []))
@@ -1339,13 +1513,16 @@ def link_tros(
laenge = len(chain)
geschlossen = el.rotation in (ROTATION_CW, ROTATION_CCW)
if geschlossen:
# Umlauf schliessen: erstes Element hinten anhaengen. Stationen am
# Listenanfang sind kein Kettenglied - ueber sie hinweg bis zum
# ersten anderen Element verlaengern, sonst fehlt der Schluss
# (HundM05 Kreisel1 beginnt mit den Beladungen 0021/0022: ohne das
# fehlte TRO19 -> TRO20).
# Umlauf schliessen: den Listenanfang hinten anhaengen, und zwar
# bis einschliesslich des ersten Kettenglieds (Element mit TRO).
# Alles davor - Stationen, aber auch Ein-/Ausschleuselemente ohne
# eigenen TRO - ist kein Glied; endete die Verlaengerung dort, fehlte
# der Schluss (HundM05 Kreisel1 beginnt mit den Beladungen 0021/
# 0022: ohne das fehlte TRO19 -> TRO20; Dematic Kreisel2 beginnt
# mit dem Einschleuselement 0060: es fehlte TRO01 -> Speicher,
# Review 2026-10-06).
k = 0
while k < laenge - 1 and chain[k] in station_hosts:
while k < laenge - 1 and chain[k] not in ketten_owner:
k += 1
chain.extend(chain[:k + 1])