diff --git a/doc/Python_Scripts.md b/doc/Python_Scripts.md index cc8b7a3..1cd05c1 100644 --- a/doc/Python_Scripts.md +++ b/doc/Python_Scripts.md @@ -55,12 +55,8 @@ Supporting libraries (no CLI of their own, imported by the tools above): (rotation + offset) transform between the CSV export's coordinate system and the DXF drawing's. - **`lib/tro_overrides.py`** — reads the hand-maintained - `%SKEL_CFG%\tro_overrides.ini` (type corrections, open points, TRO merges) - that survive re-runs of `tro_flow.py`. See §6a. -- **`lib/tro_geometry.py`** — resolves a separator's host and a switch's exit - count for `tro_flow.py` **directly from the part world-coordinates** (shapely - `STRtree`) instead of from the CSV `Zuordnung`/`Nachbarn` columns, for exports - where those columns have gaps. Only used with `tro_flow.py --geometry`. See §6b. + `%SKEL_CFG%\tro_overrides.ini`; only `material_flow.py --connect` still uses + it. See §6a. Separate, documentation-only tool (not part of the generator pipeline): @@ -99,8 +95,7 @@ Conventions shared by the CLI tools: output file paths); non-fatal findings are printed as `?`/`!` lines and also do not stop the run. - Requires the packages pinned in `requirements.txt` (`pydantic>=2.0.0`; - `ezdxf>=1.4.0` for `tro_annotate.py` only; `shapely>=2.1` for - `tro_flow.py --geometry` only) — run `bin\install_py.bat` once. + `ezdxf>=1.4.0` for `tro_annotate.py` only) — run `bin\install_py.bat` once. - `--tosvg` needs Graphviz's `dot` (or `neato`, for `tro_flow.py --use-cords`) on `PATH`, or the `GRAPHVIZ_DOT` environment variable pointing at the `dot` executable. `bin/material_flow.bat`/`.sh` and `bin/tro_flow.bat`/`.sh` @@ -284,8 +279,12 @@ colors all come from `lib/tro_catalog.py` — nothing is hard-coded here. | `--tosvg` | — | off | Also render the TRO flow diagram to `.svg`. | | `--use-cords` (alias `--use-coords`) | — | off | Place every TRO at the centroid of the components it was built from and render with `neato -n` (fixed positions) instead of `dot`'s computed layout, so the diagram matches the plant layout. Only has an effect together with `--tosvg`; fails if any TRO has no coordinate. | | `--doc` | — | off | Also write a Markdown report (`_tro_doc.md`) with the TRO list, connections, findings and the full type catalog. | -| `--geometry` | — | off | Determine each separator's host and each switch's exit count (fanout) **from the part world-coordinates** via a shapely `STRtree` (`lib/tro_geometry.py`, see §6b) instead of from the CSV `Zuordnung`/`Nachbarn` columns. Use this for exports where those columns are incomplete (the columns are then bypassed entirely for these two inputs). Requires `shapely>=2.1`. | -| `--geo-tol-mm` | `MM` (float) | catalog default | Only with `--geometry`: the tolerance in mm within which two connection points count as the same transfer point. | +| `--connect` | — | off | Also write the TRO topology as `_tro_connect.ini` (nodes, connections, system boundaries, predecessors/successors per TRO). Pure output, never read back. | +| `--true-scale` | — | off | Only with `--use-cords`: draw to scale. By default empty stretches longer than 1500 mm between nodes are shortened (order and neighbourhood are kept). | + +TROs, types and connections always come from the layout alone; +`cfg/tro_overrides.ini` is not read. Errors in the derivation are fixed in +`lib/tro_flow.py`, not in a correction file. **Output** (in `%SKEL_RESULTS%`): `_tro_flow.dot` (always, with `pos` attributes when `--use-cords` was used), `_tro_flow.svg` (with @@ -342,9 +341,8 @@ overridden. | `--check` | — | off | Dry run: only analyze and report, write no output file. | | `--emit-lisp` | — | off | Regenerate `cad/tro_types.lsp` (the type list for the `TRO_INSERT`/`TRO_EDIT` dialogs) from `tro_catalog.py` and exit — no `--dxf` needed. Written under `%DXFM_LISP%` if set (next to the `dxfmakros` SSG_LIB modules that provide the dialogs), otherwise under `/cad`. | -Layers always carry the prefix `TRO_`. `cfg/tro_overrides.ini` is never read, -so the markers always match `tro_flow.py` (which reads it only with -`--use-overrides`). +Layers always carry the prefix `TRO_`. Like `tro_flow.py`, it never reads +`cfg/tro_overrides.ini`, so the markers always match the flow diagram. **Output** (in `%SKEL_RESULTS%`, unless `--check`): `_annotated.dxf`, `_registration.json` (the coordinate transform used, with its proof). @@ -467,6 +465,12 @@ free. Consumed by `tro_flow.py` and `tro_annotate.py`; not run directly. ## 6a. `tro_overrides.py` — manual corrections to the derived TROs (library, no CLI) +> **Since 2026-10-05 only `material_flow.py --connect` reads this file.** +> `tro_flow.py` and `tro_annotate.py` no longer apply corrections: the sections +> were keyed on running TRO numbers and went stale with every new export. +> Errors in the TRO derivation are fixed in `lib/tro_flow.py` instead. The +> format below is kept for `material_flow.py`. + `tro_flow.py` derives TRO types from the mechanical layout alone. Some things are simply not in the CSV export: a switch that the export does not carry, a scanner that lifts the type to `1Sep_SSCC`, or a spot that a walkthrough @@ -536,46 +540,12 @@ Three kinds of entry, deliberately different in weight: flow node is primary, the rest stay attached so the linking still sees both sides of a transition. Like splits, this renumbers. - **`openpoint`** — a spot reported in a review that is **not** turned into a - TRO. The structure stays untouched; the point is only marked in the doc - (section 6). This keeps "derived" and "asserted" distinguishable until the - spot has been checked in BricsCAD. + TRO. The structure stays untouched. -Consumers: `tro_flow.py` (only with `--use-overrides`: type corrections + doc -section 6), `material_flow.py` (writes the -`[.manual]` section into `connect.ini`). Unknown TRO IDs and types are -not applied silently — they are reported as warnings in the console output and -in doc section 6.3, which catches the common trap: **TRO IDs are running -numbers of one run and are only valid for the CSV named in the section.** Swap -the data basis and the numbers shift. - -## 6b. `tro_geometry.py` — geometric separator-host and switch-fanout (library, no CLI) - -`tro_flow.py` normally derives two of its inputs from the CSV columns -(`material_flow.py::build_graph`): - -``` -Zuordnung (host of the separator) -> which conveyor/Kreisel a separator sits on -Nachbarn / graph edges (fanout) -> how many exits a switch has (1Sep / 1Sep1Swi / 1Sep2Swi) -``` - -For some exports those columns are incomplete — e.g. `hundmM05_1.csv` reports -"nur ein Partner" for several `Strecke` rows, and the older `hundm_export.csv` -failed outright (5 TROs, all `Vario`, 0 connections, all isolated) because -`Zuordnung`/`Nachbarn` could not be resolved. This module derives the same two -values **directly from the part world-coordinates** instead, so the TRO -derivation no longer depends on the quality of those two columns: - -- a shapely `STRtree` over the AS/ES chain ends (K1/K2) of every - Strecke/Gefällestrecke plus the two Kreisel lane-ends; -- `resolve_host_geometric` — `.nearest()` to find the closest host to a - separator; -- `count_geometric_exits` — `.query(..., predicate="dwithin")` to find every - other chain that actually meets at a given point (within a tolerance, mm). - -It is activated by `tro_flow.py --geometry` (via `make_geometric_resolvers`); -`--geo-tol-mm` sets the meeting tolerance. This module knows nothing about -`tro_flow.py` — only `material_flow.py::Element` lists. Requires `shapely>=2.1` -(pinned in `requirements.txt`, needed only for this path). +Consumer: `material_flow.py --connect` (writes the `[.manual]` section +into `connect.ini`). Keep in mind: **TRO IDs are running numbers of one run and +are only valid for the CSV named in the section.** Swap the data basis and the +numbers shift. ## 7. `dxf_registration.py` — CSV↔DXF coordinate registration (library, no CLI) diff --git a/lib/tro_flow.py b/lib/tro_flow.py index f121533..9cc55b2 100644 --- a/lib/tro_flow.py +++ b/lib/tro_flow.py @@ -98,7 +98,6 @@ import csv import json import math import sys -from collections.abc import Callable from dataclasses import dataclass, field from datetime import datetime from pathlib import Path @@ -136,15 +135,6 @@ from tro_catalog import ( get_tro, mermaid_classdefs, ) -from tro_overrides import ( - Overrides, - apply_connections, - apply_merges, - apply_overrides, - apply_splits, - load_connect_ini, - load_overrides, -) # --------------------------------------------------------------------------- # Konstanten @@ -250,9 +240,7 @@ class Analysis: isolated: list[Tro] = field(default_factory=list) unknown_types: list[str] = field(default_factory=list) 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) + # Wie Host/Fanout ermittelt wurden (Doku/Bericht) method: str = "Materialfluss-Graph (Zuordnung/Nachbarn)" # Systemgrenzen (BTMT Be-/Entladestationen) - siehe derive_boundaries; kein # TRO, nur damit der tatsaechliche Materialein-/ausgang sichtbar wird. @@ -547,20 +535,13 @@ 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, sepliste_counts: dict[str, int] | None = None, ) -> list[Tro]: """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. + Host eines Separators aus der CSV-Spalte "Zuordnung" (_default_resolve_host), + abgehende Wege eines TRO-Hosts aus graph.edges (_switch_exits). sepliste_counts (optional, siehe load_sepliste_counts): Container-Id -> Anzahl 'SEP'-Eintraege aus dem Roh-JSON. Reine Gegenprobe - stimmt die aus @@ -568,8 +549,11 @@ def derive_tros( nicht mit dieser Zahl ueberein, wird ein Finding angehaengt; die TRO-Daten selbst bleiben unangetastet (CSV bleibt die massgebliche Quelle). """ - 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)) + def resolve_host(sep: Element, findings: list[str]) -> str | None: + return _default_resolve_host(sep, findings, graph) + + def count_exits(hosts: list[str]) -> list[str]: + return _switch_exits(hosts, graph) separators = sorted( (e for e in elements if e.kind == "Separator"), key=lambda e: e.teile_id @@ -969,41 +953,12 @@ def derive_tros( tro.reason = f"{tro.reason}; {len(exits)} abgehende Wege -> Weiche" tro.confidence = "niedrig" - # 4b) Manuell ergaenzte Uebergabestellen (cfg/tro_overrides.ini, [.split.]). - # Vor der Nummernvergabe, damit die neuen TROs in derselben Reihenfolge - # eine Nummer bekommen wie die abgeleiteten. - if overrides is not None and overrides.splits: - def _neuer_tro(spec, hosts, label): - return Tro( - tro_id="", - type_name=spec.type_name, - label=label, - hosts=hosts, - separators=list(spec.separators), - scanners=list(spec.scanners), - reason=spec.reason or "manuell ergaenzte Uebergabestelle", - confidence=spec.confidence, - ) - - apply_splits(tros, overrides, _neuer_tro) - - # 4c) Zusammenfassungen - nach den Splits, damit auch ein frisch - # herausgeloester TRO Teil einer Zusammenfassung sein kann. - if overrides is not None and overrides.merges: - apply_merges(tros, overrides) - # 5) Stabile, laufende Nummern in Reihenfolge der Host-Knoten tros.sort(key=lambda t: (t.hosts[0], t.separators[0] if t.separators else "")) width = max(2, len(str(len(tros)))) for index, tro in enumerate(tros, start=1): tro.tro_id = f"TRO{index:0{width}d}" - # 5b) Manuelle Korrekturen aus cfg/tro_overrides.ini. Sie greifen erst hier, - # weil sie ueber die TRO-ID adressiert sind, und noch vor Schritt 6, damit - # ein manuell gesetzter Typ den SSCC-Hinweis nicht mehr ausloest. - if overrides is not None: - apply_overrides(tros, overrides) - # 6) Scanner am Separator: 1Sep_SSCC laesst sich mechanisch nicht erkennen for tro in tros: if tro.type_name == "1Sep" and tro.scanners: @@ -1597,20 +1552,12 @@ def locate_tros(tros: list[Tro], elements: list[Element], graph: Graph) -> None: def analyse_tros( 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, sepliste_counts: dict[str, int] | None = None, ) -> Analysis: """TROs ermitteln, verknuepfen, verorten und gegen die Typdefinition pruefen.""" analysis = Analysis() - analysis.overrides = overrides if overrides is not None else 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.tros = derive_tros(graph, elements, analysis.findings, sepliste_counts=sepliste_counts) analysis.boundaries = derive_boundaries(elements, graph) # Fuer Verortung und Verknuepfung zaehlen TROs und Systemgrenzen zusammen: @@ -1623,25 +1570,6 @@ def analyse_tros( locate_tros(linked, elements, graph) analysis.edges, analysis.edge_kinds = link_tros(graph, linked, elements) - # 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( - linked, analysis.overrides - ) - gesetzt = set(analysis.edges) - for src, dst in sorted(abgeleitet - gesetzt): - analysis.findings.append( - f"abgeleitete Verbindung {src} -> {dst} entfaellt - steht nicht " - f"in der gesetzten Topologie" - ) - for src, dst in sorted(gesetzt - abgeleitet): - analysis.findings.append( - f"Verbindung {src} -> {dst} nur gesetzt, aus dem Layout nicht " - f"ableitbar" - ) - for tro in analysis.tros: definition = tro.definition() if definition: @@ -2020,29 +1948,26 @@ def render_dot(analysis: Analysis, source: str, use_coords: bool = False, # --------------------------------------------------------------------------- -# connect.ini (TRO-Ebene) - --connect schreibt, --from-connect liest +# connect.ini (TRO-Ebene) - --connect schreibt # --------------------------------------------------------------------------- def render_connect_ini(analysis: Analysis, section: str, source: str) -> str: """ TRO-Topologie als connect.ini, angelehnt an doc/TRO_Katalog/connect.ini - (hmf-connect-v1). Zurueck eingelesen von tro_overrides.load_connect_ini - (--from-connect): geaenderte Typen und Verbindungen gehen dann in DOT/SVG. + (hmf-connect-v1). Reine Ausgabe - wird von keinem Lauf zurueckgelesen. Abweichung vom Referenzformat: Systemgrenzen stehen nicht als EXTERN, - sondern mit ihrer ID (SYSn), damit die Datei verlustfrei zurueckgelesen - werden kann. [.tros] ist eine reine Uebersicht (Vorgaenger/ - Nachfolger je TRO) und wird beim Einlesen ignoriert. + sondern mit ihrer ID (SYSn). [.tros] ist eine reine Uebersicht + (Vorgaenger/Nachfolger je TRO). """ out: list[str] = [] add = out.append add("; ============================================================") add("; connect.ini - Routing / Zusammenhalt der TROs (TRO-Ebene)") add(f"; Automatisch erzeugt von lib/tro_flow.py aus '{source}'.") - add("; Wird bei jedem Lauf mit --connect neu geschrieben. Eine von Hand") - add("; geaenderte Fassung wird mit --from-connect gezeichnet; dauerhafte") - add("; Korrekturen gehoeren nach cfg/tro_overrides.ini.") + add("; Wird bei jedem Lauf mit --connect neu geschrieben - nicht von Hand") + add("; aendern, Korrekturen gehoeren in die Ableitung (lib/tro_flow.py).") add(";") add("; [.nodes] TRO = | | ") add("; SYSn = Materialeingang/-ausgang (kein TRO)") @@ -2250,11 +2175,6 @@ def render_doc( add(f"| TRO-Verbindungen | {len(analysis.edges)} |") add(f"| Nicht verbundene TROs | {len(analysis.isolated)} |") add(f"| Hinweise | {len(analysis.findings)} |") - if analysis.overrides.applied: - add(f"| Manuelle Typkorrekturen | {len(analysis.overrides.applied)} |") - if analysis.overrides.open_points: - add(f"| Offene Punkte (noch kein TRO) | " - f"{len(analysis.overrides.open_points)} |") add("") add("| Typ | FB-Baustein | Anzahl hier | Bauteile je TRO | Farbgruppe | Symbol |") add("|---|---|---:|---|---|---|") @@ -2360,56 +2280,7 @@ def render_doc( add("_Keine._") add("") - add("## 6. Manuelle Korrekturen und offene Punkte") - add("") - overrides = analysis.overrides - if not overrides and not overrides.warnings: - add("_Keine._ Alles unten Stehende ist rein aus dem Layout abgeleitet.") - add("") - else: - add(f"Quelle: `{overrides.source}`" if overrides.source - else "_Keine Override-Datei gefunden._") - add("") - - if overrides.applied: - add("### 6.1 Angewandte Typkorrekturen") - add("") - add("Diese Typen stammen **nicht** aus dem Layout, sondern aus der " - "Begehung bzw. dem Review - im Layout sind sie nicht erkennbar.") - add("") - add("| Korrektur | Begruendung |") - add("|---|---|") - for entry in overrides.applied: - head, _, tail = entry.partition(" (") - add(f"| `{head}` | {tail[:-1] if tail.endswith(')') else tail} |") - add("") - - if overrides.open_points: - add("### 6.2 Offene Punkte - noch kein TRO") - add("") - add("Im Review gemeldete Stellen. Die TRO-Struktur ist hier " - "**bewusst unveraendert**; die Punkte sind nur markiert (auch in der " - "annotierten Zeichnung, Layer `TRO_OPENPOINT`), damit sie in BricsCAD " - "geprueft werden koennen.") - add("") - add("| Punkt | X [mm] | Y [mm] | Z [mm] | erwartet | heute Teil von | Anmerkung |") - add("|---|---:|---:|---:|---|---|---|") - for point in overrides.open_points: - z_text = f"{point.z:.0f}" if point.z is not None else "-" - expect = f"`{point.expect}`" if point.expect else "-" - owner = f"`{point.belongs_to}`" if point.belongs_to else "-" - add(f"| {point.name} | {point.x:.0f} | {point.y:.0f} | {z_text} " - f"| {expect} | {owner} | {point.note or '-'} |") - add("") - - if overrides.warnings: - add("### 6.3 Probleme mit der Override-Datei") - add("") - for warning in overrides.warnings: - add(f"- {warning}") - add("") - - add("## 7. Ableitungsregeln und Typdefinition") + add("## 6. Ableitungsregeln und Typdefinition") add("") add("Ein TRO ist die Steuerungslogik **einer Uebergabestelle**. Ausgangspunkt " "ist der Separator und das, was an ihm haengt (Host-Objekt, Scanner).") @@ -2516,29 +2387,6 @@ def report( print("FEHLER: TRO-Typen ohne Eintrag in TRO_CATALOG: " + ", ".join(analysis.unknown_types)) - overrides = analysis.overrides - if overrides.applied: - print("") - print(f"Manuelle Typkorrekturen ({len(overrides.applied)}) aus " - f"{overrides.source}:") - for entry in overrides.applied: - print(f" * {entry}") - - if overrides.open_points: - print("") - print(f"Offene Punkte ({len(overrides.open_points)}) - nur markiert, " - f"TRO-Struktur unveraendert:") - for point in overrides.open_points: - expect = f" erwartet {point.expect}" if point.expect else "" - owner = f", heute Teil von {point.belongs_to}" if point.belongs_to else "" - print(f" o {point.name} bei {point.describe_position()}{expect}{owner}") - - if overrides.warnings: - print("") - print(f"Probleme mit {overrides.source} ({len(overrides.warnings)}):") - for warning in overrides.warnings: - print(f" ! {warning}") - if analysis.findings: print("") print(f"Hinweise ({len(analysis.findings)}):") @@ -2611,35 +2459,6 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace: "schreiben: Knoten (TRO = Typ), Verbindungen, Systemgrenzen und je " "TRO Vorgaenger/Nachfolger.", ) - parser.add_argument( - "--from-connect", - dest="from_connect", - nargs="?", - const="", - default=None, - metavar="INI", - help="Typen und Verbindungen aus einer (von Hand geaenderten) connect.ini " - "uebernehmen und damit DOT/SVG/Doku erzeugen. Ohne Wert: " - "_tro_connect.ini in %%SKEL_RESULTS%%. Die connect.ini wird " - "dabei nicht ueberschrieben.", - ) - 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", - ) parser.add_argument( "--true-scale", dest="true_scale", @@ -2648,34 +2467,9 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace: f"Leerstrecken ueber {COORD_MAX_LUECKE} mm zwischen den Knoten " "(Reihenfolge und Nachbarschaft bleiben erhalten).", ) - parser.add_argument( - "--use-overrides", - dest="use_overrides", - action="store_true", - help="Manuelle Korrekturen aus cfg/tro_overrides.ini anwenden. Standard " - "ist AUS: TROs, Typen und Verbindungen kommen rein aus dem Layout " - "(die INI-Abschnitte haengen an laufenden TRO-Nummern und veralten " - "mit jedem neuen Export).", - ) - parser.add_argument( - "--no-overrides", - dest="use_overrides", - action="store_false", - help=argparse.SUPPRESS, - ) 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) @@ -2709,48 +2503,15 @@ 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)" - sepliste_counts: dict[str, int] = {} raw_json_file = find_sibling_raw_json(csv_file) if raw_json_file is not None: sepliste_counts = load_sepliste_counts(raw_json_file, warnings) - if args.use_overrides: - overrides = load_overrides(csv_file.name, env_dir("SKEL_CFG", "cfg")) - else: - overrides = Overrides() - - stem = csv_file.stem + ("_geo" if args.geometry else "") + stem = csv_file.stem connect_target = results / f"{stem}_tro_connect.ini" - connect_source: Path | None = None - if args.from_connect is not None: - connect_source = Path(args.from_connect) if args.from_connect else connect_target - if not connect_source.is_absolute() and connect_source.parent == Path("."): - connect_source = results / connect_source - if not connect_source.is_file(): - print(f"FEHLER: connect.ini nicht gefunden: {connect_source}", file=sys.stderr) - print(" Erst mit --connect erzeugen.", file=sys.stderr) - return 1 - try: - overrides = load_connect_ini(connect_source, overrides) - except ValueError as exc: - print(f"FEHLER: {exc}", file=sys.stderr) - return 1 - analysis = analyse_tros(graph, elements, overrides, - resolve_host=resolve_host, count_exits=count_exits, - method=method, sepliste_counts=sepliste_counts) + analysis = analyse_tros(graph, elements, sepliste_counts=sepliste_counts) if not analysis.tros: print(f"FEHLER: {csv_file} enthaelt keine Separatoren - keine TROs " f"ableitbar.", file=sys.stderr) @@ -2777,7 +2538,7 @@ def main(argv: list[str] | None = None) -> int: ) connect_file = None - if args.connect and connect_source is None: + if args.connect: connect_target.write_text( render_connect_ini(analysis, section=stem, source=csv_file.name), encoding="utf-8", @@ -2809,11 +2570,6 @@ def main(argv: list[str] | None = None) -> int: warnings, engine=engine) if connect_file: print(f"connect.ini = {connect_file}") - if connect_source: - print(f"Gezeichnet aus = {connect_source}") - if args.connect and connect_source is not None: - print("HINWEIS: --connect zusammen mit --from-connect - die connect.ini " - "wird nicht ueberschrieben.") if svg_error: print(f"FEHLER: SVG nicht erzeugt: {svg_error}", file=sys.stderr) diff --git a/lib/tro_geometry.py b/lib/tro_geometry.py deleted file mode 100644 index 2f04b71..0000000 --- a/lib/tro_geometry.py +++ /dev/null @@ -1,338 +0,0 @@ -# -*- 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/lib/tro_overrides.py b/lib/tro_overrides.py index 3a72018..77503ba 100644 --- a/lib/tro_overrides.py +++ b/lib/tro_overrides.py @@ -10,10 +10,13 @@ eigener TRO erkannt wurde. Solche Erkenntnisse kommen aus der Begehung bzw. aus dem Review der erzeugten Diagramme. Damit sie den naechsten Lauf ueberleben, stehen sie in einer INI-Datei -(`%SKEL_CFG%/tro_overrides.ini`) statt in den erzeugten Dateien. `tro_flow.py` -und `tro_annotate.py` lesen sie und wenden sie nach der Ableitung an; die Doku -weist jede angewandte Korrektur getrennt aus, damit abgeleitet und behauptet -unterscheidbar bleiben. +(`%SKEL_CFG%/tro_overrides.ini`) statt in den erzeugten Dateien. + +Seit 2026-10-05 nur noch von `material_flow.py --connect` gelesen (Abschnitt +`[.manual]` der connect.ini). `tro_flow.py` und `tro_annotate.py` +wenden keine Korrekturen mehr an: die Abschnitte hingen an laufenden +TRO-Nummern und veralteten mit jedem Export - Fehler der Ableitung werden +seitdem im Code behoben (lib/tro_flow.py), nicht in der INI. Aufbau der INI (Sektionsnamen mit dem CSV-Dateinamen als Praefix, damit mehrere Anlagen in einer Datei stehen koennen): @@ -31,9 +34,7 @@ mehrere Anlagen in einer Datei stehen koennen): belongs_to = TRO10 TRO, der das Bauteil heute enthaelt note = ... Erlaeuterung -Offene Punkte aendern die TRO-Struktur ausdruecklich **nicht**. Sie werden nur -in der Doku und in der annotierten Zeichnung markiert, damit sie in BricsCAD -geprueft werden koennen, bevor daraus ein TRO wird. +Offene Punkte aendern die TRO-Struktur ausdruecklich **nicht**. """ from __future__ import annotations @@ -320,274 +321,3 @@ def load_overrides(csv_name: str, cfg_dir: Path | None = None) -> Overrides: result.open_points.sort(key=lambda p: p.name) return result - - -def apply_splits(tros, overrides: Overrides, make_tro) -> list: - """Neue TROs aus bestehenden herausloesen (vor der Nummernvergabe). - - `make_tro(spec, hosts, label)` erzeugt das projektspezifische Tro-Objekt - - so bleibt dieses Modul frei von einem Import aus tro_flow (das umgekehrt - schon von hier importiert). - """ - for spec in overrides.splits: - quellen = [t for t in tros if set(spec.separators) & set(t.separators)] - if not quellen: - overrides.warnings.append( - f"Split '{spec.name}': keiner der Separatoren " - f"{', '.join(spec.separators)} kommt in diesem Lauf vor - " - f"nicht angelegt" - ) - continue - - hosts: list[str] = [] - for quelle in quellen: - quelle.separators = [s for s in quelle.separators - if s not in spec.separators] - quelle.scanners = [s for s in quelle.scanners if s not in spec.scanners] - hosts.extend(quelle.hosts) - if not quelle.separators and quelle.type_name not in ("Vario",): - overrides.warnings.append( - f"Split '{spec.name}': {quelle.label} hat danach keinen " - f"Separator mehr - bitte pruefen" - ) - - ziel_hosts = [spec.host] if spec.host else sorted(set(hosts))[:1] - label = spec.label or ( - f"Separator {', '.join(spec.separators)} " - f"(manuell aus {quellen[0].label} herausgeloest)" - ) - tros.append(make_tro(spec, ziel_hosts, label)) - overrides.applied.append( - f"neu {spec.name}: {spec.type_name} auf {', '.join(ziel_hosts)} " - f"mit Separator {', '.join(spec.separators)}, herausgeloest aus " - f"{', '.join(sorted({q.label for q in quellen}))} " - f"({spec.reason or 'manuelle Ergaenzung'})" - ) - return tros - - -def apply_merges(tros, overrides: Overrides) -> list: - """Mehrere TROs zu einem zusammenfassen (nach den Splits, vor der Nummer). - - Der erste betroffene TRO bleibt bestehen und uebernimmt Bauteile, Hosts und - Typ; die uebrigen werden aus der Liste entfernt. - """ - for spec in overrides.merges: - betroffen = [t for t in tros if set(spec.separators) & set(t.separators)] - if len(betroffen) < 2: - gefunden = ", ".join(t.tro_id or t.label for t in betroffen) or "keiner" - overrides.warnings.append( - f"Merge '{spec.name}': nur {len(betroffen)} der Separatoren " - f"{', '.join(spec.separators)} gefunden ({gefunden}) - " - f"nicht zusammengefasst" - ) - continue - - ziel = betroffen[0] - quellen = betroffen[1:] - for quelle in quellen: - ziel.separators = sorted(set(ziel.separators) | set(quelle.separators)) - ziel.scanners = sorted(set(ziel.scanners) | set(quelle.scanners)) - ziel.hosts = sorted(set(ziel.hosts) | set(quelle.hosts)) - tros.remove(quelle) - - if spec.host: - # Ein Host bestimmt, wo der zusammengefasste TRO im Fluss haengt; - # die uebrigen Bahnen bleiben als weitere Hosts erhalten, damit die - # Verknuepfung beide Seiten des Uebergangs sieht. - ziel.hosts = [spec.host] + [h for h in ziel.hosts if h != spec.host] - ziel.type_name = spec.type_name - ziel.confidence = spec.confidence - ziel.label = spec.label or ziel.label - grund = spec.reason or "manuell zusammengefasst" - ziel.reason = f"{ziel.reason}; manuell zusammengefasst ({grund})" - overrides.applied.append( - f"merge {spec.name}: {len(betroffen)} TROs -> ein {spec.type_name} " - f"mit Separatoren {', '.join(ziel.separators)} ({grund})" - ) - return tros - - -def apply_connections(tros, overrides: Overrides): - """Gesetzte Topologie an die Stelle der abgeleiteten Kanten setzen. - - Ersetzt vollstaendig, nicht ergaenzend: wenn die Verbindungen vor Ort - aufgenommen wurden, sind die abgeleiteten damit erledigt - sonst blieben - genau die Kanten stehen, die korrigiert werden sollten. Was dabei wegfaellt - bzw. hinzukommt, steht anschliessend in overrides.applied. - - Rueckgabe: (edges, edge_kinds) wie von link_tros. - """ - bekannt = {t.tro_id for t in tros} - edges: list[tuple[str, str]] = [] - kinds: dict[tuple[str, str], str] = {} - for spec in overrides.connections: - for seite, wert in (("von", spec.src), ("nach", spec.dst)): - if wert not in bekannt: - overrides.warnings.append( - f"Verbindung {spec.src} -> {spec.dst}: {seite} '{wert}' ist " - f"kein TRO dieses Laufs - Verbindung ignoriert" - ) - break - else: - paar = (spec.src, spec.dst) - if paar in kinds: - continue - edges.append(paar) - kinds[paar] = spec.kind or "gesetzt" - - for tro in tros: - tro.successors = sorted(d for s, d in edges if s == tro.tro_id) - tro.predecessors = sorted(s for s, d in edges if d == tro.tro_id) - tro.colocated = [] - - ohne = [t.tro_id for t in tros if not t.successors and not t.predecessors] - if ohne: - overrides.warnings.append( - "ohne Verbindung in der gesetzten Topologie: " + ", ".join(sorted(ohne)) - ) - overrides.applied.append( - f"Topologie gesetzt: {len(edges)} Verbindungen aus der Override-Datei " - f"ersetzen die abgeleiteten" - ) - return sorted(edges), kinds - - -def apply_overrides(tros, overrides: Overrides) -> None: - """Typkorrekturen auf die abgeleitete TRO-Liste anwenden. - - Wird aufgerufen, nachdem die TRO-IDs vergeben sind (die Korrekturen sind - ueber die ID adressiert) und bevor FB-Baustein und Bauteilliste aus dem - Katalog gezogen werden - der Typwechsel zieht beides mit. - """ - if not overrides.tros: - return - for section_key, override in sorted(overrides.tros.items()): - treffer = [t for t in tros if override.matches(t)] - if not treffer: - wie = "Separator" if override.by_separator else "TRO-ID" - overrides.warnings.append( - f"[{section_key}]: kein TRO mit {wie} '{override.key}' in diesem " - f"Lauf - Korrektur nicht angewandt" - + ("" if override.by_separator else - " (Nummerierung geaendert? Besser ueber SEP adressieren)") - ) - continue - if len(treffer) > 1: - overrides.warnings.append( - f"[{section_key}]: trifft {len(treffer)} TROs " - f"({', '.join(t.tro_id for t in treffer)}) - auf alle angewandt" - ) - for tro in treffer: - if not override.type_name or override.type_name == tro.type_name: - continue - vorher = tro.type_name - tro.type_name = override.type_name - tro.confidence = override.confidence - grund = override.reason or "manuelle Korrektur" - tro.reason = (f"{tro.reason}; manuell auf {override.type_name} " - f"gesetzt ({grund})") - overrides.applied.append( - f"{tro.tro_id}: {vorher} -> {override.type_name} ({grund})" - ) - - -# --------------------------------------------------------------------------- -# connect.ini (TRO-Ebene) zurueck einlesen - tro_flow.py --from-connect -# --------------------------------------------------------------------------- - -# Vertrauensangabe fuer Typen, die aus einer (evtl. von Hand geaenderten) -# connect.ini stammen - in der Doku von Layout und Begehung unterscheidbar. -CONNECT_CONFIDENCE = "connect.ini" - - -def _connect_pair(raw: str) -> tuple[str, str, str] | None: - """' -> | | ' -> (von, nach, kind); None wenn - kein Pfeil. Die JamArea wird (noch) nicht ausgewertet.""" - pfeil, _, rest = str(raw).partition("|") - src, sep, dst = (t.strip() for t in pfeil.partition("->")) - if not sep or not src or not dst: - return None - felder = [f.strip() for f in rest.split("|")] - kind = felder[1] if len(felder) > 1 else "" - return src, dst, kind - - -def load_connect_ini(path: Path, base: Overrides | None = None) -> Overrides: - """ - Eine von tro_flow.py --connect geschriebene connect.ini als Korrekturen lesen. - - Die Datei wird dabei nicht als eigenes Modell behandelt, sondern auf die - vorhandenen Korrekturarten abgebildet - so laeuft sie durch dieselbe - Pipeline wie cfg/tro_overrides.ini: - - [.nodes] TRO-ID = | ... -> Typkorrektur je TRO-ID - [.connections] cNNN = von -> nach | JamArea | kind - [.externals] eNNN = SYSn -> TRO | ... bzw. TRO -> SYSn | ... - -> zusammen die gesetzte Topologie, die die - abgeleiteten Kanten vollstaendig ersetzt - - [.tros] (Vorgaenger/Nachfolger je TRO) ist nur eine Uebersicht und - wird ignoriert - massgeblich sind die Verbindungen. - - `base` (die Korrekturen aus cfg/tro_overrides.ini) bleibt erhalten; Typen - aus der connect.ini gewinnen, die Topologie der connect.ini ersetzt eine - dort gesetzte. - """ - result = base if base is not None else Overrides() - parser = configparser.ConfigParser(inline_comment_prefixes=(";",), - interpolation=None) - parser.optionxform = str # TRO-IDs nicht in Kleinschreibung wandeln - # utf-8-sig: von Hand gespeicherte Dateien (Windows-Editoren, PowerShell) - # tragen oft ein BOM, an dem configparser sonst in Zeile 1 scheitert. - try: - parser.read(path, encoding="utf-8-sig") - except configparser.Error as exc: - raise ValueError(f"{path.name} nicht lesbar: {exc}") from exc - - connections: list[ConnectionSpec] = [] - for section in parser.sections(): - art = section.rsplit(".", 1)[-1].lower() - values = parser[section] - if art == "nodes": - for tro_id, raw in values.items(): - type_name = str(raw).partition("|")[0].strip() - # Systemgrenzen (SYSn) sind kein TRO, ihr "Typ" ist nur Anzeige - if tro_id.upper().startswith("SYS") or not type_name: - continue - if get_tro(type_name) is None: - result.warnings.append( - f"{path.name} [{section}] {tro_id}: Typ '{type_name}' steht " - f"nicht im TRO-Katalog - Typ bleibt wie abgeleitet" - ) - continue - result.tros[tro_id.upper()] = TroOverride( - key=tro_id.upper(), type_name=type_name, - reason=f"aus {path.name}", confidence=CONNECT_CONFIDENCE, - ) - elif art in ("connections", "externals"): - for key, raw in values.items(): - pair = _connect_pair(raw) - if pair is None: - result.warnings.append( - f"{path.name} [{section}] {key}: '{raw}' ist keine " - f"Verbindung ' -> '" - ) - continue - src, dst, kind = pair - if "EXTERN" in (src.upper(), dst.upper()): - result.warnings.append( - f"{path.name} [{section}] {key}: EXTERN wird nicht " - f"ausgewertet - Systemgrenze als SYSn angeben" - ) - continue - connections.append(ConnectionSpec(src=src, dst=dst, kind=kind)) - - if connections: - result.connections = connections - else: - result.warnings.append( - f"{path.name}: keine Verbindungen gefunden - Topologie bleibt wie abgeleitet" - ) - result.applied.append(f"connect.ini eingelesen: {path}") - return result diff --git a/requirements.txt b/requirements.txt index 74d5ddc..b014d57 100644 --- a/requirements.txt +++ b/requirements.txt @@ -6,10 +6,6 @@ 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