diff --git a/lib/material_flow.py b/lib/material_flow.py index ae033cd..51eea38 100644 --- a/lib/material_flow.py +++ b/lib/material_flow.py @@ -769,8 +769,23 @@ def _nearest_node(station: Element, nodes: dict[str, Node]) -> str | None: return best +def _nearest_kreisel(station: Element, elements: list[Element]) -> Element | None: + """Naechstgelegener Kreisel zur Station (2D-Abstand zur Kreisel-Achsposition).""" + if station.x is None or station.y is None: + return None + best, best_d = None, None + for el in elements: + if not el.is_circle or el.x is None or el.y is None: + continue + d = (el.x - station.x) ** 2 + (el.y - station.y) ** 2 + if best_d is None or d < best_d: + best, best_d = el, d + return best + + def _attach_stations( - elements: list[Element], nodes: dict[str, Node], warnings: list[str] + 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. @@ -778,7 +793,11 @@ def _attach_stations( 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), wird das naechstgelegene Objekt per Abstand genommen. + 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. """ edges: list[Edge] = [] for element in elements: @@ -796,6 +815,18 @@ def _attach_stations( else: partners += [f"{base}-{L}" for L in (LANE_LEFT, LANE_RIGHT) if f"{base}-{L}" in nodes] + 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") + 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})" + ) if not partners: nearest = _nearest_node(element, nodes) if nearest: @@ -827,6 +858,30 @@ def _kreisel_lane_node(kreisel_base: str, strecke: Element, nodes: dict[str, Nod return kreisel_base if kreisel_base in nodes else None +def _weiche_lane_roles(kreisel_base: str, by_id: dict[str, Element], nodes: dict[str, Node]) -> tuple[str, str] | None: + """Eingangs-/Ausgangsbahn eines Kreisels an einer Kreisel<->Kreisel-Weiche. + + Ein Kreisel ist intern ein Umlauf zwischen seinen zwei Bahnen (Merkmal + "Drehrichtung"); am Beruehrpunkt mit dem anderen Kreisel liegt daher + zwingend die eine Bahn als Eingang (freies Ende vor dem internen Umlauf) + und die andere als Ausgang (freies Ende danach): + UZS (Umlauf R -> L) -> Eingang R, Ausgang L + GUZ (Umlauf L -> R) -> Eingang L, Ausgang R + None, wenn die Drehrichtung fehlt/unbekannt ist oder eine Bahn im Graph fehlt. + """ + element = by_id.get(kreisel_base) + if element is None or not element.is_circle: + return None + left, right = f"{kreisel_base}-{LANE_LEFT}", f"{kreisel_base}-{LANE_RIGHT}" + if left not in nodes or right not in nodes: + return None + if element.rotation == ROTATION_CW: + return right, left + if element.rotation == ROTATION_CCW: + return left, right + return None + + def _apply_transfers( elements: list[Element], by_id: dict[str, Element], nodes: dict[str, Node], edges: list[Edge], warnings: list[str] @@ -836,7 +891,10 @@ def _apply_transfers( Ausschleuselement: Kreisel -> Strecke (Streckenanfang). Einschleuselement: Strecke -> Kreisel (Streckenende). - Weiche: Kreisel<->Kreisel als Weiche markieren. + Weiche: Kreisel<->Kreisel, Bahn und Richtung aus der Drehrichtung + beider Kreisel (siehe _weiche_lane_roles); ohne bekannte + Drehrichtung bleibt die Verbindung unbestimmt, wird aber + weiterhin als Weiche markiert. Die so definierten Kanten sind massgeblich und ersetzen die aus Hoehe/ Nachbarn abgeleitete Richtung fuer genau diese Knotenpaare. """ @@ -844,6 +902,7 @@ def _apply_transfers( # ein = Auslauf (Strecke->Kreisel, Ende) per_strecke: dict[str, dict[str, str]] = {} weiche_pairs: list[set[str]] = [] + weiche_ab: list[tuple[str, str]] = [] for el in elements: if el.kind in ("Ausschleus", "Einschleus"): k = str(_merkmal(el.merkmale, "Kreisel") or "").strip() @@ -856,6 +915,28 @@ def _apply_transfers( b = str(_merkmal(el.merkmale, "Kreisel B") or "").strip() if a and b: weiche_pairs.append({a, b}) + weiche_ab.append((a, b)) + + # Weiche zwischen zwei Kreiseln: Bahn und Richtung deterministisch aus der + # Drehrichtung beider Kreisel ableiten (Ausgang der einen Bahn -> Eingang + # der anderen, in beiden Richtungen) statt aus dem rohen Nachbarn-Kreuzprodukt, + # das links/rechts nicht unterscheiden kann. + weiche_edges: list[Edge] = [] + weiche_resolved: set[frozenset] = set() + for a, b in weiche_ab: + roles_a = _weiche_lane_roles(a, by_id, nodes) + roles_b = _weiche_lane_roles(b, by_id, nodes) + if roles_a is None or roles_b is None: + warnings.append( + f"Weiche {a}<->{b}: Drehrichtung mindestens eines Kreisels unbekannt - " + f"Bahnzuordnung nicht bestimmbar, Verbindung bleibt unbestimmt" + ) + continue + in_a, out_a = roles_a + in_b, out_b = roles_b + weiche_edges.append(Edge(out_a, in_b, EDGE_FLOW, "Weiche")) + weiche_edges.append(Edge(out_b, in_a, EDGE_FLOW, "Weiche")) + weiche_resolved.add(frozenset({a, b})) directed: dict[frozenset, tuple[str, str]] = {} for s, roles in per_strecke.items(): @@ -885,9 +966,13 @@ def _apply_transfers( # einem dritten Kreisel liegt) ist ein Falschtreffer und wird verworfen. covered = set(per_strecke) result: list[Edge] = [] + superseded: set[tuple[str, str]] = set() for e in edges: if frozenset({e.src, e.dst}) in directed: continue # wird gerichtet neu gesetzt + if frozenset({e.src.split("-")[0], e.dst.split("-")[0]}) in weiche_resolved: + superseded.add((e.src, e.dst)) + continue # wird durch die Weiche-Kanten ersetzt ends = {e.src, e.dst} strecke_end = ends & covered others = ends - strecke_end @@ -902,7 +987,23 @@ def _apply_transfers( result.append(e) for (src, dst, note) in directed.values(): result.append(Edge(src, dst, EDGE_FLOW, note)) - # Kreisel<->Kreisel-Kanten, die eine Weiche verbinden, markieren + result.extend(weiche_edges) + # Die anfaengliche Hoehen-basierte Ausrichtung (_orient) kann eine + # Kreisel<->Kreisel-Verbindung nie bestimmen und meldet sie immer als + # "beidseitig gezeichnet" - das stimmt nicht mehr, sobald die Weiche sie + # gerade durch obenstehende Kanten ersetzt hat. Die dadurch ueberholte + # Meldung wird entfernt. + if superseded: + stale = { + f"Verbindung {a} <-> {b}: Richtung nicht bestimmbar " + f"(kein Transportobjekt beteiligt) - beidseitig gezeichnet" + for src, dst in superseded for a, b in ((src, dst), (dst, src)) + } + warnings[:] = [w for w in warnings if w not in stale] + # Kreisel<->Kreisel-Kanten, die eine Weiche verbinden, markieren (deckt vor + # allem die Faelle ab, die mangels Drehrichtung nicht gerichtet aufgeloest + # werden konnten - die aufgeloesten Weiche-Kanten tragen die Markierung + # bereits). for e in result: if {e.src.split("-")[0], e.dst.split("-")[0]} in weiche_pairs: e.note = "Weiche" @@ -949,7 +1050,7 @@ def build_graph(elements: list[Element], source: str, warnings: list[str]) -> Gr edges.append(Edge(right, left, EDGE_CIRCLE, "unbestimmt")) # Ein-/Ausgabestationen (BTMT Be-/Entladung) als Quelle/Senke anbinden - edges.extend(_attach_stations(elements, nodes, warnings)) + edges.extend(_attach_stations(elements, by_id, nodes, warnings)) # Kreisel<->Strecke gerichtet aus den Transferelementen (massgeblich) edges = _apply_transfers(elements, by_id, nodes, edges, warnings) diff --git a/lib/tro_flow.py b/lib/tro_flow.py index 8995f76..0e3f87e 100644 --- a/lib/tro_flow.py +++ b/lib/tro_flow.py @@ -60,6 +60,7 @@ import argparse import csv import math import sys +from collections.abc import Callable from dataclasses import dataclass, field from datetime import datetime from pathlib import Path @@ -88,6 +89,7 @@ from material_flow import ( from tro_catalog import ( CATALOG_AS_OF, ITEM_FB, + STYLE_EXT, STYLES, TRO_CATALOG, TroDefinition, @@ -111,6 +113,14 @@ from tro_overrides import ( # einem Separator) wird ein Speicherblock (PinStore_Auto) angenommen. MIN_STORE_LANES = 2 +# Toleranz (mm), innerhalb derer ein Separator als "sitzt an einem Aus-/ +# Einschleuselement" gilt (siehe _find_switch_separators). Kalibriert an +# hundmM05_1.csv: die tatsaechlich am Uebergang montierten Separatoren liegen +# 250-350mm vom Element entfernt, die naechsten unbeteiligten (auf der +# eigenen Streckenmitte montierten) Separatoren erst ab ca. 885mm - die +# Grenze liegt bewusst in der Mitte dieser Luecke. +SWITCH_PROXIMITY_MM = 500.0 + GENERATED_MARKER = "" # Zeichnungsmassstab fuer --use-cords: Graphviz rechnet in Punkt, die Anlage in @@ -187,9 +197,14 @@ class Analysis: findings: list[str] = field(default_factory=list) # Manuelle Korrekturen aus cfg/tro_overrides.ini (Typkorrekturen, offene Punkte) overrides: Overrides = field(default_factory=Overrides) + # Wie Host/Fanout ermittelt wurden - siehe derive_tros(resolve_host, count_exits) + method: str = "Materialfluss-Graph (Zuordnung/Nachbarn)" + # Systemgrenzen (BTMT Be-/Entladestationen) - siehe derive_boundaries; kein + # TRO, nur damit der tatsaechliche Materialein-/ausgang sichtbar wird. + boundaries: list[Tro] = field(default_factory=list) def by_id(self, tro_id: str) -> Tro | None: - for tro in self.tros: + for tro in self.tros + self.boundaries: if tro.tro_id == tro_id: return tro return None @@ -243,13 +258,71 @@ def _switch_exits(hosts: list[str], graph: Graph) -> list[str]: ) +def _default_resolve_host(sep: Element, findings: list[str], graph: Graph) -> str | None: + """Host eines Separators aus der CSV-Spalte 'Zuordnung' (Default-Resolver).""" + host = sep.assignment + if host and host != UNASSIGNED and host in graph.nodes: + return host + findings.append( + f"Separator {sep.teile_id} '{sep.name}': Zuordnung '{host or '-'}' " + f"zeigt auf keinen Flussknoten - keinem TRO zugeordnet" + ) + return None + + +def _find_switch_separators( + separators: list[Element], elements: list[Element], tol_mm: float +) -> dict[str, tuple[Element, float]]: + """Separator-Id -> (naechstgelegenes Aus-/Einschleuselement, Abstand mm), + nur wenn Abstand <= tol_mm. + + HundM05 hat an seinen Kreisel<->Strecke-Uebergaengen kein eigenes "ILS + Weiche"-Bauteil (anders als ein Kreisel<->Kreisel-Beruehrpunkt, siehe + Mubea) - das von dxfmakros synthetisierte Ausschleuselement (Kreisel -> + Strecke, am Streckenanfang) bzw. Einschleuselement (Strecke -> Kreisel, + am Streckenende) uebernimmt dort faktisch die Rolle der Weiche. Sitzt ein + Separator direkt daneben, gehoert er zu dieser Weiche - unabhaengig davon, + ob seine CSV-Spalte "Zuordnung" auf die Kreisel-Bahn oder auf die Strecke + zeigt (in den Daten kommt beides vor, siehe derive_tros). + """ + transfers = [e for e in elements if e.kind in ("Ausschleus", "Einschleus")] + result: dict[str, tuple[Element, float]] = {} + for sep in separators: + if sep.x is None or sep.y is None: + continue + best: tuple[float, Element] | None = None + for transfer in transfers: + if transfer.x is None or transfer.y is None: + continue + d = math.hypot(transfer.x - sep.x, transfer.y - sep.y) + if d <= tol_mm and (best is None or d < best[0]): + best = (d, transfer) + if best is not None: + result[sep.teile_id] = (best[1], best[0]) + return result + + def derive_tros( graph: Graph, elements: list[Element], findings: list[str], overrides: Overrides | None = None, + *, + resolve_host: Callable[[Element, list[str]], str | None] | None = None, + count_exits: Callable[[list[str]], list[str]] | None = None, ) -> list[Tro]: - """TRO-Liste aus den Separatoren und ihren angeschlossenen Objekten bilden.""" + """TRO-Liste aus den Separatoren und ihren angeschlossenen Objekten bilden. + + resolve_host(sep, findings) -> Host-Knoten-Id oder None; Default liest die + CSV-Spalte "Zuordnung" (siehe _default_resolve_host). count_exits(hosts) -> + Liste abgehender Nachbarn eines TRO-Hosts; Default liest graph.edges (siehe + _switch_exits). tro_geometry.py::make_geometric_resolvers() liefert ein + Paar, das beides stattdessen direkt aus den Bauteil-Weltkoordinaten sucht + (STRtree) - siehe tro_flow.py --geometry. + """ + resolve_host = resolve_host or (lambda sep, findings: _default_resolve_host(sep, findings, graph)) + count_exits = count_exits or (lambda hosts: _switch_exits(hosts, graph)) + separators = sorted( (e for e in elements if e.kind == "Separator"), key=lambda e: e.teile_id ) @@ -257,30 +330,39 @@ def derive_tros( (e for e in elements if e.kind == "Scanner"), key=lambda e: e.teile_id ) - # Separator -> Host-Knoten (Spalte "Zuordnung") + # Separatoren an einem Aus-/Einschleuselement (siehe _find_switch_separators) + # werden VOR der normalen Host-Aufloesung herausgenommen: sie bilden gleich + # 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(separators, elements, SWITCH_PROXIMITY_MM) + + # Separator -> Host-Knoten sep_host: dict[str, str] = {} for sep in separators: - host = sep.assignment - if host and host != UNASSIGNED and host in graph.nodes: + if sep.teile_id in switch_hits: + continue + host = resolve_host(sep, findings) + if host is not None: sep_host[sep.teile_id] = host - else: - findings.append( - f"Separator {sep.teile_id} '{sep.name}': Zuordnung '{host or '-'}' " - f"zeigt auf keinen Flussknoten - keinem TRO zugeordnet" - ) # Scanner -> Separator; "Naechster Separator" ist die funktionale Zuordnung, # "Zuordnung" nur der Montageort. sep_scanners: dict[str, list[str]] = {} for scanner in scanners: target = str(_merkmal(scanner.merkmale, "Naechster Separator") or "").strip() - if target in sep_host: + if target in sep_host or target in switch_hits: sep_scanners.setdefault(target, []).append(scanner.teile_id) mounted = scanner.assignment - if mounted and mounted != UNASSIGNED and mounted != sep_host[target]: + target_host = sep_host.get(target) + # Fuer Switch-Separatoren (3b) gibt es keinen einzelnen Host zum + # Vergleichen (sie haengen zwischen zwei Ketten) - der Abgleich + # entfaellt dort, ohne den Scanner deshalb als "unbekannt" zu werten. + if (mounted and mounted != UNASSIGNED and target_host is not None + and mounted != target_host): findings.append( f"Scanner {scanner.teile_id} '{scanner.name}': montiert an {mounted}, " - f"liest aber an Separator {target} auf {sep_host[target]}" + f"liest aber an Separator {target} auf {target_host}" ) else: findings.append( @@ -380,11 +462,41 @@ def derive_tros( ) ) + # 3b) Separatoren an einem Aus-/Einschleuselement -> eigene 1Sep1Swi- + # Uebergabestelle. Das Element uebernimmt die Rolle der (hier fehlenden) + # Weiche; siehe _find_switch_separators. Host bewusst die Strecke (nicht + # die Kreisel-Bahn): beide Elementarten tragen "Strecke" im Merkmal, und + # die Strecke hat bereits reale, gerichtete Kanten zum Kreisel + # (_apply_transfers) - darueber haengt sich die Verknuepfung (link_tros) + # von selbst ein, ohne einen weiteren synthetischen Host zu erfinden. + for sep_id, (transfer, distance) in sorted(switch_hits.items()): + strecke_id = str(_merkmal(transfer.merkmale, "Strecke") or "") + ende = str(_merkmal(transfer.merkmale, "Streckenende") or "") + own_scanners = sorted(sep_scanners.get(sep_id, [])) + art = "Ausschleuselement" if transfer.kind == "Ausschleus" else "Einschleuselement" + tros.append( + Tro( + tro_id="", + type_name="1Sep1Swi", + label=f"Separator {sep_id} an {art} {transfer.teile_id} " + f"(Strecke {strecke_id}, {ende})", + hosts=[strecke_id] if strecke_id in graph.nodes else [transfer.teile_id], + separators=[sep_id], + scanners=own_scanners, + reason=( + f"{distance:.0f}mm von {art} {transfer.teile_id} entfernt - " + f"kein eigenes Weichen-Bauteil im Layout, das {art} " + f"uebernimmt diese Rolle" + ), + confidence="mittel", + ) + ) + # 4) Weichen-Aufwertung: 2 bzw. 3 abgehende Flusskanten for tro in tros: if tro.type_name != "1Sep": continue - exits = _switch_exits(tro.hosts, graph) + exits = count_exits(tro.hosts) if not 2 <= len(exits) <= 3: continue upgraded = "1Sep1Swi" if len(exits) == 2 else "1Sep2Swi" @@ -443,6 +555,60 @@ def derive_tros( return tros +# Typnamen fuer Systemgrenz-Knoten (siehe derive_boundaries) - bewusst NICHT +# in TRO_CATALOG: eine BTMT Be-/Entladestation ist kein Steuerobjekt mit +# eigenem FB-Baustein, nur der tatsaechliche Materialein-/ausgang. +BOUNDARY_ENTRY = "Materialeingang" +BOUNDARY_EXIT = "Materialausgang" +BOUNDARY_TYPES = (BOUNDARY_ENTRY, BOUNDARY_EXIT) + + +def derive_boundaries(elements: list[Element], graph: Graph) -> list[Tro]: + """BTMT Be-/Entladestationen als sichtbare Systemgrenz-Knoten. + + Ohne diese Knoten "verschwindet" der Materialfluss stillschweigend an der + Station: link_tros() registriert eine Kante nur, wenn die Traversierung + einen Knoten erreicht, der ein TRO-Host ist. Eine BTMT-Station ist nie + ein TRO-Host (sie hat keinen Separator), also wird sie beim Durchlaufen + als passiver Zwischenknoten behandelt - und da eine Entladung (Senke) + keine ausgehende Kante mehr hat, bricht die Suche dort einfach ab, ohne + eine Kante einzutragen. Eine Beladung (Quelle) wird umgekehrt nie als + Ausgangspunkt einer Traversierung betrachtet, weil kein TRO dort beginnt. + Damit fehlte bislang genau der Teil des Flusses, den man am Anfang oder + Ende eines Durchlaufs durch die Anlage sieht. + + Kein eigener TRO-Typ (keine FB_ILS_MTRO_*-Zuordnung, kein Eintrag in + TRO_CATALOG) - nur ein sichtbarer Marker fuer den echten Anfang/das echte + Ende des Materialflusses. Eigene Nummerierung ("SYS1", "SYS2", ...) + getrennt von den TRO-IDs, damit auf den ersten Blick klar bleibt: das + sind keine Steuerobjekte. + """ + boundaries: list[Tro] = [] + for element in elements: + if element.kind not in ("Beladung", "Entladung"): + continue + if element.teile_id not in graph.nodes: + continue + boundaries.append( + Tro( + tro_id="", + type_name=BOUNDARY_ENTRY if element.is_entry else BOUNDARY_EXIT, + label=element.name, + hosts=[element.teile_id], + separators=[], + scanners=[], + reason="Systemgrenze (BTMT Be-/Entladestation) - kein TRO, " + "nur Materialein-/ausgang", + confidence="hoch", + ) + ) + boundaries.sort(key=lambda t: t.hosts[0]) + width = max(1, len(str(len(boundaries)))) + for index, boundary in enumerate(boundaries, start=1): + boundary.tro_id = f"SYS{index:0{width}d}" + return boundaries + + def _tro_edge_kind(edge: Edge) -> str: """Materialkante -> Kantenart des TRO-Flusses (fuer Einfaerbung).""" if edge.note == "Weiche": @@ -620,23 +786,38 @@ def locate_tros(tros: list[Tro], elements: list[Element], graph: Graph) -> None: def analyse_tros( - graph: Graph, elements: list[Element], overrides: Overrides | None = None + graph: Graph, + elements: list[Element], + overrides: Overrides | None = None, + *, + resolve_host: Callable[[Element, list[str]], str | None] | None = None, + count_exits: Callable[[list[str]], list[str]] | None = None, + method: str | None = None, ) -> Analysis: """TROs ermitteln, verknuepfen, verorten und gegen die Typdefinition pruefen.""" analysis = Analysis() analysis.overrides = overrides if overrides is not None else Overrides() - analysis.tros = derive_tros(graph, elements, analysis.findings, analysis.overrides) + if method is not None: + analysis.method = method + analysis.tros = derive_tros(graph, elements, analysis.findings, analysis.overrides, + resolve_host=resolve_host, count_exits=count_exits) + analysis.boundaries = derive_boundaries(elements, graph) + # Fuer Verortung und Verknuepfung zaehlen TROs und Systemgrenzen zusammen: + # link_tros() registriert eine Kante nur zu einem Knoten, der Host eines + # Eintrags in der uebergebenen Liste ist - eine BTMT-Station muss also mit + # in dieser Liste stehen, sonst bleibt sie unsichtbar (siehe derive_boundaries). + linked = analysis.tros + analysis.boundaries # Verorten vor dem Verknuepfen: die Reihenfolge mehrerer TROs auf einer # Kreisel-Bahn wird ueber ihre Koordinate bestimmt. - locate_tros(analysis.tros, elements, graph) - analysis.edges, analysis.edge_kinds = link_tros(graph, analysis.tros) + locate_tros(linked, elements, graph) + analysis.edges, analysis.edge_kinds = link_tros(graph, linked) # Vor Ort aufgenommene Topologie schlaegt die abgeleitete. Die Differenz # wird protokolliert, damit sichtbar bleibt, was die Ableitung falsch hatte. if analysis.overrides.connections: abgeleitet = set(analysis.edges) analysis.edges, analysis.edge_kinds = apply_connections( - analysis.tros, analysis.overrides + linked, analysis.overrides ) gesetzt = set(analysis.edges) for src, dst in sorted(abgeleitet - gesetzt): @@ -688,14 +869,16 @@ def _tro_label(tro: Tro) -> str: def missing_positions(analysis: Analysis) -> list[str]: - """TROs ohne Koordinate - mit ihnen ist kein lagerichtiges Layout moeglich.""" - return [t.tro_id for t in analysis.tros if not t.position] + """TROs/Systemgrenzen ohne Koordinate - mit ihnen ist kein lagerichtiges + Layout moeglich (auch eine Systemgrenze wird als Knoten gezeichnet).""" + return [t.tro_id for t in analysis.tros + analysis.boundaries if not t.position] def _legend_pos(analysis: Analysis) -> tuple[float, float]: """Legende links unterhalb der Anlage platzieren (mm).""" - xs = [t.position[0] for t in analysis.tros if t.position] - ys = [t.position[1] for t in analysis.tros if t.position] + all_nodes = analysis.tros + analysis.boundaries + xs = [t.position[0] for t in all_nodes if t.position] + ys = [t.position[1] for t in all_nodes if t.position] if not xs: return (0.0, 0.0) return (min(xs) - LEGEND_MARGIN, min(ys) - LEGEND_MARGIN) @@ -764,6 +947,20 @@ def render_dot(analysis: Analysis, source: str, use_coords: bool = False) -> str add(f' "{tro.tro_id}" [label="{_tro_label(tro)}", {rendered}{pos}];') add("") + # Systemgrenzen (BTMT Be-/Entladestationen): kein Katalogtyp, eigene + # Farbe/Form (ext-Stil, Ellipse statt Box) - deutlich als "kein TRO" + # erkennbar, aber sichtbar als tatsaechlicher Anfang/Ende des Flusses. + for boundary in analysis.boundaries: + rendered = ", ".join(f'{k}="{v}"' for k, v in STYLE_EXT.dot_attrs().items()) + pos = "" + if boundary.position: + scale = COORD_SCALE if use_coords else 0.01 + pos = (f', pos="{boundary.position[0] * scale:.2f},' + f'{boundary.position[1] * scale:.2f}"') + add(f' "{boundary.tro_id}" [label="{_tro_label(boundary)}", ' + f'shape=ellipse, style=filled, {rendered}{pos}];') + add("") + for src, dst in analysis.edges: add(f' "{src}" -> "{dst}" [{_tro_edge_attrs(analysis.edge_kinds.get((src, dst)))}];') @@ -903,6 +1100,10 @@ def _tro_mermaid(analysis: Analysis) -> list[str]: label = f"{tro.tro_id}
{tro.type_name}
{_mermaid_text(tro.label)}" out.append(f' {tro.tro_id}["{label}"]:::{group}') + for boundary in analysis.boundaries: + label = f"{boundary.tro_id}
{boundary.type_name}
{_mermaid_text(boundary.label)}" + out.append(f' {boundary.tro_id}(["{label}"]):::ext') + for src, dst in analysis.edges: out.append(f" {src} --> {dst}") @@ -935,6 +1136,7 @@ def render_doc( add(f"**Erzeugt:** {datetime.now().strftime('%Y-%m-%d %H:%M')} von " f"`lib/tro_flow.py` ") add(f"**Quelle:** `{csv_file}` ") + add(f"**Methode:** {analysis.method} ") add(f"**Typdefinition:** `lib/tro_catalog.py` (Stand {CATALOG_AS_OF})") add("") add("> Automatisch aus dem CSV-Export erzeugt. Die TRO-Typen sind aus dem " @@ -952,6 +1154,7 @@ def render_doc( add("| | |") add("|---|---|") add(f"| Ermittelte TROs | {len(analysis.tros)} |") + add(f"| Systemgrenzen (Materialein-/ausgang) | {len(analysis.boundaries)} |") add(f"| TRO-Verbindungen | {len(analysis.edges)} |") add(f"| Nicht verbundene TROs | {len(analysis.isolated)} |") add(f"| Hinweise | {len(analysis.findings)} |") @@ -1037,6 +1240,24 @@ def render_doc( add(f"Details: [`{error_file.name}`]({error_file.name})") add("") + if analysis.boundaries: + add("### Systemgrenzen (Materialein-/ausgang)") + add("") + add("BTMT Be-/Entladestationen - kein TRO (kein Separator, kein " + "FB-Baustein), aber der tatsaechliche Anfang bzw. das Ende des " + "Materialflusses. Ohne diese Zeile wuerde die Verbindung an der " + "Station stillschweigend enden (siehe `derive_boundaries` in " + "`lib/tro_flow.py`).") + add("") + add("| Knoten | Typ | Bezeichnung | verbunden mit |") + add("|---|---|---|---|") + for boundary in analysis.boundaries: + other = ", ".join(f"`{x}`" for x in + sorted(set(boundary.successors) | set(boundary.predecessors))) or "-" + add(f"| `{boundary.tro_id}` | `{boundary.type_name}` | " + f"{boundary.label} | {other} |") + add("") + add("## 5. Hinweise") add("") if analysis.findings: @@ -1107,6 +1328,8 @@ def render_doc( add("| Gefaellestrecke, einzelne Linie | `1Sep` |") add("| Strecke (angetrieben) | `Vario` (je Streckensegment) |") add("| Kreisel-Bahn | `1Sep` |") + add(f"| Separator <= {SWITCH_PROXIMITY_MM:.0f}mm von einem Aus-/Einschleuselement " + "| `1Sep1Swi` (Element uebernimmt die fehlende Weiche) |") add("| zusaetzlich: 2 bzw. 3 abgehende Wege | Aufwertung zu `1Sep1Swi` / `1Sep2Swi` |") add("") add("Ein **Scanner** am Separator aendert den Typ nicht. In der Referenzanlage " @@ -1167,6 +1390,7 @@ def report( print("TRO-FLUSS") print("================================================================") print(f"Eingabe = {csv_file}") + print(f"Methode = {analysis.method}") print(f"TROs = {len(analysis.tros)}" + (f" ({', '.join(f'{k}: {v}' for k, v in sorted(types_used.items()))})" if types_used else "")) @@ -1284,9 +1508,36 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace: help="Dokumentation der TROs als Markdown in %%SKEL_RESULTS%% erzeugen " "(TRO-Liste, Verbindungen, Typdefinition).", ) + parser.add_argument( + "--geometry", + action="store_true", + help="Separator-Host und Weichen-Fanout per STRtree direkt aus den " + "Bauteil-Weltkoordinaten bestimmen (lib/tro_geometry.py) statt " + "aus den CSV-Spalten 'Zuordnung'/'Nachbarn' - unabhaengig davon, " + "ob diese Spalten fuer die Eingabedatei luecken haben.", + ) + parser.add_argument( + "--geo-tol-mm", + type=float, + default=tro_geometry_default_tolerance_mm(), + metavar="MM", + help="Nur mit --geometry: Toleranz (mm), innerhalb derer zwei " + "Anschlusspunkte als derselbe Uebergabepunkt gelten. " + "Standard: %(default)s", + ) return parser.parse_args(argv) +def tro_geometry_default_tolerance_mm() -> float: + """Default fuer --geo-tol-mm; importiert tro_geometry nur bei Bedarf, damit + --help/normale Laeufe ohne --geometry kein shapely brauchen.""" + try: + from tro_geometry import DEFAULT_TOLERANCE_MM + except ImportError: + return 50.0 + return DEFAULT_TOLERANCE_MM + + def main(argv: list[str] | None = None) -> int: args = parse_args(argv) @@ -1320,14 +1571,28 @@ def main(argv: list[str] | None = None) -> int: print(f"FEHLER: {csv_file} enthaelt keine Flussobjekte.", file=sys.stderr) return 1 + resolve_host = count_exits = None + method = None + if args.geometry: + try: + from tro_geometry import make_geometric_resolvers + except ImportError as exc: + print(f"FEHLER: --geometry braucht shapely (pip install shapely): {exc}", + file=sys.stderr) + return 1 + resolve_host, count_exits = make_geometric_resolvers(elements, tol_mm=args.geo_tol_mm) + method = f"geometrisch (STRtree, Toleranz {args.geo_tol_mm:.0f}mm)" + overrides = load_overrides(csv_file.name, env_dir("SKEL_CFG", "cfg")) - analysis = analyse_tros(graph, elements, overrides) + analysis = analyse_tros(graph, elements, overrides, + resolve_host=resolve_host, count_exits=count_exits, + method=method) if not analysis.tros: print(f"FEHLER: {csv_file} enthaelt keine Separatoren - keine TROs " f"ableitbar.", file=sys.stderr) return 1 - stem = csv_file.stem + stem = csv_file.stem + ("_geo" if args.geometry else "") dot_file = results / f"{stem}_tro_flow.dot" svg_file = results / f"{stem}_tro_flow.svg" doc_target = results / f"{stem}_tro_doc.md" diff --git a/lib/tro_geometry.py b/lib/tro_geometry.py new file mode 100644 index 0000000..2f04b71 --- /dev/null +++ b/lib/tro_geometry.py @@ -0,0 +1,338 @@ +# -*- coding: utf-8 -*- +# Absichtlich ohne Shebang: der Windows-py-Launcher wuerde daraus "python3" +# ableiten und scheitert, wenn keine PythonCore-Installation registriert ist. +""" +tro_geometry.py - Separator-Host und Weichen-Fanout direkt aus der Geometrie. + +Ergaenzt tro_flow.py::derive_tros() um einen zweiten Weg, zwei seiner Eingaben +zu bestimmen: + + Zuordnung (Host des Separators) -> resolve_host_geometric + Nachbarn/graph.edges (Fanout) -> count_geometric_exits + +Beide Werte kommen normalerweise aus den CSV-Spalten "Zuordnung"/"Nachbarn" +(material_flow.py::build_graph). Dieses Modul sucht sie stattdessen direkt in +den Weltkoordinaten der Bauteile: ein shapely-STRtree ueber die AS/ES-Enden +(K1/K2) jeder Strecke/Gefaellestrecke sowie die beiden Kreisel-Bahnenden +liefert per .nearest() den naechstgelegenen Host zu einem Separator und per +.query(..., predicate="dwithin") alle anderen Ketten, die sich an einem +gegebenen Punkt tatsaechlich treffen - unabhaengig davon, ob die betroffene +Zeile im CSV eine vollstaendige "Nachbarn"-Liste hat. + +Hintergrund: hundmM05_1.csv zeigt fuer mehrere "ILS 2.0 Strecke"-Zeilen den +Fehler "nur ein Partner" (Spalte Fehler) - die von dxfmakros berechnete +Nachbarschaft ist dort unvollstaendig. Eine fruehere, gleichartige Datei +(hundm_export.csv) zeigte den Ausfall dieser Abhaengigkeit in voller Wucht: +tro_flow.py leitete 5 TROs ab, alle vom Typ Vario, 0 Verbindungen, alle +isoliert (siehe results/hundm_export_tro_doc.md) - Zuordnung/Nachbarn liessen +sich fuer diesen Export nicht ausreichend aufloesen. Dieses Modul macht die +TRO-Ableitung fuer genau diesen Fall unabhaengig von der Qualitaet dieser +beiden Spalten. + +Nutzung: `tro_flow.py --geometry` (siehe dort, `make_geometric_resolvers`). +Dieses Modul kennt tro_flow.py nicht - nur material_flow.py::Element-Listen. +""" + +from __future__ import annotations + +import math +from dataclasses import dataclass + +from shapely import STRtree +from shapely.geometry import LineString, Point + +from material_flow import ( + LANE_LEFT, + LANE_RIGHT, + UNASSIGNED, + Element, + _as_float, + _merkmal, +) + +# Kreisel-Durchmesser (mm), falls kein Merkmal mit dem Durchmesser vorliegt - +# Default wie dxfmakros' cfg/export.cfg [Kreisel] durchmesser_mm. +DEFAULT_KREISEL_DURCHMESSER_MM = 800.0 + +# Merkmal-Namen, unter denen der Achsabstand der beiden Kreiselstationen +# (AN8/SP8) im CSV-Export auftaucht - in Metern, siehe kreisel_capsule(). +ABSTAND_MERKMALE = ( + "Abstand (Kreiselachse A - Kreiselachse) in Meter", + "Abstand_m", + "ABSTAND", +) + +# Default-Toleranz (mm), innerhalb derer zwei Anschlusspunkte als "derselbe +# Uebergabepunkt" gelten. Bewusst zwischen dxfmakros' omniflo_ks_toleranz_mm +# (10mm, enge KOS-Deckung von Omniflo-Steckverbindern) und vf_linienzug.lsp's +# *vfl-kette-tol-weit* (100mm, "sieht verbunden aus, ist es aber vielleicht +# nicht") - zwei Kettenenden an einem echten mechanischen Uebergang treffen +# sich nicht auf den zehntel Millimeter, muessen aber auch nicht 10cm +# auseinanderliegen duerfen, um noch als verbunden zu gelten. +DEFAULT_TOLERANCE_MM = 50.0 + + +@dataclass +class _EndpointEntry: + """Ein indizierter Anschlusspunkt: welches Bauteil, welches Ende.""" + + teile_id: str + end: str # "K1"/"K2" (AS/ES) oder "a"/"b" (Kreisel-Bahnende) + + +def chain_endpoints( + element: Element, +) -> tuple[tuple[float, float] | None, tuple[float, float] | None]: + """AS-(K1) und ES-(K2) Weltpunkt einer Strecke/Gefaellestrecke. + + Fehlen K1/K2 (aeltere Exporte ohne AS/ES-Punkte), Ersatz durch die Mitten + der kurzen Stirnseiten der Bounding-Box entlang ihrer laengeren Ausdehnung + - gleiche Idee wie dxfmakros' export_neighbors.py::_strecke_ends, nur ohne + dessen "Anfang liegt naeher am Einfuegepunkt"-Zuordnung: fuer die reine + Fanout-Zaehlung an einem Punkt (count_geometric_exits) ist unerheblich, + welches Ende AS und welches ES ist. + """ + by_label = dict(element.kpoints) + k1, k2 = by_label.get("K1"), by_label.get("K2") + if k1 is not None and k2 is not None: + return k1, k2 + if element.x is None or element.y is None or element.bbox is None: + return k1, k2 + cx, cy = element.x, element.y + dx, dy = element.bbox_x or 0.0, element.bbox_y or 0.0 + if dx >= dy: + bbox_ends = ((cx - dx / 2.0, cy), (cx + dx / 2.0, cy)) + else: + bbox_ends = ((cx, cy - dy / 2.0), (cx, cy + dy / 2.0)) + return k1 or bbox_ends[0], k2 or bbox_ends[1] + + +def kreisel_durchmesser_mm(element: Element) -> float: + """Kreisel-Durchmesser (mm) aus einem Merkmal, sonst DEFAULT_*.""" + for key in ("Durchmesser_mm", "Durchmesser (mm)", "Durchmesser"): + value = _as_float(_merkmal(element.merkmale, key)) + if value: + return value + return DEFAULT_KREISEL_DURCHMESSER_MM + + +def kreisel_capsule(element: Element, durchmesser_mm: float | None = None) -> dict | None: + """Achsen-Segment (Kapsel) eines Kreisels in Weltkoordinaten. + + dxfmakros' eigenes export_neighbors.py::_kreisel_capsule setzt die + Antriebsstation (AN8) lokal bei (radius, 0) und die Spannstation (SP8) bei + (radius+abstand, 0) - relativ zum BLOCK-Einfuegepunkt (lokaler Ursprung + des KREISEL_n-Wrappers, siehe KreiselInsert.lsp::draw-module). Die CSV- + Spalte "Position" fuehrt bei einem Kreisel jedoch nachweislich den + BBOX-MITTELPUNKT, nicht diesen Einfuegepunkt: die exportierte Boundingbox + ist exakt symmetrisch um Position (halbe Breite = (abstand+durchmesser)/2, + siehe hundmM05_1.csv Zeile 6/7 - AN8-lokal-(radius,0) waere das nicht). + AN8/SP8 werden daher symmetrisch um Position gelegt: AN8 bei + (-abstand/2, 0), SP8 bei (+abstand/2, 0) - beide auf der lokalen X-Achse, + gedreht um das Merkmal "Drehung" (Grad, CCW positiv). + + Rueckgabe {"a": (x,y), "b": (x,y), "r": radius} (a = AN8-, b = SP8-Mitte), + oder None ohne Position. + """ + if element.x is None or element.y is None: + return None + abstand_m = None + for key in ABSTAND_MERKMALE: + abstand_m = _as_float(_merkmal(element.merkmale, key)) + if abstand_m is not None: + break + abstand_mm = (abstand_m or 0.0) * 1000.0 + durchmesser = durchmesser_mm or kreisel_durchmesser_mm(element) + r = durchmesser / 2.0 + drehung = _as_float(_merkmal(element.merkmale, "Drehung")) or 0.0 + rad = math.radians(drehung) + cos_r, sin_r = math.cos(rad), math.sin(rad) + + def world(lx: float, ly: float) -> tuple[float, float]: + return (element.x + lx * cos_r - ly * sin_r, + element.y + lx * sin_r + ly * cos_r) + + half = abstand_mm / 2.0 + return {"a": world(-half, 0.0), "b": world(half, 0.0), "r": r} + + +def kreisel_lane_for_point(capsule: dict, point: tuple[float, float]) -> str: + """Welche Bahn (L/R) liegt ein Punkt bezogen auf die Kreiselachse a->b. + + Kreuzprodukt Achse x (Punkt - a): positiv = links der Achsrichtung + (mathematisch positive/CCW-Drehung von der Achse aus), konsistent mit + dxfmakros' "links" = lokal Y in [0, +radius] (siehe kreisel_half_bboxes). + """ + ax, ay = capsule["a"] + bx, by = capsule["b"] + px, py = point + cross = (bx - ax) * (py - ay) - (by - ay) * (px - ax) + return LANE_LEFT if cross >= 0 else LANE_RIGHT + + +class GeometryIndex: + """STRtree-Index der Bauteil-Geometrie fuer die geometrische TRO-Ableitung. + + Zwei getrennte Baeume: + _host_tree eine LineString je Gefaellestrecke/Strecke (K1->K2-Achse) + bzw. Kreisel (AN8->SP8-Achse), abgefragt mit .nearest() - + "welcher Bauteil-ACHSE liegt dieser Separator am + naechsten" (resolve_host_geometric). Bewusst die Achse, + nicht die lose Bounding-Box: ein Kreisel deckt mit + seiner Box leicht 20+ Meter ab (Antriebs- bis + Spannstation), ein Separator auf einer nahen VF-Kette + waere nach reiner Box-Distanz faelschlich "im Kreisel", + obwohl er auf der Kette sitzt (siehe Modul-Historie: + genau dieser Fehler zeigte sich beim ersten Testlauf + gegen hundmM05_1.csv). + _endpoint_tree ein Punkt je AS/ES-Ende einer Strecke/Gefaellestrecke + plus je Kreisel-Bahnende, abgefragt mit .query(..., + predicate="dwithin") - "wie viele andere Ketten treffen + sich an diesem Punkt" (count_geometric_exits). + """ + + def __init__(self, elements: list[Element], durchmesser_mm: float | None = None): + self.durchmesser_mm = durchmesser_mm + self._host_ids: list[str] = [] + host_axes = [] + # Kreisel-Kapseln nach TeileId, um am Treffer die Bahn (-L/-R) zu + # bestimmen - graph.nodes kennt Kreisel nur unter "-L"/"-R", + # nie unter der blanken TeileId (siehe material_flow.py::_build_nodes). + self._kreisel_capsules: dict[str, dict] = {} + self._endpoints: list[_EndpointEntry] = [] + endpoint_points = [] + + for element in elements: + if element.kind in ("Gefaellestrecke", "Strecke"): + k1, k2 = chain_endpoints(element) + if k1 is None or k2 is None: + continue + self._host_ids.append(element.teile_id) + host_axes.append(LineString([k1, k2])) + for label, point in (("K1", k1), ("K2", k2)): + self._endpoints.append(_EndpointEntry(element.teile_id, label)) + endpoint_points.append(Point(point)) + elif element.kind == "Kreisel": + if element.x is None or element.y is None: + continue + capsule = kreisel_capsule(element, durchmesser_mm) + if capsule is None: + continue + self._host_ids.append(element.teile_id) + host_axes.append(LineString([capsule["a"], capsule["b"]])) + self._kreisel_capsules[element.teile_id] = capsule + for end in ("a", "b"): + self._endpoints.append(_EndpointEntry(element.teile_id, end)) + endpoint_points.append(Point(capsule[end])) + + self._host_tree = STRtree(host_axes) if host_axes else None + self._endpoint_tree = STRtree(endpoint_points) if endpoint_points else None + + def nearest_host(self, x: float, y: float) -> str | None: + """Naechster Host-Knoten nach Achsabstand (Gefaellestrecke/Strecke- + TeileId, oder "-L"/"-R" - dieselbe Schreibweise wie + graph.nodes).""" + if self._host_tree is None: + return None + index = self._host_tree.nearest(Point(x, y)) + if index is None: + return None + teile_id = self._host_ids[int(index)] + capsule = self._kreisel_capsules.get(teile_id) + if capsule is None: + return teile_id + lane = kreisel_lane_for_point(capsule, (x, y)) + return f"{teile_id}-{lane}" + + def exits_near(self, x: float, y: float, tol_mm: float, exclude_ids: set[str]) -> set[str]: + """Andere Bauteil-Ids, deren Anschlusspunkt innerhalb tol_mm von (x,y) liegt.""" + if self._endpoint_tree is None: + return set() + hits = self._endpoint_tree.query(Point(x, y), predicate="dwithin", distance=tol_mm) + return {self._endpoints[int(i)].teile_id for i in hits} - exclude_ids + + +def resolve_host_geometric(separator: Element, index: GeometryIndex) -> str | None: + """Naechster Host zum Separator rein nach Weltkoordinate.""" + if separator.x is None or separator.y is None: + return None + return index.nearest_host(separator.x, separator.y) + + +def _strip_lane(host_id: str) -> str: + """"0006-L" -> "0006" (Kreisel-Bahnsuffix entfernen; alles andere unveraendert).""" + if host_id.endswith(("-L", "-R")): + return host_id[:-2] + return host_id + + +def count_geometric_exits( + hosts: list[str], + elements_by_id: dict[str, Element], + index: GeometryIndex, + tol_mm: float = DEFAULT_TOLERANCE_MM, +) -> list[str]: + """Andere Ketten, die geometrisch an einem Ende von `hosts` zusammentreffen. + + Entspricht tro_flow.py::_switch_exits, aber unabhaengig von graph.edges: + beide Enden JEDES Hosts (K1 und K2 bzw. beide Kreisel-Bahnenden) werden + abgefragt, die Treffer aller Enden vereinigt - wie eine aus beiden Enden + zusammengefuehrte "Nachbarn"-Spalte, nur direkt aus der Geometrie statt aus + der (fuer diese Bauteile teils unvollstaendigen) CSV-Spalte. + + `hosts` kann Kreisel-Bahnen als "-L"/"-R" enthalten (siehe + GeometryIndex.nearest_host) - die Endpunkt-Eintraege im Index fuehren dazu + nur die blanke TeileId, daher wird der Suffix hier wie dort entfernt. + """ + exclude = {_strip_lane(h) for h in hosts} + found: set[str] = set() + for host_id in hosts: + element = elements_by_id.get(_strip_lane(host_id)) + if element is None: + continue + if element.kind == "Kreisel": + capsule = kreisel_capsule(element, index.durchmesser_mm) + points = (capsule["a"], capsule["b"]) if capsule else () + else: + points = tuple(p for p in chain_endpoints(element) if p is not None) + for x, y in points: + found |= index.exits_near(x, y, tol_mm, exclude) + return sorted(found) + + +def make_geometric_resolvers( + elements: list[Element], + tol_mm: float = DEFAULT_TOLERANCE_MM, + durchmesser_mm: float | None = None, +): + """resolve_host/count_exits-Paar fuer tro_flow.py::derive_tros(...). + + Ersetzt Zuordnung (Spalte) und Nachbarn/graph.edges durch eine direkte + STRtree-Nachbarsuche ueber die Weltkoordinaten der Bauteile. Weicht der + geometrisch gefundene Host von der CSV-Spalte "Zuordnung" ab, wird das als + Hinweis protokolliert statt stillschweigend ignoriert - das deckt zugleich + auf, wo die CSV-Spalte fehlerhaft ist. + """ + index = GeometryIndex(elements, durchmesser_mm) + by_id = {e.teile_id: e for e in elements} + + def resolve_host(sep: Element, findings: list[str]) -> str | None: + host = resolve_host_geometric(sep, index) + if host is None: + findings.append( + f"Separator {sep.teile_id} '{sep.name}': geometrisch kein Host " + f"gefunden (CSV-Zuordnung '{sep.assignment or '-'}') - keinem " + f"TRO zugeordnet" + ) + return None + if sep.assignment and sep.assignment != UNASSIGNED and sep.assignment != host: + findings.append( + f"Separator {sep.teile_id} '{sep.name}': geometrischer Host " + f"'{host}' weicht von CSV-Zuordnung '{sep.assignment}' ab" + ) + return host + + def count_exits(hosts: list[str]) -> list[str]: + return count_geometric_exits(hosts, by_id, index, tol_mm) + + return resolve_host, count_exits diff --git a/requirements.txt b/requirements.txt index b014d57..74d5ddc 100644 --- a/requirements.txt +++ b/requirements.txt @@ -6,6 +6,10 @@ pydantic >= 2.0.0 # DXF lesen und schreiben (tro_annotate). DWG wird nicht unterstuetzt. ezdxf >= 1.4.0 +# STRtree fuer die geometrische TRO-Ableitung (tro_flow.py --geometry, siehe +# lib/tro_geometry.py) - nur zur Laufzeit von --geometry benoetigt. +shapely >= 2.1 + # Nur fuer Vorschaubilder eines DXF (ezdxf.addons.drawing) - optional: # matplotlib >= 3.8