diff --git a/CLAUDE.md b/CLAUDE.md index dab4c61..8312ab0 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -36,7 +36,9 @@ Template" below). What exists today is: `tro_annotate.py --emit-lisp`) - `cfg/dxf_registration.json` — persisted, verified CSV↔DXF coordinate transform; `cfg/tro_overrides.ini` — hand-maintained corrections to the derived TROs (see `tro_overrides.py`); - `cfg/e_planning.ini` — project-specific rules for assigning E-planning symbols to layout parts + `cfg/e_planning.ini` — project-specific rules for assigning E-planning symbols to layout parts; + `cfg/_tro_ids.ini` — per-plant TRO-ID file (versioned, IDs renamed by the SPS + programmers), created/extended by `tro_flow.py` via `lib/tro_ids.py` - `doc/*.md` — analysis documents that reverse-engineer the SCL patterns and propose the fuller JSON layout schema for the parts of the pipeline still to be built out @@ -179,7 +181,10 @@ byte — rewrite them (`SKEL_UPDATE_GOLDEN=1`) only for a reviewed, intended cha 1. **`lib/tro_flow.py --json`** (implemented) — writes `_tro.json` (schema `sps_skel-tro-v3`, `model.TroDocument`): TROs, system boundaries `SYSn`, connections with - their kind, findings. See `doc/Python_Scripts.md` §4. + their kind, findings. TRO-IDs come from `cfg/_tro_ids.ini` (`lib/tro_ids.py`): each + TRO is found again by the position of its separators, so IDs survive a new layout export + (TeileIds do not); `--renumber` renumbers along the material flow, `--ids` points another + export of the same plant at the same file. See `doc/Python_Scripts.md` §4. 2. **`lib/scl_skeleton.py`** (implemented) — reads that TRO-JSON through `TroDocument` (a wrong/renamed key or `items` not matching the catalog stops the run) and emits an `FB_Main` SCL skeleton (`_FB_Main.scl`): one `REGION` per TRO with the instance call and every diff --git a/doc/Konzept_Layout_Hardware_SCL.md b/doc/Konzept_Layout_Hardware_SCL.md index b5e79af..cbcbeff 100644 --- a/doc/Konzept_Layout_Hardware_SCL.md +++ b/doc/Konzept_Layout_Hardware_SCL.md @@ -220,26 +220,42 @@ stabil: Vergleich zweier Exporte derselben Anlage Stabil ist nur die **Lage** des Bauteils. -**Vorschlag:** +**Umgesetzt (2026-10-11, `lib/tro_ids.py`, Schalter in `tro_flow.py`; +Einzelheiten in `doc/Python_Scripts.md` §4 "TRO-IDs"):** -- Eine versionierte ID-Datei je Anlage, z. B. `cfg/_tro_ids.ini`: je TRO - die ID und die Lage seines Anker-Separators (x, y in mm), optional ein Name. - Von `tro_flow` angelegt, von den SPS-Programmierern umbenannt. +- Eine versionierte ID-Datei je Anlage, `cfg/_tro_ids.ini`, Abschnitt + `[tro]`: je TRO die ID und die Lage seines Anker-Separators (x, y in mm), der + Kommentar dahinter nennt Typ, Bezeichnung und Separatoren. Von `tro_flow` + angelegt, von den SPS-Programmierern umbenannt. Weitere Abschnitte (später + Separator- und Förderernummern) bleiben bei jedem Lauf erhalten. Ein anderer + Export derselben Anlage nutzt dieselbe Datei über `--ids`. ```ini - ; Beispiel, Werte erfunden - [mubea] - TRO107 = 12450.0, 3310.0 ; Kreisel 2 vor Beladung - TRO108 = 13250.0, 3310.0 + [tro] + TRO01 = -400.0, 13920.0 ; 1Sep1Swi | Separator 0003 auf Kreisel2 Bahn L | Sep 0003 + K0302_1 = 10310.0, -440.1 ; von Hand umbenannt ``` - **Standard (IDs halten):** `tro_flow` liest die ID-Datei und findet jeden TRO - über die Lage des Anker-Separators wieder (Toleranz einstellbar). Gefundene TROs - behalten ihre ID. Neue TROs bekommen eine vorläufige ID (nächste freie Nummer - entlang des Materialflusses) und stehen im Bericht; weggefallene TROs stehen im - Bericht, ihre ID wird nicht automatisch wiederverwendet. + über die Lage seiner Separatoren wieder (Toleranz `--id-tolerance`, Standard + 100 mm, nächster Treffer zuerst, je TRO ein Eintrag). Gefundene TROs behalten + ihre ID, auch umbenannte. Neue TROs bekommen die nächste freie Nummer und + werden an `[tro]` angehängt; weggefallene stehen im Bericht (mit dem TRO, der + ihren Separator jetzt enthält), bleiben in der Datei, ihre ID wird nicht + wiederverwendet. `tro_annotate` liest dieselbe Datei, schreibt sie aber nicht. - **Programmschalter `--renumber` (frühe Phase):** alle IDs neu entlang des - Materialflusses vergeben und die ID-Datei neu schreiben. + Materialflusses vergeben (Tiefensuche ab den Beladestationen, auf Kreisel und + Strecke in `TrackIds`-Reihenfolge) und `[tro]` neu schreiben. Ohne ID-Datei + geschieht dasselbe beim ersten Lauf. +- **Geprüft** mit `mubea.csv` → `mubea_export.csv` (andere TeileIds): 18 TROs + wiedergefunden, darunter eine umbenannte ID; `TRO14` als weggefallen gemeldet, + weil der Export die Separatoren 0005 und 0036 zu einem `2Sep2Swi` (`TRO13`) + zusammenfasst. +- **Noch offen:** die Abschnitte `[mubea.csv.tro.TRO03]` usw. in + `cfg/tro_overrides.ini` hängen an der alten Nummerierung und passen schon + heute nicht mehr (`TRO03` ist dort Sep 0010, im aktuellen Lauf Sep 0011). + Sie werden auf die IDs der ID-Datei umgestellt, sobald `tro_flow` die + Overrides wieder liest. - Dadurch werden auch die Abschnitte `[.tro.]` in `cfg/tro_overrides.ini` und die Zuordnungen von `hw_assign` stabil. @@ -317,6 +333,7 @@ unverändert. 1. **TRO-IDs stabil machen:** ID-Datei, Wiederfinden über die Lage, `--renumber`. Darauf bauen Overrides, Staubereiche und Zuordnungen auf. + **Erledigt** (2026-10-11). 2. Slots und `jamAreas` in die TRO-JSON (Umbau 1), ein Eingangs-Staubereich je TRO — ohne Slots hat die Zuordnung kein Ziel. 3. `hw_import` mit dem TIA-Import als erster Quelle. @@ -335,7 +352,8 @@ erzeugen, `doc/Python_Scripts.md` ergänzen. - **Staubereiche:** woher die Separatornummer (`Sep1024` → `J1024`) stammt; wo auf einem Kreisel die Zählung `` beginnt (Referenz K0102 prüfen). Das SCL-Schema ist festgelegt (UH01–UH05). -- **Toleranz** beim Wiederfinden der TROs über die Lage; Verhalten, wenn ein - Separator weiter verschoben wurde (Vorschlag: als "neu" + "weggefallen" melden, - der Programmierer übernimmt die ID von Hand). +- **Toleranz** beim Wiederfinden der TROs über die Lage: vorläufig 100 mm + (`--id-tolerance`). Ein weiter verschobener Separator wird als "neu" und + "weggefallen" gemeldet, der Programmierer übernimmt die ID von Hand in der + ID-Datei. Mit echten Layoutänderungen prüfen, ob der Wert passt. - **Korrekturen in BricsCAD oder in der JSON:** zurückgestellt. diff --git a/doc/Python_Scripts.md b/doc/Python_Scripts.md index c5efc6a..cae8568 100644 --- a/doc/Python_Scripts.md +++ b/doc/Python_Scripts.md @@ -73,6 +73,10 @@ Supporting libraries (no CLI of their own, imported by the tools above): - **`lib/dxf_registration.py`** — derives, verifies and persists the rigid (rotation + offset) transform between the CSV export's coordinate system and the DXF drawing's. +- **`lib/tro_ids.py`** — the per-plant ID file `%SKEL_CFG%\_tro_ids.ini`: + reads it, finds each TRO again by the position of its separators, assigns + new IDs, writes/extends the file. Used by `tro_flow.py` (read and write) and + `tro_annotate.py` (read only). See §4, "TRO-IDs". - **`lib/tro_overrides.py`** — reads the hand-maintained `%SKEL_CFG%\tro_overrides.ini`; only `material_flow.py --connect` still uses it. See §6a. @@ -305,16 +309,44 @@ colors all come from `lib/tro_catalog.py` — nothing is hard-coded here. | `--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). | | `--json` | — | off | Also write the TRO-JSON `_tro.json` (schema `sps_skel-tro-v3`, model `TroDocument` in `lib/model.py`): TROs, system boundaries (`SYSn`), connections with their kind, findings. Input for `scl_skeleton.py`. | +| `--ids` | `DATEI` | `_tro_ids.ini` | ID file of the plant (name in `%SKEL_CFG%`, or a path). Use it to run another export of the same plant against the same IDs, e.g. `--file mubea_export.csv --ids mubea_tro_ids.ini`. | +| `--renumber` | — | off | Number all TROs afresh along the material flow and rewrite the `[tro]` section of the ID file (early project phase). Renamed IDs are lost; other sections of the file are kept. | +| `--id-tolerance` | `MM` (float) | `100.0` | Maximum distance between an ID-file entry and a separator of a TRO for the TRO to count as found again. | 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. +**TRO-IDs** (`lib/tro_ids.py`). The IDs belong to the SPS programmers and must +survive a new layout export; the layout `TeileId` changes between exports, the +position does not. So every run reads `%SKEL_CFG%\_tro_ids.ini`: + +```ini +[tro] +TRO01 = -400.0, 13920.0 ; 1Sep1Swi | Separator 0003 auf Kreisel2 Bahn L | Sep 0003 +K0302_1 = 10310.0, -440.1 ; renamed by hand +``` + +- Each entry is `ID = x, y` (mm, CSV coordinates) of the TRO's anchor + separator. An entry finds the TRO with a separator within `--id-tolerance` + (closest first, one entry per TRO), and that TRO keeps the ID — renamed IDs + included. An ID must be unique and a valid SCL identifier; `SYSn` is + reserved for system boundaries. +- No file yet: the TROs are numbered along the material flow (depth-first from + the loading stations over the material-flow graph, on a carousel or Strecke in + `TrackIds` order) and the file is created. +- A TRO not found gets the next free `TRO` and is appended to `[tro]`; + an entry without a TRO is reported ("weggefallen", with the TRO that now + holds its separator, e.g. after two TROs merged into a `2Sep2Swi`) and stays + in the file; its ID is not reused. +- The console report lists found / new / dropped IDs. + **Output** (in `%SKEL_RESULTS%`): `_tro_flow.dot` (always, with `pos` attributes when `--use-cords` was used), `_tro_flow.svg` (with `--tosvg`), `_tro_doc.md` (with `--doc`), `_tro.json` (with `--json`), `_tro_errors.md` (only if TROs are unconnected or of an -unknown type). +unknown type). In `%SKEL_CFG%`: `_tro_ids.ini` (created, extended or, +with `--renumber`, rewritten — see above). **Exit codes:** `0` ok · `1` input/CLI error · `2` SVG rendering failed (Graphviz) · `3` TRO validation errors found (error file written). @@ -365,6 +397,7 @@ overridden. | `--legend` | — | off | Draw a legend of the color groups. | | `--marker-size` | `MM` (float) | `300.0` | Marker radius in mm. | | `--text-height` | `MM` (float) | `220.0` | Text height in mm. | +| `--ids` | `DATEI` | `_tro_ids.ini` | ID file as for `tro_flow.py`, **read only**: the markers carry the same IDs as the TRO-JSON. Without the file the TROs are numbered along the material flow, as `tro_flow.py` would. | | `--offset` | `DX,DY` | — | Set the CSV↔DXF offset by hand instead of deriving/loading it. For a drawing built with dxfmakros (CSV exported from the same drawing) this is `0,0`. | | `--re-register` | — | off | Re-derive the coordinate offset even if a verified one is already cached in `%SKEL_CFG%`. | | `--check` | — | off | Dry run: only analyze and report, write no output file. | diff --git a/lib/tro_annotate.py b/lib/tro_annotate.py index 13f9c75..bd4a9fb 100644 --- a/lib/tro_annotate.py +++ b/lib/tro_annotate.py @@ -116,7 +116,9 @@ from tro_flow import ( analyse_tros, find_sibling_raw_json, load_sepliste_counts, + resolve_ids_file, ) +from tro_ids import TroIds # --------------------------------------------------------------------------- # Konstanten @@ -844,6 +846,9 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace: metavar="MM", help="Markerradius in mm. Standard: %(default)s") parser.add_argument("--text-height", type=float, default=DEFAULT_TEXT, metavar="MM", help="Texthoehe in mm. Standard: %(default)s") + parser.add_argument("--ids", metavar="DATEI", + help="ID-Datei der TROs wie bei tro_flow (nur gelesen). " + "Standard: _tro_ids.ini in %%SKEL_CFG%%.") parser.add_argument("--offset", metavar="DX,DY", help="Lagebezug von Hand setzen statt ermitteln.") parser.add_argument("--re-register", action="store_true", @@ -907,12 +912,21 @@ def main(argv: list[str] | None = None) -> int: return 1 graph = build_graph(elements, csv_file.name, read_warnings) # Derselbe Aufruf wie in tro_flow.main (ohne cfg/tro_overrides.ini, mit - # der Separator-Gegenprobe aus _raw.json), damit die Zeichnung genau + # der Separator-Gegenprobe aus _raw.json und der ID-Datei), damit die Zeichnung genau # die TROs der TRO-JSON zeigt - sie ist nur Ansicht, keine Datenquelle. raw_json_file = find_sibling_raw_json(csv_file) sepliste_counts = (load_sepliste_counts(raw_json_file, read_warnings) if raw_json_file is not None else {}) - analysis = analyse_tros(graph, elements, sepliste_counts=sepliste_counts) + # Dieselbe ID-Datei wie tro_flow, aber nur lesen: neue TROs bekommen + # dieselbe vorlaeufige ID, die tro_flow anhaengen wuerde. + ids_file = resolve_ids_file(args.ids, csv_file) + try: + tro_ids = TroIds.load(ids_file) if ids_file.is_file() else None + except (OSError, ValueError) as exc: + print(f"FEHLER: ID-Datei nicht lesbar: {exc}", file=sys.stderr) + return 1 + analysis = analyse_tros(graph, elements, sepliste_counts=sepliste_counts, + tro_ids=tro_ids) if not analysis.tros: print(f"FEHLER: keine TROs aus {csv_file} ableitbar.", file=sys.stderr) return 1 diff --git a/lib/tro_flow.py b/lib/tro_flow.py index 5c2fa63..a3e66fa 100644 --- a/lib/tro_flow.py +++ b/lib/tro_flow.py @@ -90,6 +90,14 @@ Ausgabe (alles in SKEL_RESULTS) fuer scl_skeleton.py _tro_errors.md nur wenn TRO-Fehler gefunden wurden +TRO-IDs (ID-Datei in SKEL_CFG) +------------------------------ +Die IDs kommen aus %SKEL_CFG%/_tro_ids.ini (lib/tro_ids.py): jeder TRO +wird ueber die Lage seiner Separatoren wiedergefunden und behaelt seine ID, +auch wenn die SPS-Programmierer sie umbenannt haben. Fehlt die Datei, werden +die TROs entlang des Materialflusses durchnummeriert und die Datei angelegt; +neue TROs werden angehaengt. --renumber nummeriert alles neu. + Aufruf ueber bin/tro_flow.bat bzw. bin/tro_flow.sh. """ @@ -126,6 +134,16 @@ from material_flow import ( resolve_input, ) from model import Analysis, DerivedTro, TroDocument +from tro_ids import ( + DEFAULT_TOLERANCE, + TroIds, + assign_ids, + ids_path, + insert_entries, + other_sections, + render_ids, + render_new_entries, +) from tro_catalog import ( CATALOG_AS_OF, ITEM_FB, @@ -655,6 +673,7 @@ def derive_tros( findings: list[str], *, sepliste_counts: dict[str, int] | None = None, + tro_ids: TroIds | None = None, ) -> list[DerivedTro]: """TRO-Liste aus den Separatoren und ihren angeschlossenen Objekten bilden. @@ -666,6 +685,10 @@ def derive_tros( der CSV abgeleitete Separatorenzahl eines Strecke/Gefaellestrecke-Hosts nicht mit dieser Zahl ueberein, wird ein Finding angehaengt; die TRO-Daten selbst bleiben unangetastet (CSV bleibt die massgebliche Quelle). + + tro_ids (optional): ID-Datei der Anlage. Ohne sie werden die TROs entlang + des Materialflusses neu durchnummeriert; mit ihr behalten wiedergefundene + TROs ihre ID (Ergebnis in tro_ids.result). """ def resolve_host(sep: Element, findings: list[str]) -> str | None: return _default_resolve_host(sep, findings, graph) @@ -1156,11 +1179,16 @@ def derive_tros( tro.reason = f"{tro.reason}; {len(exits)} abgehende Wege -> Weiche" tro.confidence = "niedrig" - # 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}" + # 5) IDs: Reihenfolge entlang des Materialflusses, wiedergefundene TROs + # behalten ihre ID aus der ID-Datei (siehe lib/tro_ids.py) + tros.sort(key=_flow_key(graph, elements_by_id)) + anchors = separator_anchors(elements) + if tro_ids is None: + assign_ids(tros, anchors, [], renumber=True) + else: + tro_ids.result = assign_ids( + tros, anchors, tro_ids.entries, + tolerance=tro_ids.tolerance, renumber=tro_ids.renumber) # 6) Scanner am Separator: 1Sep_SSCC laesst sich mechanisch nicht erkennen for tro in tros: @@ -1174,6 +1202,56 @@ def derive_tros( return tros +def separator_anchors(elements: list[Element]) -> dict[str, tuple[float, float]]: + """Separator-TeileId -> (x, y) in mm, Bezugslage fuer die ID-Datei.""" + return {e.teile_id: (e.position[0], e.position[1]) + for e in elements if e.kind == "Separator" and e.position} + + +def _flow_rank(graph: Graph) -> dict[str, int]: + """Knoten -> Rang entlang des Materialflusses (Tiefensuche). + + Start an den Materialeingaengen (Beladung/Eingang), danach an jedem noch + nicht erreichten Knoten - eine geschlossene Anlage ohne Station wird so + trotzdem vollstaendig erfasst. Gleichrangige Nachfolger nach Knotennamen. + """ + succ: dict[str, list[str]] = {} + for edge in graph.edges: + succ.setdefault(edge.src, []).append(edge.dst) + starts = sorted(n for n, node in graph.nodes.items() if node.element.is_entry) + rank: dict[str, int] = {} + for start in starts + sorted(graph.nodes): + stack = [start] + while stack: + node = stack.pop() + if node in rank: + continue + rank[node] = len(rank) + stack.extend(n for n in sorted(succ.get(node, []), reverse=True) + if n not in rank) + return rank + + +def _flow_key(graph: Graph, elements_by_id: dict[str, Element]): + """Sortierschluessel fuer TROs entlang des Materialflusses. + + Erst der Rang des Host-Knotens, dann die Lage des Separators in der + TrackIds-Kette des Hosts (Kreisel: Umlaufrichtung, Strecke: Baureihenfolge). + """ + rank = _flow_rank(graph) + + def key(tro: DerivedTro) -> tuple: + host_rank = min((rank.get(h, len(rank)) for h in tro.hosts), + default=len(rank)) + sep = tro.separators[0] if tro.separators else "" + host = elements_by_id.get(tro.hosts[0]) if tro.hosts else None + track = host.track_ids if host else [] + in_track = track.index(sep) if sep in track else len(track) + return (host_rank, in_track, sep) + + return key + + # Typnamen fuer Systemgrenz-Knoten (siehe derive_boundaries) - bewusst NICHT # in TRO_CATALOG: eine Be-/Entladestation (BTMT oder generisches Eingang/ # Ausgang-Symbol) ist kein Steuerobjekt mit eigenem FB-Baustein, nur der @@ -1781,11 +1859,15 @@ def analyse_tros( elements: list[Element], *, sepliste_counts: dict[str, int] | None = None, + tro_ids: TroIds | None = None, ) -> Analysis: - """TROs ermitteln, verknuepfen, verorten und gegen die Typdefinition pruefen.""" + """TROs ermitteln, verknuepfen, verorten und gegen die Typdefinition pruefen. + + tro_ids: ID-Datei der Anlage (siehe derive_tros); None = neu nummerieren. + """ analysis = Analysis() analysis.tros = derive_tros(graph, elements, analysis.findings, - sepliste_counts=sepliste_counts) + sepliste_counts=sepliste_counts, tro_ids=tro_ids) analysis.boundaries = derive_boundaries(elements, graph) # Fuer Verortung und Verknuepfung zaehlen TROs und Systemgrenzen zusammen: # link_tros() registriert eine Kante nur zu einem Knoten, der Host eines @@ -2638,6 +2720,8 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace: " tro_flow.bat --file mubea.csv --tosvg --doc\n" " tro_flow.bat --file mubea.csv --tosvg --use-cords\n" " tro_flow.bat --file mubea.csv --json\n" + " tro_flow.bat --file mubea_export.csv --ids mubea_tro_ids.ini --json\n" + " tro_flow.bat --file mubea.csv --renumber\n" "\n" "Exit-Codes:\n" " 0 ok\n" @@ -2694,9 +2778,93 @@ 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( + "--ids", + metavar="DATEI", + help="ID-Datei der Anlage (Name in %%SKEL_CFG%% oder vollstaendiger " + "Pfad). Standard: _tro_ids.ini in %%SKEL_CFG%%. Fehlt sie, " + "wird sie angelegt.", + ) + parser.add_argument( + "--renumber", + action="store_true", + help="Alle TRO-IDs neu entlang des Materialflusses vergeben und die " + "Abschnitt [tro] der ID-Datei neu schreiben (fruehe Projektphase - " + "umbenannte IDs gehen verloren, andere Abschnitte bleiben).", + ) + parser.add_argument( + "--id-tolerance", + dest="id_tolerance", + type=float, + default=DEFAULT_TOLERANCE, + metavar="MM", + help="Hoechster Abstand (mm) zwischen Eintrag der ID-Datei und " + "Separator, bis zu dem ein TRO als wiedergefunden gilt. " + "Standard: %(default)s", + ) return parser.parse_args(argv) +def resolve_ids_file(arg: str | None, csv_file: Path) -> Path: + """Pfad der ID-Datei: --ids (Name in SKEL_CFG oder Pfad), sonst Standard.""" + cfg_dir = env_dir("SKEL_CFG", "cfg") + if not arg: + return ids_path(csv_file.stem, cfg_dir) + path = Path(arg) + if path.is_absolute() or path.parent != Path("."): + return path + return cfg_dir / path + + +def write_ids_file(tro_ids: TroIds, analysis: Analysis, elements: list[Element], + source: str) -> str: + """ID-Datei anlegen, neu schreiben oder neue TROs anhaengen. + + Rueckgabe: kurze Beschreibung fuer den Bericht. + """ + result = tro_ids.result + if result is None: + return "unveraendert" + anchors = separator_anchors(elements) + if result.renumbered: + tro_ids.path.parent.mkdir(parents=True, exist_ok=True) + rest = other_sections(tro_ids.path) + text = render_ids(analysis.tros, anchors, source) + tro_ids.path.write_text(text + (f"\n{rest}\n" if rest else ""), + encoding="utf-8") + return "neu geschrieben" if tro_ids.exists else "angelegt" + if result.new: + insert_entries(tro_ids.path, + render_new_entries(analysis.tros, anchors, result.new, source)) + return f"{len(result.new)} neue(r) TRO(s) angehaengt" + return "unveraendert" + + +def report_ids(tro_ids: TroIds, status: str) -> None: + """Abschnitt TRO-IDs im Konsolenbericht.""" + result = tro_ids.result + if result is None: + return + print(f"TRO-IDs = {tro_ids.path} ({status})") + if result.renumbered: + grund = "--renumber" if tro_ids.renumber else "keine ID-Datei" + print(f" neu entlang des Materialflusses nummeriert " + f"({grund})") + return + print(f" wiedergefunden: {len(result.kept)}, " + f"neu: {len(result.new)}, weggefallen: {len(result.missing)} " + f"(Toleranz {tro_ids.tolerance:g} mm)") + if result.new: + print(f" + neu (vorlaeufige ID, in der ID-Datei umbenennbar): " + f"{', '.join(result.new)}") + for entry in result.missing: + merged = result.merged_into.get(entry.tro_id) + grund = (f"Separator gehoert jetzt zu {merged}" if merged else + f"kein Separator innerhalb {tro_ids.tolerance:g} mm") + print(f" - weggefallen: {entry.tro_id} bei {entry.x:.1f}, {entry.y:.1f} " + f"({grund}; Eintrag bleibt stehen, ID wird nicht wiederverwendet)") + + def main(argv: list[str] | None = None) -> int: args = parse_args(argv) @@ -2739,11 +2907,20 @@ def main(argv: list[str] | None = None) -> int: connect_target = results / f"{stem}_tro_connect.ini" json_target = results / f"{stem}_tro.json" - analysis = analyse_tros(graph, elements, sepliste_counts=sepliste_counts) + try: + tro_ids = TroIds.load(resolve_ids_file(args.ids, csv_file), + tolerance=args.id_tolerance, renumber=args.renumber) + except (OSError, ValueError) as exc: + print(f"FEHLER: ID-Datei nicht lesbar: {exc}", file=sys.stderr) + return 1 + + analysis = analyse_tros(graph, elements, sepliste_counts=sepliste_counts, + tro_ids=tro_ids) if not analysis.tros: print(f"FEHLER: {csv_file} enthaelt keine Separatoren - keine TROs " f"ableitbar.", file=sys.stderr) return 1 + ids_status = write_ids_file(tro_ids, analysis, elements, csv_file.name) dot_file = results / f"{stem}_tro_flow.dot" svg_file = results / f"{stem}_tro_flow.svg" @@ -2806,6 +2983,7 @@ def main(argv: list[str] | None = None) -> int: report(analysis, csv_file, dot_file, created_svg, error_file, doc_file, warnings, engine=engine) + report_ids(tro_ids, ids_status) if connect_file: print(f"connect.ini = {connect_file}") if json_file: diff --git a/lib/tro_ids.py b/lib/tro_ids.py new file mode 100644 index 0000000..d70bab4 --- /dev/null +++ b/lib/tro_ids.py @@ -0,0 +1,309 @@ +# -*- coding: utf-8 -*- +""" +tro_ids.py - stabile TRO-IDs ueber eine ID-Datei je Anlage. + +Die TRO-IDs vergeben die SPS-Programmierer; eine ID muss nur eindeutig sein. +Sobald SCL-Code steht, darf sich eine ID nicht mehr aendern, nur weil das +Layout neu exportiert wurde. Die Layout-TeileId taugt dafuer nicht (sie +wechselt zwischen zwei Exporten derselben Anlage), stabil ist nur die Lage. +Siehe doc/Konzept_Layout_Hardware_SCL.md, "Festlegungen" -> "TRO-IDs". + +Die ID-Datei (`%SKEL_CFG%/_tro_ids.ini`, versioniert, von Hand +umbenennbar) haelt je TRO die ID und die Lage seines Anker-Separators: + + [tro] + TRO01 = 1234.5, -567.8 ; 1Sep1Swi | Separator 0003 auf Kreisel2 Bahn L + K0102_1 = 2210.0, 410.0 ; von Hand umbenannt + +Ablauf (assign_ids): + +- ohne ID-Datei bzw. mit renumber: alle TROs in der uebergebenen Reihenfolge + (entlang des Materialflusses) neu durchnummerieren, TRO01, TRO02, ... +- mit ID-Datei: jeder Eintrag findet den TRO, dessen Separator seiner Lage am + naechsten liegt (innerhalb der Toleranz, je TRO hoechstens ein Eintrag). + Nicht gefundene TROs bekommen die naechste freie Nummer, nicht gefundene + Eintraege bleiben in der Datei stehen und ihre ID wird nicht wiederverwendet. + +Das Modul liest und schreibt nur die Datei; Lage und Reihenfolge der TROs +liefert tro_flow.py. +""" + +from __future__ import annotations + +import configparser +import math +import re +from dataclasses import dataclass, field +from datetime import datetime +from pathlib import Path + +__all__ = [ + "SECTION", + "DEFAULT_TOLERANCE", + "IdEntry", + "IdAssignment", + "TroIds", + "ids_path", + "load_ids", + "assign_ids", + "render_ids", + "render_new_entries", + "insert_entries", + "other_sections", +] + +SECTION = "tro" +ID_FILE_SUFFIX = "_tro_ids.ini" +PREFIX = "TRO" + +# Abstand (mm) zwischen Eintrag und Separator, bis zu dem ein TRO als +# wiedergefunden gilt. Separatoren derselben Bahn liegen deutlich weiter +# auseinander; ein verschobenes Bauteil faellt so als "neu" auf. +DEFAULT_TOLERANCE = 100.0 + +# Eine ID wird zum SCL-Bezeichner (Instanz #, "DB_TRO_TEST".). +_ID_PATTERN = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") +# SYSn sind die Systemgrenzen (tro_flow.derive_boundaries), keine TROs. +_RESERVED = re.compile(r"^SYS\d+$") +_NUMBERED = re.compile(rf"^{PREFIX}(\d+)$") + + +@dataclass +class IdEntry: + """Ein Eintrag der ID-Datei: ID und Lage des Anker-Separators (mm).""" + + tro_id: str + x: float + y: float + + +@dataclass +class IdAssignment: + """Ergebnis von assign_ids, fuer den Bericht.""" + + renumbered: bool = False + # (ID, Abstand in mm) der wiedergefundenen TROs + kept: list[tuple[str, float]] = field(default_factory=list) + # IDs, die in diesem Lauf neu vergeben wurden (ohne renumbered) + new: list[str] = field(default_factory=list) + # Eintraege der Datei ohne passenden TRO + missing: list[IdEntry] = field(default_factory=list) + # weggefallene ID -> TRO, dessen Separator dort liegt, der aber schon eine + # naehere ID traegt (z. B. zwei TROs zu einem 2Sep2Swi zusammengelegt) + merged_into: dict[str, str] = field(default_factory=dict) + + +@dataclass +class TroIds: + """ID-Datei einer Anlage plus Einstellungen fuer einen Lauf. + + entries ist leer, wenn die Datei (noch) nicht existiert. result fuellt + tro_flow.derive_tros ueber assign_ids. + """ + + path: Path + entries: list[IdEntry] = field(default_factory=list) + exists: bool = False + tolerance: float = DEFAULT_TOLERANCE + renumber: bool = False + result: IdAssignment | None = None + + @classmethod + def load(cls, path: Path, *, tolerance: float = DEFAULT_TOLERANCE, + renumber: bool = False) -> "TroIds": + exists = path.is_file() + entries = load_ids(path) if exists else [] + return cls(path=path, entries=entries, exists=exists, + tolerance=tolerance, renumber=renumber) + + +def ids_path(csv_stem: str, cfg_dir: Path) -> Path: + """Standardpfad der ID-Datei zu einer CSV: /_tro_ids.ini.""" + return cfg_dir / f"{csv_stem}{ID_FILE_SUFFIX}" + + +def _parser() -> configparser.ConfigParser: + parser = configparser.ConfigParser( + inline_comment_prefixes=(";", "#"), + interpolation=None, + strict=True, + ) + parser.optionxform = str # IDs nicht in Kleinbuchstaben wandeln + return parser + + +def load_ids(path: Path) -> list[IdEntry]: + """ID-Datei lesen. ValueError bei doppelter ID, falscher Lage oder ID.""" + parser = _parser() + try: + parser.read(path, encoding="utf-8") + except configparser.Error as exc: + raise ValueError(f"{path}: {exc}") from exc + if not parser.has_section(SECTION): + raise ValueError(f"{path}: Abschnitt [{SECTION}] fehlt") + + entries: list[IdEntry] = [] + for tro_id, value in parser.items(SECTION): + if not _ID_PATTERN.match(tro_id): + raise ValueError(f"{path}: '{tro_id}' ist kein gueltiger SCL-Bezeichner") + if _RESERVED.match(tro_id): + raise ValueError(f"{path}: '{tro_id}' ist fuer Systemgrenzen reserviert") + parts = [p.strip() for p in value.split(",")] + try: + x, y = (float(p) for p in parts) + except ValueError: + raise ValueError(f"{path}: {tro_id} = '{value}' - erwartet 'x, y' in mm") \ + from None + entries.append(IdEntry(tro_id, x, y)) + return entries + + +def _distance(tro, entry: IdEntry, anchors: dict[str, tuple[float, float]]) -> float: + """Kleinster Abstand eines Separators des TRO zum Eintrag (inf ohne Lage).""" + best = math.inf + for sep in tro.separators: + pos = anchors.get(sep) + if pos is not None: + best = min(best, math.hypot(pos[0] - entry.x, pos[1] - entry.y)) + return best + + +def _width(count: int) -> int: + return max(2, len(str(count))) + + +def assign_ids( + tros: list, + anchors: dict[str, tuple[float, float]], + entries: list[IdEntry], + *, + tolerance: float = DEFAULT_TOLERANCE, + renumber: bool = False, +) -> IdAssignment: + """tro.tro_id fuer alle TROs setzen (Reihenfolge = Reihenfolge der Liste). + + anchors: Separator-TeileId -> (x, y) in mm. + """ + result = IdAssignment(renumbered=renumber or not entries) + if result.renumbered: + width = _width(len(tros)) + for index, tro in enumerate(tros, start=1): + tro.tro_id = f"{PREFIX}{index:0{width}d}" + return result + + # Paare nach Abstand; jeder TRO und jeder Eintrag hoechstens einmal + pairs = sorted( + (d, i, j) + for i, tro in enumerate(tros) + for j, entry in enumerate(entries) + if (d := _distance(tro, entry, anchors)) <= tolerance + ) + tro_taken: dict[int, tuple[int, float]] = {} + entry_taken: set[int] = set() + for d, i, j in pairs: + if i in tro_taken or j in entry_taken: + continue + tro_taken[i] = (j, d) + entry_taken.add(j) + + used = {e.tro_id for e in entries} + numbers = [int(m.group(1)) for e in entries if (m := _NUMBERED.match(e.tro_id))] + widths = [len(m.group(1)) for e in entries if (m := _NUMBERED.match(e.tro_id))] + width = max(widths + [_width(len(tros))]) + next_number = max(numbers, default=0) + 1 + + for i, tro in enumerate(tros): + if i in tro_taken: + j, d = tro_taken[i] + tro.tro_id = entries[j].tro_id + result.kept.append((tro.tro_id, d)) + continue + while f"{PREFIX}{next_number:0{width}d}" in used: + next_number += 1 + tro.tro_id = f"{PREFIX}{next_number:0{width}d}" + used.add(tro.tro_id) + next_number += 1 + result.new.append(tro.tro_id) + + result.missing = [e for j, e in enumerate(entries) if j not in entry_taken] + for d, i, j in pairs: + if j not in entry_taken: + result.merged_into.setdefault(entries[j].tro_id, tros[i].tro_id) + return result + + +def _anchor(tro, anchors: dict[str, tuple[float, float]]) -> tuple[float, float] | None: + for sep in tro.separators: + if sep in anchors: + return anchors[sep] + return None + + +def _entry_lines(tros: list, anchors: dict[str, tuple[float, float]]) -> list[str]: + lines = [] + for tro in tros: + pos = _anchor(tro, anchors) + if pos is None: + continue + seps = ", ".join(tro.separators) + lines.append(f"{tro.tro_id} = {pos[0]:.1f}, {pos[1]:.1f}" + f" ; {tro.type_name} | {tro.label} | Sep {seps}") + return lines + + +def render_ids(tros: list, anchors: dict[str, tuple[float, float]], source: str) -> str: + """Komplette ID-Datei (erster Lauf oder renumber).""" + head = [ + "; ============================================================", + f"; TRO-IDs zu {source} - erzeugt von lib/tro_flow.py " + f"am {datetime.now():%Y-%m-%d}", + ";", + "; Je TRO: ID = x, y (mm) seines Anker-Separators im CSV-Koordinatensystem.", + "; Die IDs duerfen umbenannt werden (eindeutig, gueltiger SCL-Bezeichner);", + "; tro_flow findet jeden TRO ueber die Lage wieder. Neue TROs werden", + "; unten angehaengt, weggefallene bleiben stehen und werden gemeldet.", + "; Alles neu nummerieren: tro_flow --renumber (ueberschreibt diese Datei).", + "; Siehe doc/Konzept_Layout_Hardware_SCL.md, Abschnitt TRO-IDs.", + "; ============================================================", + "", + f"[{SECTION}]", + ] + return "\n".join(head + _entry_lines(tros, anchors)) + "\n" + + +def render_new_entries(tros: list, anchors: dict[str, tuple[float, float]], + new_ids: list[str], source: str) -> str: + """Anhang fuer neu gefundene TROs an eine bestehende ID-Datei.""" + neu = [t for t in tros if t.tro_id in set(new_ids)] + head = [f"; neu aus {source} am {datetime.now():%Y-%m-%d}"] + return "\n".join(head + _entry_lines(neu, anchors)) + "\n" + + +def other_sections(path: Path) -> str: + """Alle Abschnitte ausser [tro] als Text (bleiben bei renumber erhalten).""" + if not path.is_file(): + return "" + header = re.compile(r"^\s*\[(.+)\]\s*$") + keep: list[str] = [] + current = None + for line in path.read_text(encoding="utf-8").splitlines(): + if m := header.match(line): + current = m.group(1).strip() + if current is not None and current != SECTION: + keep.append(line) + return "\n".join(keep).strip("\n") + + +def insert_entries(path: Path, block: str) -> None: + """block am Ende des Abschnitts [tro] einfuegen, Rest der Datei unveraendert.""" + lines = path.read_text(encoding="utf-8").splitlines() + header = re.compile(r"^\s*\[(.+)\]\s*$") + start = next(i for i, line in enumerate(lines) + if (m := header.match(line)) and m.group(1).strip() == SECTION) + end = next((i for i in range(start + 1, len(lines)) if header.match(lines[i])), + len(lines)) + # vor Leerzeilen am Abschnittsende einfuegen + while end > start + 1 and not lines[end - 1].strip(): + end -= 1 + lines[end:end] = block.rstrip("\n").splitlines() + path.write_text("\n".join(lines) + "\n", encoding="utf-8") diff --git a/tests/test_tro_ids.py b/tests/test_tro_ids.py new file mode 100644 index 0000000..0d0e2f5 --- /dev/null +++ b/tests/test_tro_ids.py @@ -0,0 +1,126 @@ +# -*- coding: utf-8 -*- +"""Tests fuer lib/tro_ids.py: ID-Datei lesen/schreiben, TROs ueber die Lage +wiederfinden (doc/Konzept_Layout_Hardware_SCL.md, Abschnitt TRO-IDs).""" + +from __future__ import annotations + +from dataclasses import dataclass, field + +import pytest + +from tro_ids import ( + IdEntry, + TroIds, + assign_ids, + insert_entries, + load_ids, + other_sections, + render_ids, + render_new_entries, +) + + +@dataclass +class FakeTro: + separators: list[str] + type_name: str = "1Sep" + label: str = "" + tro_id: str = "" + hosts: list[str] = field(default_factory=list) + + +ANCHORS = { + "0001": (0.0, 0.0), + "0002": (1000.0, 0.0), + "0003": (2000.0, 0.0), + "0004": (2000.0, 500.0), +} + + +def _tros(*seps: list[str]) -> list[FakeTro]: + return [FakeTro(separators=list(s)) for s in seps] + + +def test_ohne_eintraege_wird_durchnummeriert(): + tros = _tros(["0001"], ["0002"], ["0003"]) + result = assign_ids(tros, ANCHORS, []) + assert result.renumbered + assert [t.tro_id for t in tros] == ["TRO01", "TRO02", "TRO03"] + + +def test_wiederfinden_ueber_lage_auch_umbenannt(): + # Neuer Export: andere TeileIds, gleiche Lage, andere Reihenfolge + anchors = {"0101": (2000.0, 0.0), "0102": (0.0, 0.0), "0103": (1000.0, 30.0)} + tros = _tros(["0101"], ["0102"], ["0103"]) + entries = [IdEntry("TRO01", 0.0, 0.0), IdEntry("K0102_1", 1000.0, 0.0), + IdEntry("TRO03", 2000.0, 0.0)] + result = assign_ids(tros, anchors, entries) + assert [t.tro_id for t in tros] == ["TRO03", "TRO01", "K0102_1"] + assert not result.new and not result.missing + + +def test_neue_tros_naechste_freie_nummer_weggefallene_nicht_wiederverwendet(): + tros = _tros(["0001"], ["0003"]) + entries = [IdEntry("TRO01", 0.0, 0.0), IdEntry("TRO02", 1000.0, 0.0)] + result = assign_ids(tros, ANCHORS, entries) + assert [t.tro_id for t in tros] == ["TRO01", "TRO03"] + assert result.new == ["TRO03"] + assert [e.tro_id for e in result.missing] == ["TRO02"] + + +def test_ausserhalb_toleranz_gilt_als_neu(): + tros = _tros(["0001"]) + entries = [IdEntry("TRO01", 150.0, 0.0)] + result = assign_ids(tros, ANCHORS, entries, tolerance=100.0) + assert tros[0].tro_id == "TRO02" + assert result.missing[0].tro_id == "TRO01" + assert assign_ids(tros, ANCHORS, entries, tolerance=200.0).kept[0][0] == "TRO01" + + +def test_zusammengelegte_tros_naechster_eintrag_gewinnt(): + # zwei Separatoren, frueher zwei TROs, jetzt ein 2Sep-TRO + tros = _tros(["0003", "0004"]) + entries = [IdEntry("TRO05", 2000.0, 0.0), IdEntry("TRO06", 2000.0, 510.0)] + result = assign_ids(tros, ANCHORS, entries) + assert tros[0].tro_id == "TRO05" + assert result.merged_into == {"TRO06": "TRO05"} + + +def test_renumber_ignoriert_eintraege(): + tros = _tros(["0001"], ["0002"]) + result = assign_ids(tros, ANCHORS, [IdEntry("X", 1000.0, 0.0)], renumber=True) + assert result.renumbered + assert [t.tro_id for t in tros] == ["TRO01", "TRO02"] + + +def test_datei_rundlauf_und_anhaengen(tmp_path): + path = tmp_path / "anlage_tro_ids.ini" + tros = _tros(["0001"], ["0002"]) + assign_ids(tros, ANCHORS, []) + path.write_text(render_ids(tros, ANCHORS, "anlage.csv") + + "\n[separator]\n; spaeter\n", encoding="utf-8") + entries = load_ids(path) + assert [(e.tro_id, e.x, e.y) for e in entries] == [ + ("TRO01", 0.0, 0.0), ("TRO02", 1000.0, 0.0)] + + tros = _tros(["0001"], ["0002"], ["0003"]) + result = assign_ids(tros, ANCHORS, entries) + insert_entries(path, render_new_entries(tros, ANCHORS, result.new, "anlage.csv")) + assert [e.tro_id for e in load_ids(path)] == ["TRO01", "TRO02", "TRO03"] + # neuer Eintrag steht in [tro], nicht im folgenden Abschnitt + assert other_sections(path) == "[separator]\n; spaeter" + assert TroIds.load(path).exists + + +@pytest.mark.parametrize("inhalt, meldung", [ + ("[tro]\nTRO01 = 1, 2\nTRO01 = 3, 4\n", "already exists"), + ("[tro]\nSYS1 = 1, 2\n", "reserviert"), + ("[tro]\nTRO-1 = 1, 2\n", "SCL-Bezeichner"), + ("[tro]\nTRO01 = 1\n", "erwartet"), + ("[andere]\n", "fehlt"), +]) +def test_fehlerhafte_datei(tmp_path, inhalt, meldung): + path = tmp_path / "x.ini" + path.write_text(inhalt, encoding="utf-8") + with pytest.raises(ValueError, match=meldung): + load_ids(path)