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

tro_flow: --use-overrides, --from-connect und --geometry entfernt

Browse Source 
TROs, Typen und Verbindungen kommen nur noch aus dem Layout. Fehler der
Ableitung werden im Code behoben, nicht in einer Korrekturdatei.

- tro_flow: Optionen --use-overrides/--no-overrides, --from-connect,
  --geometry, --geo-tol-mm entfernt; analyse_tros/derive_tros ohne
  overrides/resolve_host/count_exits-Parameter, Analysis.overrides entfernt,
  Override-Schritte 4b/4c/5b, Doku-Abschnitt "6. Manuelle Korrekturen" und
  die Override-Zeilen im Bericht entfernt; connect.ini ist reine Ausgabe
- lib/tro_geometry.py geloescht (nur von --geometry genutzt), shapely aus
  requirements.txt
- tro_overrides: apply_splits/_merges/_connections/_overrides und
  load_connect_ini entfernt - nur noch load_overrides fuer
  material_flow.py --connect
- doc/Python_Scripts.md nachgezogen

Gegenprobe: DOT (beide Layouts) und connect.ini fuer HundM05 und Mubea
unveraendert (28 bzw. 26 Verbindungen), Doku nur ohne den leeren
Abschnitt 6.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Samer Ayadi
2026-10-05 16:02:16 +02:00
co-authored by Claude Opus 5.5
parent a5fab7e2dc
commit e0defad011
5 changed files with 49 additions and 935 deletions
Download Patch File Download Diff File Expand all files Collapse all files
doc/Python_Scripts.md
+22 -52
View File
@@ -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 (`<name>_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 `<name>_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%`): `<name>_tro_flow.dot` (always, with `pos`
attributes when `--use-cords` was used), `<name>_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 `<SPS_SKEL>/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`): `<name>_annotated.dxf`,
`<name>_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
`[<plant>.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 `[<plant>.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)
lib/tro_flow.py
+19 -263
View File
@@ -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. [<Anlage>.tros] ist eine reine Uebersicht (Vorgaenger/
Nachfolger je TRO) und wird beim Einlesen ignoriert.
sondern mit ihrer ID (SYSn). [<Anlage>.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("; [<Anlage>.nodes] TRO = <Typ> | <Bezeichnung> | <Bauteile>")
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: "
"<name>_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)
lib/tro_geometry.py
-338
View File
@@ -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 "<id>-L"/"<id>-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 "<Kreisel-TeileId>-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 "<TeileId>-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
lib/tro_overrides.py
+8 -278
View File
@@ -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
`[<Anlage>.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<Nummer> 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> | <JamArea> | <kind>' -> (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:
[<Anlage>.nodes] TRO-ID = <Typ> | ... -> Typkorrektur je TRO-ID
[<Anlage>.connections] cNNN = von -> nach | JamArea | kind
[<Anlage>.externals] eNNN = SYSn -> TRO | ... bzw. TRO -> SYSn | ...
-> zusammen die gesetzte Topologie, die die
abgeleiteten Kanten vollstaendig ersetzt
[<Anlage>.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 '<von> -> <nach>'"
)
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
requirements.txt
-4
View File
@@ -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