From 1ac17b95e263a63bcfad29d366c6ff3a300ef49e Mon Sep 17 00:00:00 2001 From: Samer Ayadi Date: Tue, 27 Oct 2026 14:32:10 +0100 Subject: [PATCH] validate_anlage, stabiler TRO-Schluessel, Verbindungen aus BricsCAD, Zeichnung aus der Anlagen-JSON - validate_anlage.py (+ bin-Wrapper): Pruefung vor der SCL fuer beide Wege (Struktur, Zuordnungen, Aktualitaet der JSON, Korrekturen ohne Ziel); scl_skeleton ruft sie auf (Warnung, --strict bricht ab) - DerivedTro.key: stabiler Schluessel (ID der ID-Datei); hw_overrides.ini, _jam_names.ini und tro_overrides.ini greifen ueber ID oder Schluessel, eine Umbenennung in TRO_EDIT laesst sie bestehen; JAM_AREA traegt KEY - --read-tros liest die CONNECTION_ARROW-Bloecke mit -> [.edit_connections]; tro_flow.apply_connection_edits wendet sie nach link_tros an - tro_annotate --anlage zeichnet Marker/Verbindungen aus der Anlagen-JSON - INI-Korrekturen auch mit BOM lesbar (utf-8-sig) - Ablauf CSV/DXF+TIA -> Anlagen-JSON -> (BricsCAD) -> SCL in doc/Python_Scripts.md Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 14 +- bin/validate_anlage.bat | 30 +++ bin/validate_anlage.sh | 31 +++ doc/Python_Scripts.md | 93 ++++++++- lib/merge_hw_layout.py | 8 +- lib/model.py | 24 ++- lib/scl_context.py | 28 ++- lib/scl_skeleton.py | 17 ++ lib/tro_annotate.py | 197 +++++++++++++++--- lib/tro_flow.py | 51 ++++- lib/tro_overrides.py | 59 +++++- lib/validate_anlage.py | 364 ++++++++++++++++++++++++++++++++++ tests/test_anlage_workflow.py | 232 ++++++++++++++++++++++ 13 files changed, 1100 insertions(+), 48 deletions(-) create mode 100644 bin/validate_anlage.bat create mode 100644 bin/validate_anlage.sh create mode 100644 lib/validate_anlage.py create mode 100644 tests/test_anlage_workflow.py diff --git a/CLAUDE.md b/CLAUDE.md index c9097ea..bd68427 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -212,7 +212,9 @@ byte — rewrite them (`SKEL_UPDATE_GOLDEN=1`) only for a reviewed, intended cha types, rows from `*TRO-SLOTS*`, pick-in-drawing from the XRECORD `SPS_SKEL_PARTS`) edits them, and `--read-tros` writes the differences as `edit`/`edit_add` sections into `cfg/tro_overrides.ini` — the only kinds of that file `tro_flow` applies - (`tro_flow.apply_edits`, keyed on the stable ID-file IDs). + (`tro_flow.apply_edits`, keyed on the stable ID-file IDs); since 2026-10-27 also the + connection blocks (`edit_connections`, `tro_flow.apply_connection_edits`). With + `--anlage` the markers/connections are drawn from the plant JSON itself. 4. **`lib/tro_extract.py`** — **parked**: reads the annotated DXF back into the old `sps_skel-tro-extract-v1` JSON, which `scl_skeleton.py` no longer accepts. Not deleted, not extended; if drawing edits (`TRO_INSERT`/`TRO_EDIT`) are ever needed again, it should @@ -248,6 +250,16 @@ byte — rewrite them (`SKEL_UPDATE_GOLDEN=1`) only for a reviewed, intended cha open/contradicting bindings stay `TODO(E-Planung)`; inserted names are checked against `_hw.json` (`--strict` aborts). Templates are unchanged. See `doc/Python_Scripts.md` §5b. +9. **`lib/validate_anlage.py`** (implemented 2026-10-27) — the gate before SCL for both paths + (no changes / after BricsCAD edits): structure, bindings, freshness (plant JSON older than + TRO-JSON/hw.json/hw_overrides → rerun merge), corrections that match nothing. Exit 5 = not + ready; `scl_skeleton` calls it (warns, `--strict` aborts). The ordered command list for + both paths is in `doc/Python_Scripts.md` §1 "The workflow, step by step". + +Since 2026-10-27 every TRO carries a stable `key` (ID-file ID) next to its display `id`; +all corrections (`hw_overrides.ini`, `_jam_names.ini`, `tro_overrides.ini`) resolve by +ID **or** key, so a rename in `TRO_EDIT` keeps them. The plant JSON is never hand-patched — +it is regenerated from inputs + these overlays. What is **not** built out yet, relative to the original two-tool design: diff --git a/bin/validate_anlage.bat b/bin/validate_anlage.bat new file mode 100644 index 0000000..b262b90 --- /dev/null +++ b/bin/validate_anlage.bat @@ -0,0 +1,30 @@ +@echo off +REM ================================================================ +REM SPS_SKEL - Anlagen-JSON pruefen, bevor daraus SCL wird +REM ================================================================ +REM Aufruf: +REM bin\validate_anlage.bat --json mubea_anlage.json +REM bin\validate_anlage.bat --json dematic_merge_anlage.json --strict +REM +REM Die JSON wird in %SKEL_RESULTS% gesucht. Exit-Code 0 = bereit, +REM 5 = nicht bereit (Fehler; mit --strict auch Warnungen). +REM ================================================================ + +setlocal + +REM Umgebung setzen (SKEL_DATA, SKEL_RESULTS, SKEL_LIB, PYTHONPATH ...) +call "%~dp0setenv.bat" >nul + +REM Virtuelle Umgebung nutzen, falls vorhanden (pydantic wird benoetigt) +if exist "%SPS_SKEL%\.venv\Scripts\activate.bat" ( + call "%SPS_SKEL%\.venv\Scripts\activate.bat" +) + +REM Python-Interpreter bestimmen (py-Launcher bevorzugt, sonst python) +set "PY=py" +py --version >nul 2>&1 || set "PY=python" + +"%PY%" "%SKEL_LIB%\validate_anlage.py" %* +set "RC=%ERRORLEVEL%" + +endlocal & exit /b %RC% diff --git a/bin/validate_anlage.sh b/bin/validate_anlage.sh new file mode 100644 index 0000000..e39c7d0 --- /dev/null +++ b/bin/validate_anlage.sh @@ -0,0 +1,31 @@ +#!/usr/bin/env bash +# ================================================================ +# SPS_SKEL - Anlagen-JSON pruefen, bevor daraus SCL wird +# ================================================================ +# Aufruf: +# bash bin/validate_anlage.sh --json mubea_anlage.json +# bash bin/validate_anlage.sh --json dematic_merge_anlage.json --strict +# +# Die JSON wird in $SKEL_RESULTS gesucht. Exit-Code 0 = bereit, +# 5 = nicht bereit (Fehler; mit --strict auch Warnungen). +# ================================================================ + +set -eu + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" + +# Umgebung setzen (SKEL_DATA, SKEL_RESULTS, SKEL_LIB, PYTHONPATH ...) +# shellcheck source=/dev/null +. "$SCRIPT_DIR/setenv.sh" >/dev/null + +# Virtuelle Umgebung nutzen, falls vorhanden (pydantic wird benoetigt) +if [ -f "$SPS_SKEL/.venv/bin/activate" ]; then + # shellcheck source=/dev/null + . "$SPS_SKEL/.venv/bin/activate" +fi + +# Python-Interpreter bestimmen +PY="python3" +command -v python3 >/dev/null 2>&1 || PY="python" + +exec "$PY" "$SKEL_LIB/validate_anlage.py" "$@" diff --git a/doc/Python_Scripts.md b/doc/Python_Scripts.md index 421a726..25c2927 100644 --- a/doc/Python_Scripts.md +++ b/doc/Python_Scripts.md @@ -63,7 +63,40 @@ parked: tro_extract.py drawing ──► JSON (not extended; see §5a) The drawing is a **view** for checking the result, not a data source: the TRO-JSON is written by `tro_flow.py --json` straight from the CSV, and -`tro_annotate.py` derives exactly the same TROs to draw them. +`tro_annotate.py` derives exactly the same TROs to draw them — with +`--anlage` it draws the plant JSON itself (since 2026-10-27). + +### The workflow, step by step (since 2026-10-27) + +The plant JSON `_anlage.json` is the **unified JSON**: layout (TROs, +connections, jam areas) + hardware bindings. It is always **regenerated**; +corrections made by people live in INI overlays in `%SKEL_CFG%` that are +keyed on stable IDs and survive a new CSV export: + +| Overlay | Written by | Applied by | Keyed on | +|---|---|---|---| +| `tro_overrides.ini` `edit`/`edit_add`/`edit_connections` | `tro_annotate --read-tros` | `tro_flow` (and `tro_annotate`) | TRO key (`DerivedTro.key`) | +| `_jam_names.ini` | `tro_annotate --read-jams` | `scl_skeleton`, `tro_annotate` | `.` | +| `hw_overrides.ini` | by hand | `merge_hw_layout` | TRO ID **or** key | + +``` +1 tro_flow --file --json CSV -> _tro.json +2 hw_import --tia _TIA.xlsx --positions|--dxf E-plan + TIA -> _hw.json +3 merge_hw_layout --tro _tro.json --hw _hw.json -> _anlage.json +4 validate_anlage --json _anlage.json ready for SCL? (§5f) + +Path A (no changes): scl_skeleton --json _anlage.json [--templates] [--strict] +Path B (changes): tro_annotate --file --dxf .dxf --flow --anlage _anlage.json + ... edit in BricsCAD (TRO_EDIT, TRO_INSERT, JAM_EDIT, + CONNECTION_INSERT/_EDIT, delete markers/arrow blocks), save as DXF + tro_annotate --file --read-tros -> tro_overrides.ini + tro_annotate --file --read-jams -> _jam_names.ini + then steps 1, 3, 4 again, then Path A +``` + +`scl_skeleton` runs the step-4 check itself (warns; `--strict` aborts on +errors), so an outdated plant JSON — e.g. step 3 forgotten after step 1 — is +reported, not silently turned into SCL. - `material_flow.py` models the material flow of the **mechanical** objects only (Gefällestrecke, Strecke, Kreisel). @@ -467,7 +500,36 @@ falls back to the nearest part of the same kind at that spot. The sections replace all earlier `edit`/`edit_add` sections of the CSV in `tro_overrides.ini` (a `reason =` added by hand is kept); the rest of the file stays line for line. The next `tro_flow` / `tro_annotate` run applies them -(§4); a re-read of the new drawing gives the same sections. `--check` only +(§4); a re-read of the new drawing gives the same sections. + +**Connections (since 2026-10-27).** `--read-tros` also reads every +`CONNECTION_ARROW` block. Each end is matched to a marker of the same drawing +by its visible name (`FROM_TRO`/`TO_TRO`), else by the internal number +(`FROM_UID`/`TO_UID`; older dxfmakros blocks `FROM_ID`/`TO_ID`) — so arrows +still point right after a marker was renamed. The arrows are compared with the +derivation *including* the TRO corrections (edges of a new or edited TRO do not +count as hand-drawn); the differences go to `[.edit_connections]` as +`+ KEY_A -> KEY_B | kind` / `- KEY_A -> KEY_B`, ends as stable keys. +`tro_flow.apply_connection_edits` applies them after `link_tros` and resets +predecessors/successors, so the jam areas follow. Station arrows carry no +block and stay as derived; an arrow is deleted by deleting its block. Before +this, `CONNECTION_INSERT`/`_EDIT` in dxfmakros wrote attribute names the +blocks did not have (`ID`/`FROM_ID`/`TO_ID`, marker name from `TRO_ID`) — fixed +there on the same day. + +**Stable key (since 2026-10-27).** Every TRO in the TRO-JSON carries `key`, +its ID in `_tro_ids.ini` (or, for a TRO added in BricsCAD, its own ID). +It differs from `id` only after a rename in `TRO_EDIT`. All corrections are +resolved by ID **or** key (`TroDocument.resolve`, `scl_context.jam_override`), +so a rename does not orphan `hw_overrides.ini` or `_jam_names.ini` +entries. Jam blocks carry a hidden `KEY` (`.`) that +`--read-jams` writes; an entry under the old display ID is moved to the key. + +**With `--anlage` the drawing shows the plant JSON** (since 2026-10-27): +markers, connections and stations come from `anlage.layout` +(`analysis_from_document`), i.e. exactly what becomes SCL. The CSV is still +read for the registration and the parts table; if its derivation differs from +the JSON, a warning says the JSON is probably outdated. `--check` only reports. If the drawing has no marker of the derived TROs (wrong file, or annotated before the `KEY`/`SLOTS` attributes), nothing is written — re-annotating an old drawing adds the missing attributes to the existing @@ -496,7 +558,7 @@ verified on every run, and cached per CSV/DXF filename pair in | `--emit-lisp` | — | off | Regenerate `cad/tro_types.lsp` for the `TRO_INSERT`/`TRO_EDIT` dialog — type list (`*TRO-TYPES*`), slots per type with the storage-line default (`*TRO-SLOTS*`), part kinds per slot (`*TRO-SLOT-KINDS*`) — from `tro_catalog.py`/`model.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`. | | `--anlage` | `NAME` | — | Plant JSON (`_anlage.json` from `merge_hw_layout`, in `%SKEL_RESULTS%` or a path): also draw the **jam areas** (see below). Must belong to the same CSV. | | `--read-jams` | `[DXF]` | `_annotated.dxf` | Read the jam-area numbers edited in BricsCAD back from the annotated drawing (saved as DXF) into `%SKEL_CFG%\_jam_names.ini`, then exit — no `--dxf` needed. | -| `--read-tros` | `[DXF]` | `_annotated.dxf` | Read the TRO markers edited in BricsCAD (`TRO_EDIT`/`TRO_INSERT`, deleted markers) back from the annotated drawing (saved as DXF) into the `edit`/`edit_add` sections of `%SKEL_CFG%\tro_overrides.ini`, then exit — no `--dxf` needed. With `--check` only reported. | +| `--read-tros` | `[DXF]` | `_annotated.dxf` | Read the TRO markers edited in BricsCAD (`TRO_EDIT`/`TRO_INSERT`, deleted markers) and the connection blocks (`CONNECTION_INSERT`/`_EDIT`, deleted blocks) back from the annotated drawing (saved as DXF) into the `edit`/`edit_add`/`edit_connections` sections of `%SKEL_CFG%\tro_overrides.ini`, then exit — no `--dxf` needed. With `--check` only reported. | **Jam areas (`--anlage`).** One block `JAM_AREA` per jam area of the plant JSON on layer `TRO_JAM`: an upright grey rectangle with the name written @@ -1007,6 +1069,31 @@ fingerprint of the stored registration, CLI end to end with a real offset, merged drawing without registration, two drive TROs on one conveyor, Dematic rules file). +## 5f. `validate_anlage.py` — is the plant JSON ready for SCL? + +One check for both paths of the workflow (§1): directly after +`merge_hw_layout` and after BricsCAD corrections have been read back and the +JSON rebuilt. `scl_skeleton` calls it before writing (findings as warnings; +with `--strict`, errors stop the run). Works on a plant JSON and, reduced, on +a TRO-JSON. + +| Area | ERROR (no SCL like this) | WARNING (SCL with open spots) | +|---|---|---| +| Structure | duplicate TRO-ID, type not in catalog, connection or jam area to an unknown TRO, separator in two TROs | separator slot without layout part, TRO without connection, TRO not reachable from the material entry | +| Bindings | binding to an unknown TRO/slot, status `widerspruch`, BMK missing in `_hw.json` | open mandatory roles (counted per role) | +| Freshness | plant JSON older than TRO-JSON, hardware planning or `hw_overrides.ini` (only if it has sections for this CSV) → run `merge_hw_layout` again | `tro_overrides.ini` with BricsCAD corrections newer than the TRO-JSON → run `tro_flow --json` again | +| Corrections | — | entries in `hw_overrides.ini` / `_jam_names.ini` that match no TRO/jam area, neither by ID nor by key | + +| Switch | Argument | Default | Meaning | +|---|---|---|---| +| `--json` | `NAME` | — | Plant JSON (or TRO-JSON) in `%SKEL_RESULTS%`, or a path. | +| `--hw` | `NAME` | `hwFile` of the plant JSON | Hardware planning for the BMK check. | +| `--strict` | — | off | Warnings also make the JSON "not ready". | + +**Exit codes:** `0` ready · `1` input error · `5` not ready. +Mubea on 2026-10-27: the plant JSON in `results/` was reported older than the +TRO-JSON — exactly the TRO18/TRO19 mismatch `tro_annotate --anlage` showed. + ## 6. `tro_catalog.py` — TRO type catalog (library, no CLI) The single source of truth for everything needed to generate or draw a TRO diff --git a/lib/merge_hw_layout.py b/lib/merge_hw_layout.py index fed64c4..d1e78c7 100644 --- a/lib/merge_hw_layout.py +++ b/lib/merge_hw_layout.py @@ -616,7 +616,8 @@ def load_overrides(path: Path, layout_source: str) -> Overrides: parser = configparser.ConfigParser(inline_comment_prefixes=(";", "#"), interpolation=None) parser.optionxform = str - with path.open(encoding="utf-8") as fh: + # utf-8-sig: von Hand (Editor, PowerShell) gespeichert oft mit BOM + with path.open(encoding="utf-8-sig") as fh: parser.read_file(fh) section = f"{layout_source}.bindings" if parser.has_section(section): @@ -639,7 +640,10 @@ def _apply_overrides(doc: TroDocument, bindings: list[Binding], hw: HardwareDocument, warnings: list[str]) -> list[Binding]: devices = hw.by_bmk() slots = {(t.tro_id, s.slot): s for t in doc.tros for s in t.slots} - for (tro, slot, role), bmk in overrides.bindings.items(): + for (name, slot, role), bmk in overrides.bindings.items(): + # ID oder - nach einer Umbenennung in BricsCAD - der stabile Schluessel + found = doc.resolve(name) + tro = found.tro_id if found is not None else name s = slots.get((tro, slot)) if s is None: warnings.append(f"{OVERRIDES_FILE}: {tro}.{slot} gibt es in der TRO-JSON nicht " diff --git a/lib/model.py b/lib/model.py index 4b1c3a2..c9b77af 100644 --- a/lib/model.py +++ b/lib/model.py @@ -332,6 +332,12 @@ class DerivedTro(TroBase): """ label: str + # Stabiler Schluessel: die ID aus _tro_ids.ini (bzw. die Kennung eines + # in BricsCAD angelegten TRO). Weicht von ``tro_id`` ab, wenn der TRO im + # Dialog TRO_EDIT umbenannt wurde. Handkorrekturen (hw_overrides.ini, + # _jam_names.ini, tro_overrides.ini) gelten ueber ID *oder* Schluessel, + # damit sie eine Umbenennung ueberleben. "" in Dateien vor 2026-10-27. + key: str = "" hosts: list[str] scanners: list[str] # Weichen-Bauteile (Ausschleuselement oder ILS Weiche), Reihenfolge wie die @@ -351,6 +357,11 @@ class DerivedTro(TroBase): def connected(self) -> bool: return bool(self.successors or self.predecessors or self.colocated) + @property + def stable_key(self) -> str: + """Schluessel fuer Handkorrekturen (``key``, sonst die ID).""" + return self.key or self.tro_id + class Connection(BaseModel): """Gerichtete Verbindung zwischen zwei TROs/Stationen (JSON: from/to/kind).""" @@ -451,7 +462,9 @@ class TroDocument(BaseModel): traegt und nicht erst das Rendern sie kennt. ``parts`` (Lage der Layout-Bauteile) gibt tro_flow aus dem CSV mit. """ - tros = [t.model_copy(update={"slots": derive_slots(t)}) for t in analysis.tros] + tros = [t.model_copy(update={"slots": derive_slots(t), + "key": analysis.edit_key(t.tro_id)}) + for t in analysis.tros] return cls( source=source, note=note, @@ -490,6 +503,15 @@ class TroDocument(BaseModel): return tro return None + def resolve(self, name: str) -> DerivedTro | None: + """TRO zu einer ID oder - nach einer Umbenennung - zu seinem Schluessel. + + Die ID geht vor: ein neuer TRO darf die alte Kennung eines umbenannten + tragen, dann meint ein Eintrag mit dieser Kennung den neuen. + """ + return self.by_id(name) or next( + (t for t in self.tros if t.key and t.key == name), None) + # --------------------------------------------------------------------------- # Anlagen-JSON: Layout + Zuordnungen der Hardwareplanung (merge_hw_layout) diff --git a/lib/scl_context.py b/lib/scl_context.py index 00fbabc..09beb18 100644 --- a/lib/scl_context.py +++ b/lib/scl_context.py @@ -68,6 +68,8 @@ __all__ = [ "flow_order", "JAM_NAMES_SUFFIX", "jam_key", + "jam_override", + "jam_stable_key", "jam_label", "jam_names", "jam_numbers", @@ -653,6 +655,25 @@ def jam_key(area: JamArea) -> str: return f"{area.tro}.{area.entry}" +def jam_stable_key(area: JamArea, document: TroDocument) -> str: + """Wie jam_key, aber mit dem stabilen Schluessel des TRO (DerivedTro.key): + bleibt gleich, wenn der TRO in BricsCAD umbenannt wird.""" + tro = document.by_id(area.tro) + return f"{tro.stable_key}.{area.entry}" if tro is not None else jam_key(area) + + +def jam_override(area: JamArea, document: TroDocument, + overrides: dict[str, tuple[str, int]] | None) -> tuple[str, int] | None: + """Korrektur eines Staubereichs: ueber den stabilen Schluessel, sonst ueber + die aktuelle TRO-ID (Dateien vor 2026-10-27 bzw. ohne Umbenennung).""" + if not overrides: + return None + for key in (jam_stable_key(area, document), jam_key(area)): + if key in overrides: + return overrides[key] + return None + + def jam_label(nr: str, k: int) -> str: """Kurzname fuer Zeichnung und Dialog: Jam0054_3.""" return f"Jam{nr}_{k}" @@ -671,7 +692,8 @@ def load_jam_overrides(path: Path) -> dict[str, tuple[str, int]]: parser = configparser.ConfigParser(inline_comment_prefixes=(";", "#"), interpolation=None) parser.optionxform = str - with path.open(encoding="utf-8") as fh: + # utf-8-sig: von Hand (Editor, PowerShell) gespeichert oft mit BOM + with path.open(encoding="utf-8-sig") as fh: parser.read_file(fh) result: dict[str, tuple[str, int]] = {} if parser.has_section("jam"): @@ -711,8 +733,8 @@ def jam_names(document: TroDocument, geordnet: list[TroBase], hw: HwSource, """ nummern = jam_numbers(document, geordnet, hw) for area in document.jam_areas_or_derived(): - if overrides and jam_key(area) in overrides: - nummern[area.jam_id] = overrides[jam_key(area)] + if (wert := jam_override(area, document, overrides)) is not None: + nummern[area.jam_id] = wert return {jam_id: f'"DB_JamArea{nr}".stJam{nr}_{k}' for jam_id, (nr, k) in nummern.items()} diff --git a/lib/scl_skeleton.py b/lib/scl_skeleton.py index 7c63f44..57d7ca9 100644 --- a/lib/scl_skeleton.py +++ b/lib/scl_skeleton.py @@ -59,6 +59,7 @@ from pydantic import BaseModel, ValidationError from e_planning import EPLAN_SCHEMA, EplanDocument from hw_import import HardwareDocument +from validate_anlage import validate from material_flow import env_dir from model import ANLAGE_SCHEMA, SCHEMA, AnlageDocument, TroDocument from scl_context import ( @@ -441,6 +442,20 @@ def main(argv: list[str] | None = None) -> int: print(f"FEHLER: {json_file} enthaelt keine TROs.", file=sys.stderr) return 1 + # Pruefung vor der SCL (validate_anlage): Fehler brechen nur mit --strict + # ab, sonst stehen sie als Warnung im Bericht. + pruefung = validate(json_file, hw_file=_in_results(args.hw, results) if args.hw else None, + results=json_file.parent) + for befund in pruefung.befunde: + if befund.stufe == "FEHLER" or args.strict: + print(f"{'FEHLER' if args.strict else 'WARNUNG'}: {befund.bereich}: {befund.text}", + file=sys.stderr) + if args.strict and not pruefung.bereit(): + print(f" (--strict) {pruefung.anzahl_fehler} Fehler in der Pruefung - keine " + f"SCL geschrieben (Details: bin/validate_anlage --json {json_file.name}).", + file=sys.stderr) + return 1 + templates = None if args.templates is not None: ordner = Path(args.templates) @@ -550,6 +565,8 @@ def main(argv: list[str] | None = None) -> int: print(f"Gegenpruefung = {len(unbekannt)} unbekannte Namen ({hw_file.name})") if jam_overrides: print(f"Staubereiche = {len(jam_overrides)} Korrekturen aus {jam_file}") + print(f"Pruefung = {pruefung.anzahl_fehler} Fehler, {pruefung.anzahl_warnungen} " + f"Warnungen (validate_anlage)") print(f"Offene Stellen = {offen} (Marker '{TODO}')") print(f"SCL = {out_file}") print("================================================================") diff --git a/lib/tro_annotate.py b/lib/tro_annotate.py index d06d7ac..d0d8f56 100644 --- a/lib/tro_annotate.py +++ b/lib/tro_annotate.py @@ -143,20 +143,29 @@ from model import ( AnlageDocument, DerivedTro, JamArea, + TroDocument, derive_slots, has_lines, slot_kind, slot_names, ) from tro_ids import id_problem -from tro_overrides import PartRef, TroEdit, load_overrides, override_path, save_edits +from tro_overrides import ( + ConnectionEdit, + PartRef, + TroEdit, + load_overrides, + override_path, + save_edits, +) from scl_context import ( JAM_NAMES_SUFFIX, HwSource, flow_order, - jam_key, jam_label, jam_numbers, + jam_override, + jam_stable_key, load_jam_overrides, save_jam_overrides, ) @@ -180,6 +189,8 @@ JAM_BLOCK = "JAM_AREA" PARTS_XRECORD = "SPS_SKEL_PARTS" # Unsichtbare Attribute der Marker (build_symbol) HIDDEN_TAGS = ("UID", "KEY", "ITEMS", "CONFIDENCE", "SLOTS") +# Unsichtbare Attribute der Staubereiche (build_jam_symbol) +JAM_HIDDEN_TAGS = ("CONVEYOR", "NUMBER", "TRO", "ENTRY", "AREA", "AUTO", "KEY") # Anzeige eines Staubereichs ohne Foerderer (Nummer nicht ableitbar) JAM_OFFEN = "Jam????_?" @@ -854,35 +865,49 @@ def build_jam_symbol(doc, text: float) -> str: AREA unsichtbar - Kennung der TRO-JSON (J0039) AUTO unsichtbar - abgeleitete Nummer NNNN_k; --read-jams schreibt nur, was davon abweicht - TRO + ENTRY ist der stabile Schluessel (scl_context.jam_key). + KEY unsichtbar - stabiler Schluessel . + (scl_context.jam_stable_key), ueberlebt eine Umbenennung + KEY (sonst TRO + ENTRY) ist der Schluessel in _jam_names.ini. """ + w, h = text * JAM_HALB_BREITE, text * JAM_HALB_HOEHE if JAM_BLOCK in doc.blocks: # Form aus einem frueheren Lauf (liegendes Rechteck) ersetzen; die - # Attributdefinitionen bleiben, damit vorhandene Bloecke gueltig sind. + # Attributdefinitionen bleiben, damit vorhandene Bloecke gueltig sind, + # fehlende (KEY) kommen dazu. blk = doc.blocks.get(JAM_BLOCK) for e in list(blk): if e.dxftype() != "ATTDEF": blk.delete_entity(e) _jam_form(blk, text) + vorhanden = set() for a in blk.query("ATTDEF"): + vorhanden.add(a.dxf.tag.upper()) if a.dxf.tag.upper() == "NAME": _jam_name(a, text) + offset = min((a.dxf.insert.y for a in blk.query("ATTDEF")), default=-h) + for tag_name in JAM_HIDDEN_TAGS: + if tag_name not in vorhanden: + offset -= text + _jam_attdef(blk, tag_name, -w, offset, text) return JAM_BLOCK blk = doc.blocks.new(name=JAM_BLOCK) _jam_form(blk, text) - w, h = text * JAM_HALB_BREITE, text * JAM_HALB_HOEHE name = blk.add_attdef("NAME", insert=(0, 0), height=text * JAM_TEXT, dxfattribs={"style": "Standard", "text": JAM_OFFEN}) _jam_name(name, text) offset = -h - text - for tag_name in ("CONVEYOR", "NUMBER", "TRO", "ENTRY", "AREA", "AUTO"): - blk.add_attdef(tag_name, insert=(-w, offset), height=text * 0.7, - dxfattribs={"style": "Standard", "color": 256, "text": "", - "invisible": 1}) + for tag_name in JAM_HIDDEN_TAGS: + _jam_attdef(blk, tag_name, -w, offset, text) offset -= text return JAM_BLOCK +def _jam_attdef(blk, tag_name: str, x: float, y: float, text: float) -> None: + blk.add_attdef(tag_name, insert=(x, y), height=text * 0.7, + dxfattribs={"style": "Standard", "color": 256, "text": "", + "invisible": 1}) + + def _jam_form(blk, text: float) -> None: """Stehendes Rechteck: graue Fuellung, grauer Rand.""" w, h = text * JAM_HALB_BREITE, text * JAM_HALB_HOEHE @@ -967,7 +992,7 @@ def place_jams( f"nicht gezeichnet") continue auto = automatisch.get(area.jam_id) - nr, k = overrides.get(jam_key(area), auto or ("", 0)) + nr, k = jam_override(area, layout, overrides) or auto or ("", 0) ref = msp.add_blockref(block, transform.to_dxf(*lage), dxfattribs={"layer": layers["jam"]}) ref.add_auto_attribs({ @@ -978,6 +1003,8 @@ def place_jams( "ENTRY": str(area.entry), "AREA": area.jam_id, "AUTO": f"{auto[0]}_{auto[1]}" if auto else "", + # Schluessel fuer --read-jams: ueberlebt eine Umbenennung des TRO + "KEY": jam_stable_key(area, layout), }) tag(ref, area.tro) gezeichnet += 1 @@ -999,11 +1026,16 @@ def read_jams(dxf_file: Path, ini_file: Path) -> tuple[dict[str, tuple[str, int] namen: dict[str, str] = {} for ref in doc.modelspace().query(f'INSERT[name=="{JAM_BLOCK}"]'): a = {att.dxf.tag.upper(): (att.dxf.text or "").strip() for att in ref.attribs} - key = f"{a.get('TRO', '')}.{a.get('ENTRY', '')}" + # Stabiler Schluessel (KEY, seit 2026-10-27), sonst TRO-ID + Eingang + anzeige = f"{a.get('TRO', '')}.{a.get('ENTRY', '')}" + key = a.get("KEY") or anzeige nr, k = a.get("CONVEYOR", ""), a.get("NUMBER", "") if not a.get("TRO") or not a.get("ENTRY"): meldungen.append(f"Staubereich ohne TRO/ENTRY ({a.get('NAME', '?')}) - uebersprungen") continue + if key != anzeige and anzeige in overrides: + # Eintrag unter der alten Anzeige-ID auf den Schluessel umstellen + overrides.setdefault(key, overrides.pop(anzeige)) if not nr and not k: continue if not (nr.isdigit() and len(nr) == 4 and k.isdigit() and int(k) > 0): @@ -1085,33 +1117,48 @@ def write_parts(doc, elements, transform: Transform) -> int: return len(tags) // 5 -def read_markers(dxf_file: Path) -> list[dict[str, str]]: - """Attribute aller TRO-Marker (Bloecke TRO_SYM_*) der Zeichnung.""" +def read_markers(dxf_file: Path) -> tuple[list[dict[str, str]], list[dict[str, str]]]: + """Attribute aller TRO-Marker (Bloecke TRO_SYM_*) und aller + Verbindungsbloecke (CONNECTION_ARROW) der Zeichnung.""" doc = ezdxf.readfile(str(dxf_file)) - markers = [] + markers, arrows = [], [] for ref in doc.modelspace().query("INSERT"): - if not ref.dxf.name.upper().startswith(BLOCK_PREFIX): + name = ref.dxf.name.upper() + if name.startswith(BLOCK_PREFIX): + ziel = markers + elif name == CONNECTION_BLOCK: + ziel = arrows + else: continue a = {att.dxf.tag.upper(): (att.dxf.text or "").strip() for att in ref.attribs} a["_BLOCK"] = ref.dxf.name - markers.append(a) - return markers + ziel.append(a) + return markers, arrows def read_tros(markers: list[dict[str, str]], raw: Analysis, elements, csv_name: str, cfg_dir: Path | None = None, - write: bool = True) -> tuple[list[TroEdit], list[str]]: + write: bool = True, + arrows: list[dict[str, str]] | None = None, + base: Analysis | None = None) -> tuple[list[TroEdit], list[str]]: """Marker der Zeichnung mit der Ableitung vergleichen -> Handkorrekturen. ``raw`` ist die Ableitung OHNE Handkorrekturen (IDs aus der ID-Datei). Je Marker mit KEY eines abgeleiteten TRO entsteht ein edit-Abschnitt, wenn Typ, Kennung oder Bauteile abweichen; ein Marker ohne passenden KEY ist ein neuer TRO (edit_add); ein abgeleiteter TRO mit Lage, zu dem kein - Marker mehr da ist, wurde geloescht. Die Abschnitte ersetzen alle - bisherigen edit-/edit_add-Abschnitte der CSV in tro_overrides.ini - (Begruendungen bleiben). Rueckgabe: (Korrekturen, Meldungen). + Marker mehr da ist, wurde geloescht. Mit ``arrows`` (Verbindungsbloecke) + kommen die geaenderten Verbindungen dazu (_read_connections, Abschnitt + edit_connections). Die Abschnitte ersetzen alle bisherigen + edit-/edit_add-/edit_connections-Abschnitte der CSV in tro_overrides.ini + (Begruendungen bleiben). Ohne ``arrows`` bleibt ein vorhandener + edit_connections-Abschnitt stehen. ``base`` ist die Ableitung MIT den + TRO-Korrekturen, aber ohne Verbindungskorrekturen - verglichen werden die + Pfeile mit ihr, damit Kanten eines geaenderten oder neuen TRO nicht als + Handkorrektur gelten (Standard: ``raw``). Rueckgabe: (Korrekturen, Meldungen). """ meldungen: list[str] = [] + marker_key: dict[int, str] = {} by_id = {e.teile_id: e for e in elements} alt = {(e.section_kind, e.key): e for e in load_overrides(csv_name, cfg_dir).edits} @@ -1176,6 +1223,8 @@ def read_tros(markers: list[dict[str, str]], raw: Analysis, elements, raw_slots, raw_lines = tro_slots(derived) else: raw_slots, raw_lines = {}, [] + # Schluessel des Markers fuer die Verbindungen: abgeleitet = KEY, neu = ID + marker_key[id(m)] = key if derived is not None else tro_id if lines is None: lines = raw_lines if has_lines(type_name) else [] pruefen(where, type_name, slots, lines) @@ -1216,11 +1265,64 @@ def read_tros(markers: list[dict[str, str]], raw: Analysis, elements, edits.append(edit) meldungen.append(_describe_edit(edit)) + if arrows is not None: + connections = _read_connections(arrows, markers, marker_key, base or raw, meldungen) + else: + # ohne Verbindungsbloecke: bisherige Verbindungskorrekturen behalten + connections = load_overrides(csv_name, cfg_dir).connection_edits if write: - save_edits(csv_name, edits, cfg_dir) + save_edits(csv_name, edits, cfg_dir, connections) return edits, meldungen +def _read_connections(arrows: list[dict[str, str]], markers: list[dict[str, str]], + marker_key: dict[int, str], base: Analysis, + meldungen: list[str]) -> list[ConnectionEdit]: + """Verbindungsbloecke der Zeichnung gegen die abgeleiteten TRO-Kanten. + + Die Enden eines CONNECTION_ARROW werden ueber die Marker derselben + Zeichnung zu stabilen Schluesseln: erst ueber den sichtbaren Namen + (FROM_TRO/TO_TRO, das sieht der Bearbeiter), sonst ueber die interne + Nummer (FROM_UID/TO_UID, bei dxfmakros-Bloecken frueher FROM_ID/TO_ID) - + so stimmt es auch nach einer Umbenennung des Markers. Kanten von/zu + Stationen tragen keinen Block und bleiben, wie abgeleitet; Kanten zu + geloeschten Markern fallen mit dem TRO weg. + """ + by_name = {m.get("ID", ""): m for m in markers if id(m) in marker_key} + by_uid = {m.get("UID", ""): m for m in markers if id(m) in marker_key and m.get("UID")} + + def ende(a: dict[str, str], seite: str) -> str | None: + m = by_name.get(a.get(f"{seite}_TRO", "")) + if m is None: + m = by_uid.get(a.get(f"{seite}_UID", "") or a.get(f"{seite}_ID", "")) + return marker_key.get(id(m)) if m is not None else None + + gewollt: dict[tuple[str, str], str] = {} + for a in arrows: + src, dst = ende(a, "FROM"), ende(a, "TO") + if src is None or dst is None: + meldungen.append(f"Verbindung {a.get('FROM_TRO') or '?'} -> " + f"{a.get('TO_TRO') or '?'}: Ende passt zu keinem Marker - " + f"uebersprungen") + continue + if src == dst: + meldungen.append(f"Verbindung {src} -> {dst}: Quelle = Ziel - uebersprungen") + continue + gewollt[(src, dst)] = "weiche" if a.get("KIND") == "weiche" else "" + + vorhanden = set(marker_key.values()) + tro_ids = {t.tro_id for t in base.tros} + # Kanten der Ableitung, Enden als stabile Schluessel wie die Marker + abgeleitet = {(base.edit_key(s), base.edit_key(d)) for s, d in base.edges + if s in tro_ids and d in tro_ids} + changes = [ConnectionEdit(add=False, src=s, dst=d) for s, d in sorted(abgeleitet) + if (s, d) not in gewollt and s in vorhanden and d in vorhanden] + changes += [ConnectionEdit(add=True, src=s, dst=d, kind=k) + for (s, d), k in sorted(gewollt.items()) if (s, d) not in abgeleitet] + meldungen += [f"Verbindung {c.render()}" for c in changes] + return changes + + def _describe_edit(edit: TroEdit) -> str: if edit.delete: return f"{edit.key}: geloescht" @@ -1462,7 +1564,27 @@ def _main_read_jams(args, csv_file: Path) -> int: return 0 -def _derive(args, csv_file: Path, edits: list[TroEdit] | None): +def analysis_from_document(doc: TroDocument) -> Analysis: + """Analysis aus einer TRO-/Anlagen-JSON, damit die Zeichenfunktionen + (place_markers, draw_connections, place_stations) die JSON zeigen statt + einer neuen Ableitung. Umbenennungen kommen aus DerivedTro.key.""" + tros = [t.model_copy(deep=True) for t in doc.tros] + analysis = Analysis( + tros=tros, + boundaries=[b.model_copy(deep=True) for b in doc.boundaries], + edges=[(c.src, c.dst) for c in doc.connections], + edge_kinds={(c.src, c.dst): c.kind for c in doc.connections}, + findings=list(doc.findings), + renamed={t.tro_id: t.key for t in tros if t.key and t.key != t.tro_id}, + method=f"Anlagen-/TRO-JSON {doc.source}", + ) + analysis.unknown_types = sorted({t.type_name for t in tros if t.definition() is None}) + analysis.isolated = [t for t in tros if not t.connected] + return analysis + + +def _derive(args, csv_file: Path, edits: list[TroEdit] | None, + connection_edits: list[ConnectionEdit] | None = None): """CSV lesen und die TROs ableiten - (Elemente, Analysis) oder None. Derselbe Aufruf wie in tro_flow.main (mit der Separator-Gegenprobe aus @@ -1490,7 +1612,8 @@ def _derive(args, csv_file: Path, edits: list[TroEdit] | None): print(f"FEHLER: ID-Datei nicht lesbar: {exc}", file=sys.stderr) return None analysis = analyse_tros(graph, elements, sepliste_counts=sepliste_counts, - tro_ids=tro_ids, edits=edits) + tro_ids=tro_ids, edits=edits, + connection_edits=connection_edits) if not analysis.tros: print(f"FEHLER: keine TROs aus {csv_file} ableitbar.", file=sys.stderr) return None @@ -1514,7 +1637,7 @@ def _main_read_tros(args, csv_file: Path) -> int: print(f"FEHLER: CSV nicht gefunden: {csv_file}", file=sys.stderr) return 1 try: - markers = read_markers(dxf_file) + markers, arrows = read_markers(dxf_file) except (OSError, ezdxf.DXFError) as exc: print(f"FEHLER: {dxf_file} nicht lesbar: {exc}", file=sys.stderr) return 1 @@ -1531,8 +1654,15 @@ def _main_read_tros(args, csv_file: Path) -> int: f"mit dem aktuellen tro_annotate neu beschriften?", file=sys.stderr) return 1 try: + # 1. TRO-Korrekturen aus den Markern; 2. die Pfeile gegen die Ableitung + # MIT diesen Korrekturen (sonst gaelten Kanten eines neuen TRO als + # von Hand gezogen) + edits, _ = read_tros(markers, raw, elements, csv_file.name, cfg_dir, write=False) + derived = _derive(args, csv_file, edits) + if derived is None: + return 1 edits, meldungen = read_tros(markers, raw, elements, csv_file.name, cfg_dir, - write=not args.check) + write=not args.check, arrows=arrows, base=derived[1]) except (OSError, configparser.Error) as exc: print(f"FEHLER: {exc}", file=sys.stderr) return 1 @@ -1540,7 +1670,7 @@ def _main_read_tros(args, csv_file: Path) -> int: print("================================================================") print("TRO-HANDKORREKTUREN UEBERNEHMEN") print("================================================================") - print(f"Zeichnung = {dxf_file} ({len(markers)} Marker)") + print(f"Zeichnung = {dxf_file} ({len(markers)} Marker, {len(arrows)} Verbindungen)") ziel = override_path(cfg_dir) print(f"Korrekturen = {len(edits)} -> " f"{'- (--check, nichts geschrieben)' if args.check else ziel}") @@ -1624,7 +1754,7 @@ def main(argv: list[str] | None = None) -> int: # TROs aus dem CSV, mit den Handkorrekturen aus tro_overrides.ini overrides = load_overrides(csv_file.name, cfg_dir) warnings += [w for w in overrides.warnings if ".edit" in w] - derived = _derive(args, csv_file, overrides.edits) + derived = _derive(args, csv_file, overrides.edits, overrides.connection_edits) if derived is None: return 1 elements, analysis = derived @@ -1654,14 +1784,17 @@ def main(argv: list[str] | None = None) -> int: counts = {"marker": 0, "flow": 0, "removed": 0, "layers": 0, "blocks": 0, "first_id": 0, "jams": 0} if anlage is not None: + # Mit Anlagen-JSON zeigt die Zeichnung genau sie - das, woraus die SCL + # wird. Die Ableitung aus dem CSV dient nur noch als Gegenprobe. aus_csv = {t.tro_id for t in analysis.tros} aus_json = {t.tro_id for t in anlage.layout.tros} if aus_csv != aus_json: warnings.append( - f"TROs der Zeichnung und der Anlagen-JSON weichen ab (nur CSV: " - f"{', '.join(sorted(aus_csv - aus_json)) or '-'}; nur JSON: " - f"{', '.join(sorted(aus_json - aus_csv)) or '-'}) - tro_flow --json und " - f"merge_hw_layout neu laufen lassen") + f"Die Zeichnung zeigt die Anlagen-JSON; die Ableitung aus dem CSV weicht " + f"ab (nur CSV: {', '.join(sorted(aus_csv - aus_json)) or '-'}; nur JSON: " + f"{', '.join(sorted(aus_json - aus_csv)) or '-'}) - JSON veraltet? " + f"tro_flow --json und merge_hw_layout neu laufen lassen") + analysis = analysis_from_document(anlage.layout) if args.check: counts["marker"] = sum(1 for t in analysis.tros if t.position) diff --git a/lib/tro_flow.py b/lib/tro_flow.py index 5915c39..1ab9a79 100644 --- a/lib/tro_flow.py +++ b/lib/tro_flow.py @@ -217,7 +217,7 @@ from tro_catalog import ( get_tro, mermaid_classdefs, ) -from tro_overrides import PartRef, TroEdit, load_overrides +from tro_overrides import ConnectionEdit, PartRef, TroEdit, load_overrides # --------------------------------------------------------------------------- # Konstanten @@ -2428,6 +2428,47 @@ def apply_edits(tros: list[DerivedTro], edits: list[TroEdit], elements: list[Ele return tros +def apply_connection_edits(analysis: Analysis, changes: list[ConnectionEdit]) -> None: + """Im Plan geaenderte Verbindungen (tro_overrides.ini, edit_connections). + + Laeuft nach link_tros: '-' nimmt eine abgeleitete Kante weg, '+' fuegt eine + hinzu (Art aus dem Eintrag, sonst transfer). Die Enden sind stabile + Schluessel; ueber Analysis.renamed werden sie zur aktuellen ID. Danach + werden Vorgaenger/Nachfolger neu gesetzt - die Staubereiche folgen daraus. + """ + tros = analysis.tros + analysis.boundaries + id_of = {analysis.edit_key(t.tro_id): t.tro_id for t in tros} + id_of.update({t.tro_id: t.tro_id for t in tros}) + edges = set(analysis.edges) + for change in changes: + src, dst = id_of.get(change.src), id_of.get(change.dst) + if src is None or dst is None: + fehlt = change.src if src is None else change.dst + analysis.findings.append(f"Handkorrektur Verbindung {change.render()}: {fehlt} " + f"gibt es nicht - ignoriert") + continue + if change.add: + if (src, dst) not in edges: + edges.add((src, dst)) + analysis.findings.append(f"Handkorrektur Verbindung: {src} -> {dst} von Hand " + f"hinzugefuegt") + analysis.edge_kinds[(src, dst)] = change.kind or "transfer" + elif change.kind: + analysis.edge_kinds[(src, dst)] = change.kind + elif (src, dst) in edges: + edges.discard((src, dst)) + analysis.edge_kinds.pop((src, dst), None) + analysis.findings.append(f"Handkorrektur Verbindung: {src} -> {dst} von Hand " + f"entfernt") + else: + analysis.findings.append(f"Handkorrektur Verbindung {change.render()}: wird " + f"nicht abgeleitet - nichts zu entfernen") + analysis.edges = sorted(edges) + for tro in tros: + tro.successors = sorted(d for s, d in analysis.edges if s == tro.tro_id) + tro.predecessors = sorted(s for s, d in analysis.edges if d == tro.tro_id) + + def analyse_tros( graph: Graph, elements: list[Element], @@ -2435,6 +2476,7 @@ def analyse_tros( sepliste_counts: dict[str, int] | None = None, tro_ids: TroIds | None = None, edits: list[TroEdit] | None = None, + connection_edits: list[ConnectionEdit] | None = None, ) -> Analysis: """TROs ermitteln, verknuepfen, verorten und gegen die Typdefinition pruefen. @@ -2452,6 +2494,8 @@ def analyse_tros( nummerieren. edits: Handkorrekturen aus BricsCAD (tro_overrides.load_overrides (...).edits), angewandt direkt nach derive_tros (apply_edits). + connection_edits: im Plan geaenderte Verbindungen + (load_overrides(...).connection_edits), angewandt nach link_tros. Rueckgabe: Analysis mit tros, boundaries, edges/edge_kinds, unknown_types @@ -2476,6 +2520,8 @@ def analyse_tros( # Kreisel-Bahn wird ueber ihre Koordinate bestimmt. locate_tros(linked, elements, graph) analysis.edges, analysis.edge_kinds = link_tros(graph, linked, elements) + if connection_edits: + apply_connection_edits(analysis, connection_edits) analysis.unknown_types = sorted( {tro.type_name for tro in analysis.tros if tro.definition() is None}) @@ -3598,7 +3644,8 @@ def main(argv: list[str] | None = None) -> int: warnings.append(f"--renumber: {len(overrides.edits)} Handkorrektur(en) in " f"{overrides.source} haengen an den alten IDs - pruefen") analysis = analyse_tros(graph, elements, sepliste_counts=sepliste_counts, - tro_ids=tro_ids, edits=overrides.edits) + tro_ids=tro_ids, edits=overrides.edits, + connection_edits=overrides.connection_edits) if not analysis.tros: print(f"FEHLER: {csv_file} enthaelt keine Separatoren - keine TROs " f"ableitbar.", file=sys.stderr) diff --git a/lib/tro_overrides.py b/lib/tro_overrides.py index d9233b9..4d1ff52 100644 --- a/lib/tro_overrides.py +++ b/lib/tro_overrides.py @@ -56,6 +56,11 @@ anders als die laufenden Nummern, an denen die alten Abschnitte veraltet sind. [.edit_add.] neuer TRO (TRO_INSERT), Schluessel wie oben + [.edit_connections] Verbindungen, im Plan (CONNECTION_*) geaendert + c01 = + TRO05 -> TRO07 | transfer dazu (Art optional, Standard transfer) + c02 = - TRO05 -> TRO01 weg + Enden sind die stabilen Schluessel der TROs + Stehen Slots im Abschnitt, gelten sie vollstaendig (fehlende Slots des Typs bleiben leer); stehen keine, bleiben die Bauteile wie abgeleitet. Die Lage hinter der TeileId macht den Eintrag exportfest: passt die TeileId im neuen @@ -220,6 +225,33 @@ class TroEdit: or self.slots is not None or self.lines is not None) +@dataclass +class ConnectionEdit: + """Eine im Plan hinzugefuegte (add) oder entfernte Verbindung zwischen + zwei TROs, Enden als stabile Schluessel (DerivedTro.key).""" + + add: bool + src: str + dst: str + kind: str = "" + + def render(self) -> str: + text = f"{'+' if self.add else '-'} {self.src} -> {self.dst}" + return f"{text} | {self.kind}" if self.add and self.kind else text + + @classmethod + def parse(cls, raw: str) -> ConnectionEdit: + text = raw.strip() + if not text or text[0] not in "+-": + raise ValueError(f"'{raw}': erwartet '+ -> [| Art]' oder " + f"'- -> '") + pfeil, _, kind = text[1:].partition("|") + src, sep, dst = (t.strip() for t in pfeil.partition("->")) + if not sep or not src or not dst: + raise ValueError(f"'{raw}': erwartet ' -> '") + return cls(add=text[0] == "+", src=src, dst=dst, kind=kind.strip()) + + @dataclass class Overrides: """Alle Korrekturen zu einer CSV-Datei. @@ -237,6 +269,8 @@ class Overrides: candidates: list[TroCandidate] = field(default_factory=list) # Handkorrekturen aus BricsCAD in Dateireihenfolge (tro_flow.apply_edits) edits: list[TroEdit] = field(default_factory=list) + # Verbindungen aus BricsCAD (tro_flow.apply_connection_edits) + connection_edits: list[ConnectionEdit] = field(default_factory=list) # Protokoll der tatsaechlich angewandten Korrekturen (fuer die Doku) applied: list[str] = field(default_factory=list) warnings: list[str] = field(default_factory=list) @@ -278,7 +312,8 @@ def load_overrides(csv_name: str, cfg_dir: Path | None = None) -> Overrides: parser = configparser.ConfigParser(inline_comment_prefixes=(";",)) # Schluessel sind Kleinschreibung, Sektionsnamen bleiben wie geschrieben. - parser.read(path, encoding="utf-8") + # utf-8-sig: von Hand (Editor, PowerShell) gespeichert oft mit BOM + parser.read(path, encoding="utf-8-sig") prefix = csv_name.lower() for section in parser.sections(): @@ -297,6 +332,14 @@ def load_overrides(csv_name: str, cfg_dir: Path | None = None) -> Overrides: Connection(src=src, dst=dst, kind=rest.strip()) ) continue + # [.edit_connections] - Verbindungen aus BricsCAD + if section.lower() == f"{prefix}.edit_connections": + for key, raw in parser[section].items(): + try: + result.connection_edits.append(ConnectionEdit.parse(raw)) + except ValueError as exc: + result.warnings.append(f"[{section}] {key}: {exc}") + continue parts = section.split(".") # .. - der CSV-Name kann selbst einen Punkt enthalten @@ -475,8 +518,10 @@ def render_edit(csv_name: str, edit: TroEdit) -> str: return "\n".join(lines) -def save_edits(csv_name: str, edits: list[TroEdit], cfg_dir: Path | None = None) -> Path: - """Alle edit-/edit_add-Abschnitte zu `csv_name` durch `edits` ersetzen. +def save_edits(csv_name: str, edits: list[TroEdit], cfg_dir: Path | None = None, + connections: list[ConnectionEdit] | None = None) -> Path: + """Alle edit-/edit_add-Abschnitte und [.edit_connections] zu + `csv_name` durch `edits` bzw. `connections` ersetzen. Der Rest der Datei (von Hand gepflegt, mit Kommentaren) bleibt Zeile fuer Zeile erhalten; der neue Block steht am Ende. Kommentarzeilen direkt vor @@ -488,6 +533,8 @@ def save_edits(csv_name: str, edits: list[TroEdit], cfg_dir: Path | None = None) head = _EDIT_HEAD.format(csv=csv_name).lower() def ist_edit(section: str) -> bool: + if section.lower() == f"{prefix}.edit_connections": + return True parts = section.split(".") return (len(parts) >= 3 and parts[-2].lower() in ("edit", "edit_add") and ".".join(parts[:-2]).lower() == prefix) @@ -519,13 +566,17 @@ def save_edits(csv_name: str, edits: list[TroEdit], cfg_dir: Path | None = None) while keep and not keep[-1].strip(): keep.pop() - if edits: + if edits or connections: if keep: keep += ["", ""] keep.append(_EDIT_HEAD.format(csv=csv_name) + f" - tro_annotate --read-tros, {datetime.now():%Y-%m-%d}") for edit in edits: keep += [render_edit(csv_name, edit), ""] + if connections: + keep.append(f"[{csv_name}.edit_connections]") + keep += [f"c{n:02d} = {c.render()}" for n, c in enumerate(connections, start=1)] + keep.append("") path.parent.mkdir(parents=True, exist_ok=True) path.write_text("\n".join(keep).rstrip("\n") + "\n", encoding="utf-8") return path diff --git a/lib/validate_anlage.py b/lib/validate_anlage.py new file mode 100644 index 0000000..3033574 --- /dev/null +++ b/lib/validate_anlage.py @@ -0,0 +1,364 @@ +# -*- coding: utf-8 -*- +# Absichtlich ohne Shebang: der Windows-py-Launcher wuerde daraus "python3" +# ableiten und scheitert, wenn keine PythonCore-Installation registriert ist. +""" +validate_anlage.py - Ist die Anlagen-JSON bereit fuer die SCL? + +Eine Pruefung fuer beide Wege des Ablaufs (doc/Konzept_Layout_Hardware_SCL.md): +direkt nach merge_hw_layout (keine Aenderungen) und nach den Handkorrekturen +in BricsCAD (tro_annotate --read-tros/--read-jams, dann tro_flow --json und +merge_hw_layout neu). scl_skeleton ruft sie vor dem Schreiben auf. + +Geprueft wird (FEHLER = so keine SCL, WARNUNG = SCL mit offenen Stellen): + + Struktur TRO-IDs eindeutig, jeder Typ im Katalog, Verbindungen und + Staubereiche zeigen auf vorhandene TROs/Stationen, kein + Separator in zwei TROs (FEHLER); leerer Separator-Slot, TRO + ohne Verbindung, TRO vom Materialeingang aus nicht erreichbar + (WARNUNG) + Zuordnungen Binding auf einen unbekannten TRO/Slot, Status widerspruch, + BMK fehlt in der Hardwareplanung (FEHLER); offene + Pflichtrollen (WARNUNG, je Rolle gezaehlt) + Aktualitaet Anlagen-JSON aelter als TRO-JSON, Hardwareplanung oder + hw_overrides.ini (FEHLER: merge_hw_layout neu); + TRO-JSON aelter als die BricsCAD-Korrekturen in + tro_overrides.ini (WARNUNG: tro_flow --json neu) + Korrekturen Eintraege in hw_overrides.ini / _jam_names.ini, die zu + keinem TRO bzw. Staubereich passen - weder ueber die ID noch + ueber den stabilen Schluessel (WARNUNG) + +Aufruf ueber bin/validate_anlage.bat bzw. bin/validate_anlage.sh: + + validate_anlage --json mubea_anlage.json [--hw mubea_hw.json] [--strict] + +Exit-Codes: 0 bereit (mit --strict: auch ohne Warnungen), 1 Eingabefehler, +5 Fehler gefunden (mit --strict auch bei Warnungen). +""" + +from __future__ import annotations + +import argparse +import collections +import configparser +import json +import sys +from dataclasses import dataclass, field +from pathlib import Path + +from pydantic import ValidationError + +from hw_import import HardwareDocument +from material_flow import env_dir +from model import ANLAGE_SCHEMA, AnlageDocument, TroDocument +from scl_context import JAM_NAMES_SUFFIX, jam_key, jam_stable_key, load_jam_overrides +from tro_overrides import load_overrides, override_path + +FEHLER = "FEHLER" +WARNUNG = "WARNUNG" +# Name der Datei mit den Handkorrekturen der Hardware (merge_hw_layout) +HW_OVERRIDES_FILE = "hw_overrides.ini" +EXIT_FEHLER = 5 + + +@dataclass +class Befund: + stufe: str + bereich: str + text: str + + def __str__(self) -> str: + return f"{self.stufe:<7} {self.bereich:<11} {self.text}" + + +@dataclass +class Pruefung: + """Ergebnis einer Pruefung, Befunde in der Reihenfolge der Bereiche.""" + + befunde: list[Befund] = field(default_factory=list) + + def fehler(self, bereich: str, text: str) -> None: + self.befunde.append(Befund(FEHLER, bereich, text)) + + def warnung(self, bereich: str, text: str) -> None: + self.befunde.append(Befund(WARNUNG, bereich, text)) + + @property + def anzahl_fehler(self) -> int: + return sum(1 for b in self.befunde if b.stufe == FEHLER) + + @property + def anzahl_warnungen(self) -> int: + return sum(1 for b in self.befunde if b.stufe == WARNUNG) + + def bereit(self, strict: bool = False) -> bool: + return not self.anzahl_fehler and not (strict and self.anzahl_warnungen) + + +# --------------------------------------------------------------------------- +# Einzelpruefungen +# --------------------------------------------------------------------------- + + +def check_structure(doc: TroDocument, p: Pruefung) -> None: + """IDs, Typen, Verbindungen, Staubereiche, Separatoren, Erreichbarkeit.""" + ids = collections.Counter(t.tro_id for t in doc.tros + doc.boundaries) + for tro_id, n in sorted(ids.items()): + if n > 1: + p.fehler("Struktur", f"TRO-ID {tro_id} kommt {n}x vor") + for tro in doc.tros: + if tro.definition() is None: + p.fehler("Struktur", f"{tro.tro_id}: Typ '{tro.type_name}' steht nicht im Katalog") + bekannt = set(ids) + for c in doc.connections: + for ende in (c.src, c.dst): + if ende not in bekannt: + p.fehler("Struktur", f"Verbindung {c.src} -> {c.dst}: {ende} unbekannt") + for area in doc.jam_areas_or_derived(): + if area.tro not in bekannt: + p.fehler("Struktur", f"Staubereich {area.jam_id}: TRO {area.tro} unbekannt") + + besitzer: dict[str, str] = {} + for tro in doc.tros: + for sep in dict.fromkeys(tro.separators): + if not sep: + continue + if sep in besitzer: + p.fehler("Struktur", f"Separator {sep} steckt in {besitzer[sep]} und " + f"{tro.tro_id}") + besitzer.setdefault(sep, tro.tro_id) + for slot in tro.slots: + if slot.slot.startswith("separator") and not slot.layout_id: + p.warnung("Struktur", f"{tro.tro_id}: {slot.slot} ohne Layout-Bauteil") + + # Erreichbarkeit vom Materialeingang (Systemgrenzen ohne Vorgaenger) + nach: dict[str, set[str]] = collections.defaultdict(set) + for c in doc.connections: + nach[c.src].add(c.dst) + for tro in doc.tros + doc.boundaries: + nach[tro.tro_id].update(tro.successors) + for tro in doc.tros: + if not (tro.successors or tro.predecessors or tro.colocated): + p.warnung("Struktur", f"{tro.tro_id}: ohne Verbindung") + starts = [b.tro_id for b in doc.boundaries if not b.predecessors] + if starts: + gesehen, stapel = set(starts), list(starts) + while stapel: + for nxt in nach.get(stapel.pop(), ()): + if nxt not in gesehen: + gesehen.add(nxt) + stapel.append(nxt) + weg = [t.tro_id for t in doc.tros if t.tro_id not in gesehen + and (t.successors or t.predecessors)] + if weg: + p.warnung("Struktur", f"vom Materialeingang nicht erreichbar: {', '.join(weg)}") + + +def check_bindings(anlage: AnlageDocument, hw: HardwareDocument | None, + p: Pruefung) -> None: + """Zuordnungen der Hardwareplanung gegen Layout und TIA-Import.""" + doc = anlage.layout + slots = {(t.tro_id, s.slot) for t in doc.tros for s in t.slots} + tros = {t.tro_id for t in doc.tros} + geraete = hw.by_bmk() if hw is not None else None + offen: collections.Counter[str] = collections.Counter() + for b in anlage.bindings: + wo = f"{b.tro}.{b.slot}.{b.role}" if b.tro else (b.layout_id or b.bmk) + if b.tro and b.tro not in tros: + p.fehler("Zuordnung", f"{wo}: TRO {b.tro} steht nicht im Layout") + continue + if b.tro and (b.tro, b.slot) not in slots: + p.fehler("Zuordnung", f"{wo}: Slot {b.slot} gibt es an {b.tro} nicht") + continue + if b.status == "widerspruch": + p.fehler("Zuordnung", f"{wo}: {b.bmk or '-'} - {b.note or 'Widerspruch'}") + elif b.status == "offen": + offen[b.role] += 1 + if geraete is not None and b.bmk and b.status != "widerspruch" and b.bmk not in geraete: + p.fehler("Zuordnung", f"{wo}: {b.bmk} fehlt in der Hardwareplanung") + if offen: + p.warnung("Zuordnung", "offene Pflichtrollen (TODO(E-Planung)): " + + ", ".join(f"{r} {n}x" for r, n in sorted(offen.items()))) + + +def _mtime(path: Path | None) -> float | None: + return path.stat().st_mtime if path is not None and path.is_file() else None + + +def _has_sections(path: Path, csv_name: str) -> bool: + """Hat die INI einen Abschnitt [.*]?""" + if not path.is_file(): + return False + prefix = f"[{csv_name.lower()}." + try: + return any(line.strip().lower().startswith(prefix) + for line in path.read_text(encoding="utf-8-sig").splitlines()) + except OSError: + return False + + +def check_freshness(anlage_file: Path, anlage: AnlageDocument | None, tro_file: Path | None, + hw_file: Path | None, cfg_dir: Path, csv_name: str, + p: Pruefung) -> None: + """Ist die JSON juenger als alles, woraus sie gebaut wird?""" + stand = _mtime(anlage_file) + if stand is None: + return + if anlage is not None: + hw_ini = cfg_dir / HW_OVERRIDES_FILE + # hw_overrides.ini zaehlt nur mit Abschnitten zu dieser Anlage + if not _has_sections(hw_ini, csv_name): + hw_ini = None + for quelle, was in ((tro_file, "TRO-JSON"), (hw_file, "Hardwareplanung"), + (hw_ini, HW_OVERRIDES_FILE)): + t = _mtime(quelle) + if t is not None and t > stand: + p.fehler("Stand", f"{anlage_file.name} ist aelter als {was} ({quelle.name}) " + f"- merge_hw_layout neu laufen lassen") + # BricsCAD-Korrekturen wirken in tro_flow - die TRO-JSON muss juenger sein + layout_stand = _mtime(tro_file) if anlage is not None else stand + korrektur = override_path(cfg_dir) + t = _mtime(korrektur) + if layout_stand is not None and t is not None and t > layout_stand: + try: + if load_overrides(csv_name, cfg_dir).edits: + p.warnung("Stand", f"{korrektur.name} (Handkorrekturen aus BricsCAD) ist " + f"juenger als die TRO-JSON - tro_flow --json neu laufen lassen") + except (OSError, configparser.Error): + pass + + +def check_corrections(doc: TroDocument, cfg_dir: Path, csv_name: str, p: Pruefung) -> None: + """Handkorrekturen, die ins Leere zeigen (z. B. nach einer Umbenennung).""" + hw_ini = cfg_dir / HW_OVERRIDES_FILE + if hw_ini.is_file(): + parser = configparser.ConfigParser(inline_comment_prefixes=(";", "#"), + interpolation=None) + parser.optionxform = str + try: + with hw_ini.open(encoding="utf-8-sig") as fh: + parser.read_file(fh) + except (OSError, configparser.Error) as exc: + p.fehler("Korrektur", f"{hw_ini.name} nicht lesbar: {exc}") + else: + section = f"{csv_name}.bindings" + for key in (parser[section] if parser.has_section(section) else {}): + teile = key.split(".") + tro = doc.resolve(teile[0]) if len(teile) == 3 else None + if tro is None or not any(s.slot == teile[1] for s in tro.slots): + p.warnung("Korrektur", f"{hw_ini.name} [{section}] {key}: kein TRO/Slot " + f"dazu (ID und Schluessel geprueft)") + jam_ini = cfg_dir / f"{Path(csv_name).stem}{JAM_NAMES_SUFFIX}" + try: + jam = load_jam_overrides(jam_ini) + except (OSError, ValueError, configparser.Error) as exc: + p.fehler("Korrektur", f"{jam_ini.name} nicht lesbar: {exc}") + return + if jam: + areas = doc.jam_areas_or_derived() + treffer = {k for a in areas for k in (jam_stable_key(a, doc), jam_key(a))} + for key in sorted(set(jam) - treffer): + p.warnung("Korrektur", f"{jam_ini.name} {key}: kein Staubereich dazu") + + +# --------------------------------------------------------------------------- +# Gesamt +# --------------------------------------------------------------------------- + + +def validate(json_file: Path, *, hw_file: Path | None = None, + results: Path | None = None, cfg_dir: Path | None = None) -> Pruefung: + """TRO- oder Anlagen-JSON pruefen. Ladefehler werden zu einem FEHLER.""" + results = results or json_file.parent + cfg_dir = cfg_dir or env_dir("SKEL_CFG", "cfg") + p = Pruefung() + try: + text = json_file.read_text(encoding="utf-8") + if json.loads(text).get("schema") == ANLAGE_SCHEMA: + anlage: AnlageDocument | None = AnlageDocument.from_json(text) + doc = anlage.layout + else: + anlage, doc = None, TroDocument.from_json(text) + except (OSError, ValueError, ValidationError) as exc: + p.fehler("Modell", f"{json_file.name} nicht lesbar oder passt nicht zum Modell: " + f"{str(exc).splitlines()[0]}") + return p + + csv_name = Path(doc.source).name + check_structure(doc, p) + + hw = None + tro_file = None + if anlage is not None: + tro_file = results / Path(anlage.layout_file).name if anlage.layout_file else None + if hw_file is None and anlage.hw_file: + hw_file = results / Path(anlage.hw_file).name + if hw_file is not None and hw_file.is_file(): + try: + hw = HardwareDocument.from_json(hw_file.read_text(encoding="utf-8")) + except (OSError, ValidationError) as exc: + p.fehler("Zuordnung", f"{hw_file.name} nicht lesbar: {str(exc).splitlines()[0]}") + elif hw_file is not None: + p.warnung("Zuordnung", f"{hw_file.name} nicht gefunden - BMK nicht gegengeprueft") + check_bindings(anlage, hw, p) + check_freshness(json_file, anlage, tro_file, hw_file, cfg_dir, csv_name, p) + check_corrections(doc, cfg_dir, csv_name, p) + return p + + +def report(json_file: Path, p: Pruefung, strict: bool, out=sys.stdout) -> None: + print("", file=out) + print("================================================================", file=out) + print("PRUEFUNG ANLAGEN-JSON", file=out) + print("================================================================", file=out) + print(f"Datei = {json_file}", file=out) + print(f"Fehler = {p.anzahl_fehler}", file=out) + print(f"Warnungen = {p.anzahl_warnungen}", file=out) + print(f"Ergebnis = {'bereit fuer die SCL' if p.bereit(strict) else 'NICHT bereit'}" + + (" (--strict: Warnungen zaehlen)" if strict else ""), file=out) + print("================================================================", file=out) + for b in p.befunde: + print(f" {b}", file=out) + print("", file=out) + + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser( + prog="validate_anlage", + description="Prueft eine Anlagen-JSON (oder TRO-JSON), bevor daraus SCL wird.", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog="Beispiele:\n" + " validate_anlage.bat --json mubea_anlage.json\n" + " validate_anlage.bat --json dematic_merge_anlage.json --strict\n" + "\nExit-Codes: 0 bereit, 1 Aufruffehler, 5 nicht bereit\n", + ) + parser.add_argument("--json", required=True, metavar="NAME", + help="Anlagen-JSON (_anlage.json) oder TRO-JSON, in " + "%%SKEL_RESULTS%% oder ein Pfad.") + parser.add_argument("--hw", metavar="NAME", + help="Hardwareplanung (_hw.json). Standard: hwFile " + "der Anlagen-JSON.") + parser.add_argument("--strict", action="store_true", + help="Auch Warnungen machen die JSON 'nicht bereit'.") + return parser.parse_args(argv) + + +def _in_results(name: str, results: Path) -> Path: + path = Path(name) + return results / name if not path.is_absolute() and path.parent == Path(".") else path + + +def main(argv: list[str] | None = None) -> int: + args = parse_args(argv) + results = env_dir("SKEL_RESULTS", "results") + json_file = _in_results(args.json, results) + if not json_file.is_file(): + print(f"FEHLER: JSON nicht gefunden: {json_file}", file=sys.stderr) + return 1 + hw_file = _in_results(args.hw, results) if args.hw else None + p = validate(json_file, hw_file=hw_file, results=json_file.parent) + report(json_file, p, args.strict) + return 0 if p.bereit(args.strict) else EXIT_FEHLER + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_anlage_workflow.py b/tests/test_anlage_workflow.py new file mode 100644 index 0000000..ade740c --- /dev/null +++ b/tests/test_anlage_workflow.py @@ -0,0 +1,232 @@ +# -*- coding: utf-8 -*- +"""Tests fuer den Ablauf Anlagen-JSON -> (BricsCAD) -> SCL. + +- stabiler Schluessel (DerivedTro.key): Korrekturen ueberleben eine Umbenennung +- Verbindungen aus BricsCAD (edit_connections, tro_flow.apply_connection_edits, + tro_annotate._read_connections) +- Pruefung vor der SCL (validate_anlage) +- Zeichnung aus der JSON (tro_annotate.analysis_from_document) + +Fixture: Testanlage test_system (wie test_tro_edits.py). +""" + +from __future__ import annotations + +import os +from pathlib import Path + +import pytest + +from material_flow import build_graph, read_elements +from model import AnlageDocument, Binding, TroDocument +from scl_context import jam_override, jam_stable_key +from tro_annotate import analysis_from_document, format_slots, read_tros, tro_slots +from tro_flow import analyse_tros +from tro_overrides import ConnectionEdit, TroEdit, load_overrides, save_edits +from validate_anlage import FEHLER, validate + +FIXTURE = Path(__file__).parent / "fixtures" / "test_system_export.csv" +CSV = FIXTURE.name + + +@pytest.fixture(scope="module") +def layout(): + elements = read_elements(FIXTURE, []) + return elements, build_graph(elements, CSV, []) + + +def _analyse(layout, edits=None, connection_edits=None): + elements, graph = layout + return analyse_tros(graph, elements, edits=edits, connection_edits=connection_edits) + + +def _tro_kante(analysis): + """Eine abgeleitete Kante zwischen zwei TROs (keine Station).""" + ids = {t.tro_id for t in analysis.tros} + return next((s, d) for s, d in analysis.edges if s in ids and d in ids) + + +def _markers(analysis): + return [{"ID": t.tro_id, "KEY": analysis.edit_key(t.tro_id), "TYPE": t.type_name, + "UID": f"{n:04d}", "SLOTS": format_slots(*tro_slots(t)), + "_BLOCK": f"TRO_SYM_{t.type_name}"} + for n, t in enumerate(analysis.tros, start=100)] + + +def _arrows(analysis, markers): + uid = {m["ID"]: m["UID"] for m in markers} + ids = {t.tro_id for t in analysis.tros} + return [{"FROM_TRO": s, "TO_TRO": d, "FROM_UID": uid[s], "TO_UID": uid[d], + "KIND": "flow", "_BLOCK": "CONNECTION_ARROW"} + for s, d in analysis.edges if s in ids and d in ids] + + +# --------------------------------------------------------------------------- +# Stabiler Schluessel +# --------------------------------------------------------------------------- + + +def test_umbenannter_tro_traegt_seinen_schluessel(layout): + raw = _analyse(layout) + key = raw.tros[0].tro_id + doc = TroDocument.from_analysis(_analyse(layout, [TroEdit(key=key, name="Neu_1")]), CSV) + tro = doc.by_id("Neu_1") + assert tro.key == key and tro.stable_key == key + assert doc.resolve(key) is tro and doc.resolve("Neu_1") is tro + # Staubereichs-Korrektur unter dem alten Schluessel gilt weiter + area = next(a for a in doc.jam_areas_or_derived() if a.tro == "Neu_1") + assert jam_stable_key(area, doc) == f"{key}.{area.entry}" + assert jam_override(area, doc, {f"{key}.{area.entry}": ("0042", 3)}) == ("0042", 3) + # unveraenderte TROs: Schluessel = ID + assert all(t.key == t.tro_id for t in doc.tros if t.tro_id != "Neu_1") + + +# --------------------------------------------------------------------------- +# Verbindungen +# --------------------------------------------------------------------------- + + +def test_connection_edit_hin_und_zurueck(tmp_path): + c = ConnectionEdit.parse("+ TRO01 -> TRO05 | weiche") + assert (c.add, c.src, c.dst, c.kind) == (True, "TRO01", "TRO05", "weiche") + assert ConnectionEdit.parse(c.render()) == c + assert not ConnectionEdit.parse("- TRO01 -> TRO05").add + with pytest.raises(ValueError): + ConnectionEdit.parse("TRO01 -> TRO05") + save_edits(CSV, [TroEdit(key="TRO02", name="X")], tmp_path, + [c, ConnectionEdit(add=False, src="TRO03", dst="TRO04")]) + o = load_overrides(CSV, tmp_path) + assert o.connection_edits == [c, ConnectionEdit(add=False, src="TRO03", dst="TRO04")] + assert [e.key for e in o.edits] == ["TRO02"] + save_edits(CSV, [], tmp_path, []) + assert "edit_connections" not in (tmp_path / "tro_overrides.ini").read_text(encoding="utf-8") + + +def test_verbindungen_aendern_vorgaenger_und_nachfolger(layout): + raw = _analyse(layout) + s, d = _tro_kante(raw) + andere = next(t.tro_id for t in raw.tros + if t.tro_id not in (s, d) and (t.tro_id, d) not in raw.edges) + analysis = _analyse(layout, connection_edits=[ + ConnectionEdit(add=False, src=s, dst=d), + ConnectionEdit(add=True, src=andere, dst=d), + ConnectionEdit(add=True, src="GIBTSNICHT", dst=d)]) + assert (s, d) not in analysis.edges and (andere, d) in analysis.edges + assert analysis.edge_kinds[(andere, d)] == "transfer" + ziel = analysis.by_id(d) + assert andere in ziel.predecessors and s not in ziel.predecessors + assert d not in analysis.by_id(s).successors + assert any("GIBTSNICHT gibt es nicht" in f for f in analysis.findings) + # die Staubereiche folgen den Vorgaengern + doc = TroDocument.from_analysis(analysis, CSV) + eingang = next(a for a in doc.jam_areas if a.tro == d and a.entry == 1) + assert andere in eingang.predecessors + + +def test_verbindung_ueber_schluessel_nach_umbenennung(layout): + raw = _analyse(layout) + s, d = _tro_kante(raw) + analysis = _analyse(layout, [TroEdit(key=s, name="Quelle_1")], + [ConnectionEdit(add=False, src=s, dst=d)]) + assert ("Quelle_1", d) not in analysis.edges + + +def test_pfeile_der_zeichnung_werden_zu_verbindungskorrekturen(layout, tmp_path): + elements, _ = layout + raw = _analyse(layout) + markers = _markers(raw) + arrows = _arrows(raw, markers) + # unveraendert: keine Korrektur + _, meldungen = read_tros(markers, raw, elements, CSV, tmp_path, arrows=arrows) + assert load_overrides(CSV, tmp_path).connection_edits == [] + + s, d = _tro_kante(raw) + andere = next(t.tro_id for t in raw.tros + if t.tro_id not in (s, d) and (t.tro_id, d) not in raw.edges) + arrows = [a for a in arrows if (a["FROM_TRO"], a["TO_TRO"]) != (s, d)] + arrows.append({"FROM_TRO": andere, "TO_TRO": d, "FROM_UID": "", "TO_UID": "", + "KIND": "flow", "_BLOCK": "CONNECTION_ARROW"}) + # ein Marker umbenannt, seine Pfeile tragen noch den alten Namen -> ueber UID + quelle = next(m for m in markers if m["ID"] == s) + quelle["ID"] = "Quelle_1" + read_tros(markers, raw, elements, CSV, tmp_path, arrows=arrows) + o = load_overrides(CSV, tmp_path) + assert o.connection_edits == [ConnectionEdit(add=False, src=s, dst=d), + ConnectionEdit(add=True, src=andere, dst=d)] + assert [(e.key, e.name) for e in o.edits] == [(s, "Quelle_1")] + + # Rundlauf: angewandt, neu gezeichnet, wieder gelesen -> dasselbe + edited = _analyse(layout, o.edits, o.connection_edits) + m2 = _markers(edited) + again, _ = read_tros(m2, raw, elements, CSV, tmp_path, write=False, + arrows=_arrows(edited, m2), base=_analyse(layout, o.edits)) + assert [(e.key, e.name) for e in again] == [(s, "Quelle_1")] + + +# --------------------------------------------------------------------------- +# Pruefung +# --------------------------------------------------------------------------- + + +def _anlage_dateien(layout, tmp_path, bindings=None): + doc = TroDocument.from_analysis(_analyse(layout), CSV) + tro_file = tmp_path / "x_tro.json" + tro_file.write_text(doc.to_json(), encoding="utf-8") + anlage = AnlageDocument(layout=doc, layout_file=tro_file.name, bindings=bindings or []) + anlage_file = tmp_path / "x_anlage.json" + anlage_file.write_text(anlage.model_dump_json(by_alias=True, indent=2), encoding="utf-8") + return tro_file, anlage_file, doc + + +def test_pruefung_frische_anlage_ist_bereit(layout, tmp_path): + tro_file, anlage_file, _ = _anlage_dateien(layout, tmp_path) + os.utime(tro_file, (1_000_000, 1_000_000)) + p = validate(anlage_file, cfg_dir=tmp_path) + assert p.anzahl_fehler == 0, [str(b) for b in p.befunde] + assert p.bereit() + + +def test_pruefung_findet_fehler(layout, tmp_path): + tro = TroDocument.from_analysis(_analyse(layout), CSV).tros[0] + bindings = [Binding(tro="TRO99", slot="separator1", role="sensor", bmk="BG1"), + Binding(tro=tro.tro_id, slot="separator1", role="sensor", bmk="BG2", + status="widerspruch", note="BMK nicht im TIA-Import"), + Binding(tro=tro.tro_id, slot="separator1", role="stopper", status="offen")] + tro_file, anlage_file, _ = _anlage_dateien(layout, tmp_path, bindings) + # TRO-JSON juenger als die Anlagen-JSON -> veraltet + os.utime(anlage_file, (1_000_000, 1_000_000)) + (tmp_path / "x_jam_names.ini").write_text("[jam]\nTRO99.1 = 0042_1\n", encoding="utf-8") + p = validate(anlage_file, cfg_dir=tmp_path) + texte = [b.text for b in p.befunde if b.stufe == FEHLER] + assert any("TRO99 steht nicht im Layout" in t for t in texte) + assert any("BMK nicht im TIA-Import" in t for t in texte) + assert any("aelter als TRO-JSON" in t for t in texte) + assert any("stopper 1x" in b.text for b in p.befunde) + assert not p.bereit() + # die Staubereichsdatei heisst nach der CSV, nicht nach der JSON + assert not any("TRO99.1" in b.text for b in p.befunde) + + +def test_pruefung_unbekannte_verbindung(layout, tmp_path): + doc = TroDocument.from_analysis(_analyse(layout), CSV) + doc.connections[0].dst = "TRO99" + f = tmp_path / "x_tro.json" + f.write_text(doc.to_json(), encoding="utf-8") + p = validate(f, cfg_dir=tmp_path) + assert any("TRO99 unbekannt" in b.text for b in p.befunde if b.stufe == FEHLER) + + +# --------------------------------------------------------------------------- +# Zeichnung aus der JSON +# --------------------------------------------------------------------------- + + +def test_zeichnung_zeigt_die_json(layout): + raw = _analyse(layout) + s = raw.tros[0].tro_id + doc = TroDocument.from_analysis(_analyse(layout, [TroEdit(key=s, name="Neu_1")]), CSV) + a = analysis_from_document(doc) + assert [t.tro_id for t in a.tros] == [t.tro_id for t in doc.tros] + assert set(a.edges) == {(c.src, c.dst) for c in doc.connections} + assert a.edit_key("Neu_1") == s + assert [b.tro_id for b in a.boundaries] == [b.tro_id for b in doc.boundaries]