diff --git a/CLAUDE.md b/CLAUDE.md index 3551a10..a6555c4 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -10,22 +10,24 @@ of Siemens SCL code** ("TRO" — Transfer Route Object — blocks for a conveyor control system) that can be imported directly into TIA Portal. The repository has moved past pure analysis: `lib/` now holds working Python tooling that -runs the **full DXF → JSON → SCL pipeline in a first version**. From a CSV export (ILS 2.0) -it derives a material-flow graph and a TRO list, annotates a copy of the BricsCAD drawing -with the result, reads the (possibly hand-edited) drawing back into a TRO-JSON, and emits an -`FB_Main` **SCL skeleton** from that JSON. The skeleton is deliberately not a running +runs the **full CSV → TRO-JSON → SCL pipeline in a first version**. From a CSV export (ILS +2.0) it derives a material-flow graph and a TRO list, writes the TROs as a TRO-JSON defined +by the shared pydantic model `lib/model.py`, and emits an `FB_Main` **SCL skeleton** from +that JSON. A copy of the BricsCAD drawing annotated with the same TROs serves as a viewer +for visual checking; it is no longer a data source. The skeleton is deliberately not a running program — every value that must come from the electrical planning is left as a visible `TODO(E-Planung)` gap rather than guessed. What is **not** done yet is the fuller layout-JSON schema (`doc/HundM/Json_Layout-Konzept.md`) as an intermediate format and the generation of the other per-controller blocks (`FB_CallSensors`, `FC_Direction`/`FC_Call_Jams` as -standalone files); see "Roadmap" below. `tests/` and `examples/` are still empty scaffolding -(see "Standard Programm Template" below). What exists today is: +standalone files); see "Roadmap" below. `tests/` holds the first tests +(`tests/test_model.py`); `examples/` is still empty scaffolding (see "Standard Programm +Template" below). What exists today is: - `bin/` — environment/venv management scripts, plus one `.bat`/`.sh` wrapper pair per CLI tool in `lib/` (see "Environment scripts" below) -- `lib/*.py` — CLI tools and libraries that turn a CSV export + BricsCAD DXF into a - material-flow graph, a derived TRO list/diagram, an annotated copy of the drawing, a - TRO-JSON read back out of that drawing, and an `FB_Main` SCL skeleton. +- `lib/*.py` — CLI tools and libraries that turn a CSV export into a material-flow graph, + a derived TRO list/diagram, a TRO-JSON (`lib/model.py`), an `FB_Main` SCL skeleton, and + (with the BricsCAD DXF) an annotated copy of the drawing for visual checking. See **`doc/Python_Scripts.md`** for what each script does and which switches it takes. - `data/` — input CSV exports (gitignored, not committed) - `cad/tro_types.lsp` — generated type list for the BricsCAD `TROEDIT` dialog (written by @@ -73,13 +75,12 @@ source bin/activate_venv.sh ``` Requires Python 3.10+. `requirements.txt` pins `pydantic>=2.0.0` (used throughout `lib/` -for validated models) and `ezdxf>=1.4.0` (DXF read/write, `tro_annotate.py` only); `pytest` -is commented out as not yet needed. +for validated models), `ezdxf>=1.4.0` (DXF read/write, `tro_annotate.py` only) and +`pytest>=9.0.0`. -There is no build/lint/test command configured yet — `tests/` is empty and no test -runner or linter is set up. Once tests exist, use `pytest` (already anticipated in -`requirements.txt`, just commented out) and set `PYTHONPATH` via `bin/setenv` first so -imports resolve against `lib/`. +Tests: `python -m pytest tests/` with `.venv` active. `tests/conftest.py` puts `lib/` on +`sys.path`, so `bin/setenv` is not required for the tests. No build step or linter is set +up. ## Domain model (from doc/) @@ -89,8 +90,8 @@ this order for onboarding: 0. **`doc/Python_Scripts.md`** — the existing CLI tooling (`lib/material_flow.py`, `lib/tro_flow.py`, `lib/tro_annotate.py`, `lib/tro_extract.py`, `lib/scl_skeleton.py`, - plus the `lib/tro_catalog.py`, `lib/dxf_registration.py` and `lib/tro_overrides.py` - libraries they share): what each script does, its switches, inputs/outputs. This is + plus the `lib/tro_catalog.py`, `lib/model.py`, `lib/dxf_registration.py` and + `lib/tro_overrides.py` libraries they share): what each script does, its switches, inputs/outputs. This is working code, not a proposal — read it before touching `lib/`. 1. **`doc/HundM/Json_Layout-Konzept.md`** — the core proposal: a JSON file as single source of truth (`plc`, `controlUnits`, `sensors[]`, `conveyors[]`, `tros[]`, `loadingBooms[]`, @@ -146,19 +147,20 @@ local, fully-vendored SCL source (`FB_ILS_MTRO_Vario_workStation`, `FB_EmptyCarr `doc/TRO_Katalog/scl_templates/*.scl` as **read-only reference material** for pattern extraction, not code to execute or modify. -## Roadmap: DXF → JSON → SCL (first version implemented) +## Roadmap: CSV → TRO-JSON → SCL (first version implemented) -The DXF → JSON → SCL pipeline now runs end-to-end in a first version. After `tro_annotate.py` -burns the TROs into a copy of the drawing, the user can hand-edit that DXF in BricsCAD (via -the `TRO_INSERT`/`TRO_EDIT` commands, see `doc/HundM/BricsCAD_TRO_Symbol.md`) — moving TROs, -adding new ones, changing a type — and two further `lib/` tools carry the drawing the rest of -the way to importable SCL: +Since 2026-10-09 the main path is **CSV → `tro_flow.py --json` → TRO-JSON → +`scl_skeleton.py` → SCL**. Layer order (who may import whom): `tro_catalog.py` (TRO *types*) +← `lib/model.py` (TRO *instances* and the TRO-JSON: `TroBase` → `DerivedTro`, `Connection`, +`OpenPoint`, `Analysis`, `TroDocument`) ← derivation (`material_flow.py`, `tro_flow.py`) ← +outputs (`scl_skeleton.py`, `tro_annotate.py`, reports). `items`/`fbType` are always read +from the catalog via the TRO type, never stored. See `doc/Python_Scripts.md` §6b. -1. **`lib/tro_extract.py`** (implemented) — reads the (possibly hand-edited) annotated DXF - **only** (not the CSV) and derives a TRO-JSON (`_tro.json`): the TROs tagged with - XDATA, their `predecessors`/`successors` computed from the flow arrows, and plant - coordinates resolved through the persisted registration. See `doc/Python_Scripts.md` §5a. -2. **`lib/scl_skeleton.py`** (implemented) — reads that TRO-JSON and emits an `FB_Main` SCL +1. **`lib/tro_flow.py --json`** (implemented) — writes `_tro.json` (schema + `sps_skel-tro-v2`, `model.TroDocument`): TROs, system boundaries `SYSn`, connections with + their kind, findings. 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 parameter line in the right order, plus inline `FC_Direction` calls for switch TROs. Everything derivable from the layout is filled (REGION order, FB type, parameter-block @@ -166,20 +168,27 @@ the way to importable SCL: everything that must come from the electrical planning is a `TODO(E-Planung)` gap. Switches `--json`, `--out`, `--start` (the last picks an entry point while the plant is a closed loop with no loading/unloading station). See `doc/Python_Scripts.md` §5b. +3. **`lib/tro_annotate.py`** — **viewer only**: draws the same TROs (same `analyse_tros` + call as `tro_flow.py`) into a copy of the DXF for a visual check in BricsCAD. +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 + write `model.TroDocument`. See `doc/Python_Scripts.md` §5a. What is **not** built out yet, relative to the original two-tool design: - the fuller **layout-JSON schema** of `doc/HundM/Json_Layout-Konzept.md` (`plc`, `controlUnits`, `sensors[]`, `conveyors[]`, `tros[]`, `loadingBooms[]`, `emptyCarrBuffers[]`, `routing`, `jamAreas[]`, `scanners[]`, `connections[]`, - `destinations[]`) as the intermediate format — `tro_extract.py` currently emits a - TRO-focused JSON, not this full schema; + `destinations[]`) as the intermediate format — `model.TroDocument` is a TRO-focused + JSON, not this full schema yet; - the **other per-controller blocks** (`FB_CallSensors`, `FC_Direction`/`FC_Call_Jams` as standalone files) — only `FB_Main` is generated today; - **type-based timing defaults** from `doc/TRO_Typen.md` and the `OVERRIDE_TIMING_JSON` CAD attribute (timings are emitted as `TODO(E-Planung): pruefen` placeholders for now); -- a **`--skip-json`** convenience path that would read DXF + CSV directly and emit SCL - without an intermediate JSON file. +- `tro_overrides.py` using the shared `model.Connection` / `model.OpenPoint` classes; +- `Component` classes per part type (Separator, Switch, Scanner, VarioDrive, …) on the TRO, + driving the per-part SCL emitters in `scl_skeleton.py`. Note the `create_skel.py` / `guess_fbtype()` generator sketched in `doc/HundM/suggestion.md` and `doc/HundM/Json_Layout-Konzept.md` §14.5 is a **separate concept** and is not implemented @@ -203,12 +212,12 @@ sps_skel/ domain-model documents listed above examples/ example files — empty for now lib/ Python source, importable via SKEL_LIB on PYTHONPATH — CLI tools and - libraries that derive material flow / TRO lists / CAD annotations from a - layout, read the TROs back into JSON, and emit an FB_Main SCL skeleton - (see doc/Python_Scripts.md) + libraries that derive material flow / TRO lists from a layout, write + them as TRO-JSON (model.py), emit an FB_Main SCL skeleton, and draw + the TROs into the CAD drawing (see doc/Python_Scripts.md) log/ gitignored results/ gitignored — CLI tool output (.dot/.svg/.md/.dxf/.scl/.json) - tests/ unit tests — empty for now + tests/ unit tests (pytest) — test_model.py for the TRO-JSON model ``` When extending the pipeline (the remaining pieces in "Roadmap" above), put new modules under diff --git a/bin/scl_skeleton.bat b/bin/scl_skeleton.bat index 8cff057..9174d29 100644 --- a/bin/scl_skeleton.bat +++ b/bin/scl_skeleton.bat @@ -3,14 +3,14 @@ REM ================================================================ REM SPS_SKEL - FB_Main-Geruest als SCL erzeugen REM ================================================================ REM Aufruf: -REM bin\tro_flow.bat --file mubea.csv -REM bin\tro_flow.bat --file mubea.csv --tosvg -REM bin\tro_flow.bat --file mubea.csv --tosvg --doc +REM bin\scl_skeleton.bat +REM bin\scl_skeleton.bat --json mubea_tro.json +REM bin\scl_skeleton.bat --json mubea_tro.json --out FB_Main.scl REM REM Die JSON-Datei wird in %SKEL_RESULTS% gesucht, die SCL-Datei wird REM ebenfalls dort abgelegt. REM -REM Schwesterprogramm: bin\material_flow.bat (Materialfluss der Objekte) +REM Vorstufe: bin\tro_flow.bat --file .csv --json (CSV -> TRO-JSON) REM ================================================================ setlocal diff --git a/bin/scl_skeleton.sh b/bin/scl_skeleton.sh index a8d6d38..eeb12f2 100644 --- a/bin/scl_skeleton.sh +++ b/bin/scl_skeleton.sh @@ -10,7 +10,7 @@ # Die JSON-Datei wird in $SKEL_RESULTS gesucht, die SCL-Datei wird # ebenfalls dort abgelegt. # -# Vorstufe: bin/tro_extract.sh (Zeichnung -> JSON) +# Vorstufe: bin/tro_flow.sh --file .csv --json (CSV -> TRO-JSON) # ================================================================ set -eu diff --git a/bin/tro_flow.bat b/bin/tro_flow.bat index 19043db..a428efc 100644 --- a/bin/tro_flow.bat +++ b/bin/tro_flow.bat @@ -6,6 +6,7 @@ REM Aufruf: REM bin\tro_flow.bat --file mubea.csv REM bin\tro_flow.bat --file mubea.csv --tosvg REM bin\tro_flow.bat --file mubea.csv --tosvg --doc +REM bin\tro_flow.bat --file mubea.csv --json (TRO-JSON fuer scl_skeleton) REM REM Die CSV-Datei wird in %SKEL_DATA% gesucht, DOT, SVG, Doku und REM Fehlerdatei werden in %SKEL_RESULTS% abgelegt. diff --git a/bin/tro_flow.sh b/bin/tro_flow.sh index 72e1037..810f8ba 100644 --- a/bin/tro_flow.sh +++ b/bin/tro_flow.sh @@ -6,6 +6,7 @@ # bash bin/tro_flow.sh --file mubea.csv # bash bin/tro_flow.sh --file mubea.csv --tosvg # bash bin/tro_flow.sh --file mubea.csv --tosvg --doc +# bash bin/tro_flow.sh --file mubea.csv --json (TRO-JSON fuer scl_skeleton) # # Die CSV-Datei wird in $SKEL_DATA gesucht, DOT, SVG, Doku und # Fehlerdatei werden in $SKEL_RESULTS abgelegt. diff --git a/doc/Commandlines.md b/doc/Commandlines.md index dfa9b43..9d7cbd3 100644 --- a/doc/Commandlines.md +++ b/doc/Commandlines.md @@ -28,15 +28,19 @@ activates `.venv` if present and runs `lib/.py` with all arguments passed through. Every tool accepts `--help`. ``` -CSV ─► material_flow ─► tro_flow ─► tro_annotate ─► [edit in BricsCAD] ─► tro_extract ─► scl_skeleton - (+ DXF) (DXF only) (JSON → SCL) +CSV ─► material_flow ─► tro_flow --json ─► _tro.json ─► scl_skeleton ─► _FB_Main.scl + │ (model.py) (validates) + └─► tro_annotate (+ DXF) ─► _annotated.dxf viewer only + +parked: tro_extract (DXF → JSON), see 2.4 ``` Shared conventions: - Input names (`--file`, `--dxf`) are a bare filename resolved against - `%SKEL_DATA%`, or a full path. `tro_extract`/`scl_skeleton` look up their - input in `%SKEL_RESULTS%` instead (it is the output of the previous step). + `%SKEL_DATA%`, or a full path. `scl_skeleton` (and the parked `tro_extract`) + look up their input in `%SKEL_RESULTS%` instead (it is the output of the + previous step). - All output goes to `%SKEL_RESULTS%`. - `--tosvg` needs Graphviz `dot` (`neato` with `--use-cords`) on `PATH`, or `GRAPHVIZ_DOT` pointing at `dot`. @@ -78,16 +82,20 @@ Input: CSV export. Details: [`Python_Scripts.md` §4](Python_Scripts.md#4-tro_fl | `--true-scale` | — | off | Only with `--use-cords`: draw to scale (default shortens empty stretches > 1500 mm). | | `--doc` | — | off | Write Markdown report `_tro_doc.md`. | | `--connect` | — | off | Write TRO topology as `_tro_connect.ini`. | +| `--json` | — | off | Write the TRO-JSON `_tro.json` (model `lib/model.py`) — input for `scl_skeleton`. | Exit codes: `0` ok · `1` input/CLI error · `2` SVG rendering failed · `3` TRO validation errors (`_tro_errors.md` written). ```bat bin\tro_flow.bat --file mubea.csv --tosvg --use-cords --doc +bin\tro_flow.bat --file mubea.csv --json ``` ### 2.3 `tro_annotate` — burn the TROs into a copy of the drawing +Viewer only: shows the same TROs as `tro_flow` in the drawing for a visual +check; the drawing is not read back as data. Input: CSV export + original BricsCAD DXF. The original is never modified. Details: [`Python_Scripts.md` §5](Python_Scripts.md#5-tro_annotatepy--burn-the-tros-into-a-copy-of-the-cad-drawing). @@ -117,6 +125,10 @@ bin\tro_annotate.bat --emit-lisp ### 2.4 `tro_extract` — read the TROs back out of the drawing into JSON +**Parked** — kept, but not part of the main path. Its JSON (schema +`sps_skel-tro-extract-v1`) is rejected by `scl_skeleton`; use +`tro_flow --json` instead. + Input: the (possibly hand-edited) annotated DXF only, never the CSV. Details: [`Python_Scripts.md` §5a](Python_Scripts.md#5a-tro_extractpy--read-the-tros-back-out-of-the-drawing-into-json). @@ -133,8 +145,9 @@ bin\tro_extract.bat --dxf mubea_annotated.dxf ### 2.5 `scl_skeleton` — generate the `FB_Main` SCL skeleton -Input: TRO-JSON from `tro_extract`. Output is a skeleton with -`TODO(E-Planung)` gaps, not a compilable program. +Input: TRO-JSON from `tro_flow --json`, validated against `lib/model.py` — a +wrong or renamed key stops the run (exit `1`) instead of producing SCL. Output +is a skeleton with `TODO(E-Planung)` gaps, not a compilable program. Details: [`Python_Scripts.md` §5b](Python_Scripts.md#5b-scl_skeletonpy--generate-the-fb_main-skeleton-as-scl). | Switch | Argument | Default | Effect | @@ -143,11 +156,11 @@ Details: [`Python_Scripts.md` §5b](Python_Scripts.md#5b-scl_skeletonpy--generat | `--out` | `NAME` | `_FB_Main.scl` | Output SCL filename. | | `--start` | `TRO` | — | TRO where the REGION order begins. Only needed while the flow is a closed loop without a loading/unloading station (`SYSn`). | -Exit codes: `0` ok · `1` input/CLI error. +Exit codes: `0` ok · `1` input/CLI error or JSON does not match the model. ```bat -bin\scl_skeleton.bat --json mubea_annotated_tro.json -bin\scl_skeleton.bat --json mubea_annotated_tro.json --start TRO01 +bin\scl_skeleton.bat --json mubea_tro.json +bin\scl_skeleton.bat --json mubea_tro.json --start TRO01 ``` ## 3. End-to-end example @@ -155,18 +168,19 @@ bin\scl_skeleton.bat --json mubea_annotated_tro.json --start TRO01 ```bat bin\activate_venv.bat bin\material_flow.bat --file mubea.csv --tosvg --doc -bin\tro_flow.bat --file mubea.csv --tosvg --use-cords --doc +bin\tro_flow.bat --file mubea.csv --tosvg --use-cords --doc --json +bin\scl_skeleton.bat --json mubea_tro.json +REM optional, for a visual check in BricsCAD: bin\tro_annotate.bat --file mubea.csv --dxf Mubea.dxf --flow --legend -REM ... optionally review / edit results\mubea_annotated.dxf in BricsCAD ... -bin\tro_extract.bat --dxf mubea_annotated.dxf -bin\scl_skeleton.bat --json mubea_annotated_tro.json ``` +Tests: `python -m pytest tests/` (with `.venv` active). + ## 4. Documentation tooling (not part of the pipeline) | Command | Purpose | |---|---| | `py doc/TRO_Katalog/tro_graphs/gen_tro_graphs.py` | Regenerates the per-type SVGs, the edge legend and the UH01–UH05 topology graphs for `TRO_Katalog.md` from `doc/TRO_Katalog/connect.ini`. No switches, no `bin/` wrapper. Needs Graphviz `dot`. | -The libraries `lib/tro_catalog.py`, `lib/dxf_registration.py` and -`lib/tro_overrides.py` have no command line of their own. +The libraries `lib/tro_catalog.py`, `lib/model.py`, `lib/dxf_registration.py` +and `lib/tro_overrides.py` have no command line of their own. diff --git a/doc/Python_Scripts.md b/doc/Python_Scripts.md index 1cd05c1..ef24804 100644 --- a/doc/Python_Scripts.md +++ b/doc/Python_Scripts.md @@ -1,13 +1,14 @@ # Python Scripts — CLI Tooling Overview -**As of:** 2026-09-17 +**As of:** 2026-10-09 -`lib/` currently holds nine Python modules (five of them runnable CLI tools, -four supporting libraries), plus one documentation-generator script that lives +`lib/` currently holds ten Python modules (five of them runnable CLI tools, +five supporting libraries), plus one documentation-generator script that lives under `doc/TRO_Katalog/tro_graphs/`. Together the CLI tools turn a CSV export of the mechanical layout (ILS 2.0) into a material-flow graph, a derived TRO -list/diagram, an annotated copy of the BricsCAD drawing, a TRO-JSON read back -out of that drawing, and an `FB_Main` SCL skeleton generated from that JSON. +list/diagram, a TRO-JSON (shared model `lib/model.py`), an `FB_Main` SCL +skeleton generated from that JSON, and an annotated copy of the BricsCAD +drawing for visual checking. This document is the practical "what does each script do and which switches does it take" reference. For the underlying domain model (TRO types, JSON @@ -26,18 +27,25 @@ material_flow.py ──► _material_flow.dot / .svg / .md ▼ tro_flow.py ──► _tro_flow.dot / .svg, _tro_doc.md, _tro_errors.md │ (derives TROs, using the type catalog in tro_catalog.py) - ▼ -tro_annotate.py ──► _annotated.dxf, _registration.json - │ (also needs the BricsCAD DXF; uses dxf_registration.py for the CSV↔DXF - │ coordinate offset and tro_catalog.py for marker shape/color per TRO type) - ▼ (drawing may be hand-edited in BricsCAD before the next step) -tro_extract.py ──► _tro.json - │ (reads only the drawing; resolves plant coords via the registration) - ▼ -scl_skeleton.py ──► _FB_Main.scl - (one REGION per TRO; E-planning values left as TODO(E-Planung) gaps) + │ + ├── --json ──► _tro.json (model.TroDocument — MAIN PATH) + │ │ + │ ▼ + │ scl_skeleton.py ──► _FB_Main.scl + │ (validates the JSON against model.py; one REGION per TRO; + │ E-planning values left as TODO(E-Planung) gaps) + │ + └── same derivation ──► tro_annotate.py ──► _annotated.dxf (VIEWER ONLY) + (also needs the BricsCAD DXF; uses dxf_registration.py for the CSV↔DXF + coordinate offset and tro_catalog.py for marker shape/color per TRO type) + +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. + - `material_flow.py` models the material flow of the **mechanical** objects only (Gefällestrecke, Strecke, Kreisel). - `tro_flow.py` is its sister program: it derives the **control objects** @@ -51,6 +59,10 @@ Supporting libraries (no CLI of their own, imported by the tools above): - **`lib/tro_catalog.py`** — the single source of truth for TRO types: short name, Siemens FB block, sub-components and counts, color group, CAD marker shape. +- **`lib/model.py`** — the shared pydantic model of TRO *instances* and of the + TRO-JSON (`TroBase`, `DerivedTro`, `Connection`, `OpenPoint`, `Analysis`, + `TroDocument`). Written by `tro_flow.py --json`, read and validated by + `scl_skeleton.py`. See §6b. - **`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. @@ -77,7 +89,8 @@ Each CLI tool is invoked through its `bin/.bat` (Windows) / `bin/.sh ```bat bin\material_flow.bat --file mubea.csv --tosvg --use-cords --doc -bin\tro_flow.bat --file mubea.csv --tosvg --use-cords --doc +bin\tro_flow.bat --file mubea.csv --tosvg --use-cords --doc --json +bin\scl_skeleton.bat --json mubea_tro.json bin\tro_annotate.bat --file mubea.csv --dxf 500573_60_1.dxf --flow --fb --legend ``` @@ -281,6 +294,7 @@ colors all come from `lib/tro_catalog.py` — nothing is hard-coded here. | `--doc` | — | off | Also write a Markdown report (`_tro_doc.md`) with the TRO list, connections, findings and the full type catalog. | | `--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-v2`, model `TroDocument` in `lib/model.py`): TROs, system boundaries (`SYSn`), connections with their kind, findings. Input for `scl_skeleton.py`. | TROs, types and connections always come from the layout alone; `cfg/tro_overrides.ini` is not read. Errors in the derivation are fixed in @@ -288,16 +302,21 @@ TROs, types and connections always come from the layout alone; **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_errors.md` (only -if TROs are unconnected or of an unknown type). +`--tosvg`), `_tro_doc.md` (with `--doc`), `_tro.json` (with +`--json`), `_tro_errors.md` (only if TROs are unconnected or of an +unknown type). **Exit codes:** `0` ok · `1` input/CLI error · `2` SVG rendering failed (Graphviz) · `3` TRO validation errors found (error file written). ## 5. `tro_annotate.py` — burn the TROs into a copy of the CAD drawing +**Viewer only.** The drawing shows the result for a visual check in BricsCAD; +it is no longer a data source for the SCL (that is `tro_flow.py --json`). + Takes the CSV export, the original BricsCAD drawing (DXF) and the TROs -derived the same way as `tro_flow.py`, and writes a **copy** of the drawing +derived with exactly the same call as `tro_flow.py` (including the separator +cross-check from `_raw.json`), and writes a **copy** of the drawing with: - one colored marker block per TRO, each type with its own shape (circle, @@ -353,6 +372,13 @@ could not be proven (or its residual error exceeds 600 mm) — resolve with ## 5a. `tro_extract.py` — read the TROs back out of the drawing into JSON +> **Parked since 2026-10-09.** The TRO-JSON for `scl_skeleton.py` now comes +> from `tro_flow.py --json`. `tro_extract.py` is kept but not extended; its +> output (schema `sps_skel-tro-extract-v1`) is **not** accepted by +> `scl_skeleton.py` any more, which says so and asks for `tro_flow --json`. If +> hand edits in the drawing are ever needed again, it should write the same +> `model.TroDocument` format. + The counter-direction to `tro_annotate.py`: ``` @@ -394,12 +420,18 @@ Last step of the chain: ``` material_flow.py objects -> material flow -tro_flow.py flow -> TROs -tro_annotate.py TROs -> markers in the drawing -tro_extract.py drawing -> JSON +tro_flow.py flow -> TROs -> _tro.json (--json) scl_skeleton.py JSON -> FB_Main.scl ``` +The JSON is read through `model.TroDocument` and **validated**: an unknown or +renamed key, or `items`/`fbType` that do not match the TRO type in +`tro_catalog.py`, stops the run with exit code `1` and the validation error — +no SCL is written. A file with a different schema (e.g. the old +`tro_extract.py` output) gets a short message asking for `tro_flow --json`. +Part counts (`items`) and the FB type are taken from the catalog via the TRO +type. + Produces a **skeleton, not a running program**: one `REGION` per TRO with the instance call and every parameter line in the right order, following the real call sites in `doc/TRO_Katalog/scl_templates/*.scl`. @@ -426,14 +458,16 @@ only auto-filled for a single-switch TRO; where several switches share the exits, how they divide them is a planning decision, so it stays a gap rather than a guess. -Switches: `--json NAME` (default `export_tro.json`), `--out NAME` (default +Switches: `--json NAME` (default `export_tro.json`, looked up in +`%SKEL_RESULTS%`; written by `tro_flow.py --json`), `--out NAME` (default `_FB_Main.scl`), `--start TRO`. The last one matters while the plant has no loading/unloading station: the flow is then a closed loop with no entry, so the order has to be broken somewhere arbitrary. The chosen start and the reason are recorded in the file header. **Stations (`SYSn`).** Loading/unloading stations from `tro_flow.py` appear in -the JSON as `SYSn` neighbours, not as TROs. They get no REGION and no instance. +the JSON under `boundaries` and as `SYSn` neighbours of the TROs, not in +`tros`. They get no REGION and no instance. A TRO fed by a `SYSn` is the start of the order ("Materialeingang ueber SYSn"), so `--start` is no longer needed once the export contains a station. In the jam-area comments they are marked "(Station, kein TRO)", and an exit station @@ -461,7 +495,40 @@ verified against `data/*.scl` where the source is vendored locally, and against Everything is a Pydantic v2 model (`TroDefinition`/`TroStyle`), so it validates on every assignment and comes with `model_dump()`/`model_validate()` for the planned JSON layout model for -free. Consumed by `tro_flow.py` and `tro_annotate.py`; not run directly. +free. Consumed by `model.py`, `tro_flow.py`, `tro_annotate.py` and +`scl_skeleton.py`; not run directly. + +## 6b. `model.py` — shared model of TRO instances and the TRO-JSON (library, no CLI) + +`tro_catalog.py` says what a TRO **type** is; `model.py` says what one concrete +TRO in a plant, and the whole TRO-JSON, look like. It imports only the catalog. + +``` +tro_catalog.py type level TroDefinition (items, FB block, color, symbol) + ▲ get_tro() +model.py instance level TroBase ─► DerivedTro, Connection, OpenPoint, + Analysis, TroDocument + ▲ +tro_flow.py (writes) · scl_skeleton.py (reads) · tro_annotate.py (draws) +``` + +| Class | What it is | +|---|---| +| `TroBase` | Core of every TRO: `id`, `type`, `confidence`, `separators`, `predecessors`, `successors`, `position` (`{x,y,z}` in JSON). `definition()`, `fb_block` (JSON `fbType`) and `items` are **read from the catalog** on every access, never stored — so they follow a type change and cannot go stale. | +| `DerivedTro` | `TroBase` + what `tro_flow.py` knows from the derivation: `label`, `hosts`, `scanners`, `reason`, `colocated`, `positionFrom`, `positionPoints`. Also used for the system boundaries `SYSn`. | +| `Connection` | Directed edge `from` → `to` with its `kind` (transfer / umlauf / weiche / unbestimmt / normal / bahn). | +| `OpenPoint` | A reported spot that is not a TRO yet (empty in the JSON for now; meant to replace the class of the same name in `tro_overrides.py`). | +| `Analysis` | Working result of `tro_flow.analyse_tros` (moved here from `tro_flow.py`). | +| `TroDocument` | The TRO-JSON: `schema` (`sps_skel-tro-v2`), `source`, `created`, `note`, `tros`, `boundaries`, `connections`, `openPoints`, `findings`. `from_analysis()`, `to_json()`, `from_json()`. | + +All models validate on creation and assignment and reject unknown keys +(`extra="forbid"`). JSON keys keep the camelCase names of the TRO-JSON (`id`, +`type`, `fbType`, `openPoints` …); Python code uses the field names +(`tro_id`, `type_name`, `fb_block`, `open_points`) — both are accepted on input. + +Tests: `tests/test_model.py` (round trip, catalog-backed fields, rejection of +wrong keys/items, SCL rendering from a document) — run with +`python -m pytest tests/`. ## 6a. `tro_overrides.py` — manual corrections to the derived TROs (library, no CLI) diff --git a/lib/model.py b/lib/model.py new file mode 100644 index 0000000..40827f7 --- /dev/null +++ b/lib/model.py @@ -0,0 +1,306 @@ +# -*- coding: utf-8 -*- +""" +model.py - Gemeinsames Datenmodell der TRO-Instanzen und der TRO-JSON. + +Schichten (wer darf wen importieren): + + tro_catalog.py Typ-Ebene: was ein TRO-*Typ* ist (Bauteile, FB, Farbe, Symbol) + model.py Instanz-Ebene: ein konkreter TRO, die ganze TRO-JSON <- hier + importiert nur tro_catalog + tro_flow.py Ableitung: CSV-Graph -> Analysis -> TroDocument + scl_skeleton.py Ausgabe: TroDocument -> FB_Main.scl + tro_annotate.py Ausgabe: Analysis -> annotierte Zeichnung (nur Ansicht) + +Klassen: + + TroBase was jeder TRO hat, egal woher er stammt: ID, Typ, Separatoren, + Vorgaenger/Nachfolger, Position. FB-Baustein und Bauteile + werden aus dem Katalog gelesen, nie gespeichert. + DerivedTro TroBase + was tro_flow bei der Ableitung zusaetzlich weiss + (Hosts, Scanner, Begruendung, Positionsquelle ...) + Connection eine gerichtete Verbindung zwischen zwei TROs/Stationen + OpenPoint eine gemeldete Stelle, die noch kein TRO ist + Analysis Arbeitsergebnis von tro_flow.analyse_tros + TroDocument die TRO-JSON (_tro.json): von tro_flow --json + geschrieben, von scl_skeleton gelesen und geprueft + +In der JSON stehen die Schluessel wie bisher in der TRO-JSON (`id`, `type`, +`fbType`, `openPoints`, ...); im Python-Code heissen die Felder weiter +`tro_id`/`type_name` (Aliase, `populate_by_name`). +""" + +from __future__ import annotations + +from datetime import datetime +from typing import Any + +from pydantic import ( + BaseModel, + ConfigDict, + Field, + computed_field, + field_serializer, + field_validator, + model_validator, +) + +from tro_catalog import TroDefinition, get_tro + +__all__ = [ + "SCHEMA", + "TroBase", + "DerivedTro", + "Connection", + "OpenPoint", + "Analysis", + "TroDocument", +] + +# Kennung der TRO-JSON. v1 war die Ausgabe von tro_extract (aus der Zeichnung), +# v2 die von tro_flow --json (aus dem CSV, ueber dieses Modell). +SCHEMA = "sps_skel-tro-v2" + +# Gemeinsame Einstellungen: Pruefung bei Erzeugung und Zuweisung, unbekannte +# Schluessel sind ein Fehler (ein umbenannter JSON-Schluessel faellt auf statt +# still auf einen Standardwert zurueckzufallen), Aliase und Feldnamen gelten +# beide. +_CONFIG = ConfigDict( + validate_assignment=True, + extra="forbid", + populate_by_name=True, +) + + +class TroBase(BaseModel): + """ + Kern eines TRO (Instanz in einer Anlage), unabhaengig von seiner Quelle. + + Gegenstueck zu :class:`tro_catalog.TroDefinition`: die Definition beschreibt + den *Typ* (FB-Baustein, Bauteile), ein TRO die konkrete Stelle im Layout. + Verknuepft ueber ``type_name``, siehe :meth:`definition`. + + FB-Baustein und Bauteile haengen nur am Typ. Sie werden nicht gespeichert, + sondern bei jedem Zugriff aus dem Katalog gelesen (``fb_block``, ``items``) + und so auch in die JSON geschrieben. Steht beim Einlesen einer JSON ein + Wert drin, der nicht zum Katalog passt, ist das ein Fehler - den Typ + aendern, nicht die Bauteile. + + Listenfelder werden beim Pruefen kopiert; ``tro.hosts.append(...)`` aendert + also nur diesen TRO. + """ + + model_config = _CONFIG + + tro_id: str = Field(alias="id") + type_name: str = Field(alias="type") + confidence: str = "" + separators: list[str] = Field(default_factory=list) + predecessors: list[str] = Field(default_factory=list) + successors: list[str] = Field(default_factory=list) + # Anlagenkoordinate (mm); in der JSON als {"x": .., "y": .., "z": ..} + position: tuple[float, float, float] | None = None + + # -- aus dem Katalog ----------------------------------------------------- + + def definition(self) -> TroDefinition | None: + """Typdefinition aus tro_catalog (None wenn Typ unbekannt).""" + return get_tro(self.type_name) + + @computed_field(alias="fbType") # type: ignore[prop-decorator] + @property + def fb_block(self) -> str: + """FB-Baustein laut Katalog, z. B. 'FB_ILS_MTRO_1Sep' ("" wenn Typ unbekannt).""" + definition = self.definition() + return definition.get_fb_block() if definition else "" + + @computed_field # type: ignore[prop-decorator] + @property + def items(self) -> dict[str, int]: + """Bauteile laut Katalog, z. B. {'Separator': 1} ({} wenn Typ unbekannt).""" + definition = self.definition() + return definition.get_items() if definition else {} + + @model_validator(mode="before") + @classmethod + def _check_catalog_fields(cls, data: Any) -> Any: + """`items`/`fbType` aus einer JSON gegen den Katalog pruefen und entfernen.""" + if not isinstance(data, dict) or not ({"items", "fbType", "fb_block"} & data.keys()): + return data + data = dict(data) + items = data.pop("items", None) + fb_block = data.pop("fbType", data.pop("fb_block", None)) + type_name = data.get("type", data.get("type_name")) + definition = get_tro(type_name) if isinstance(type_name, str) else None + expected_items = definition.get_items() if definition else {} + expected_fb = definition.get_fb_block() if definition else "" + if items is not None and dict(items) != expected_items: + raise ValueError( + f"items {items} passt nicht zum Typ '{type_name}' laut tro_catalog " + f"({expected_items}) - den Typ aendern, nicht die Bauteile" + ) + if fb_block is not None and (fb_block or "") != expected_fb: + raise ValueError( + f"fbType '{fb_block}' passt nicht zum Typ '{type_name}' laut " + f"tro_catalog ('{expected_fb}')" + ) + return data + + # -- Position ------------------------------------------------------------ + + @field_validator("position", mode="before") + @classmethod + def _position_from_json(cls, value: Any) -> Any: + if isinstance(value, dict): + return (value["x"], value["y"], value.get("z", 0.0)) + return value + + @field_serializer("position", when_used="json-unless-none") + def _position_to_json(self, value: tuple[float, float, float]) -> dict[str, float]: + return {"x": round(value[0], 2), "y": round(value[1], 2), "z": round(value[2], 2)} + + @property + def x(self) -> float | None: + return self.position[0] if self.position else None + + @property + def y(self) -> float | None: + return self.position[1] if self.position else None + + @property + def z(self) -> float | None: + return self.position[2] if self.position else None + + def describe_position(self) -> str: + """Koordinate als Text, z. B. '1030, 4700'.""" + if not self.position: + return "-" + return f"{self.position[0]:.0f}, {self.position[1]:.0f}" + + +class DerivedTro(TroBase): + """ + Ein von tro_flow aus dem CSV abgeleiteter TRO (oder eine Systemgrenze). + + Zusaetzlich zum Kern: aus welchen Bauteilen er gebildet wurde, warum er + diesen Typ hat, und woher seine Position stammt. + """ + + label: str + hosts: list[str] + scanners: list[str] + reason: str + colocated: list[str] = Field(default_factory=list) + # Mittelpunkt der Bauteile, aus denen dieser TRO gebildet wurde + position_from: str = Field(default="", alias="positionFrom") + position_points: int = Field(default=0, alias="positionPoints") + + @property + def connected(self) -> bool: + return bool(self.successors or self.predecessors or self.colocated) + + +class Connection(BaseModel): + """Gerichtete Verbindung zwischen zwei TROs/Stationen (JSON: from/to/kind).""" + + model_config = _CONFIG + + src: str = Field(alias="from") + dst: str = Field(alias="to") + # transfer / umlauf / weiche / unbestimmt / normal (siehe tro_flow.link_tros) + kind: str = "" + + +class OpenPoint(BaseModel): + """Eine im Review gemeldete Stelle, die noch kein eigener TRO ist.""" + + model_config = _CONFIG + + name: str + x: float + y: float + z: float | None = None + expect: str = "" + belongs_to: str = Field(default="", alias="belongsTo") + note: str = "" + + def describe_position(self) -> str: + return f"{self.x:.0f}, {self.y:.0f}" + + +class Analysis(BaseModel): + """Ergebnis von tro_flow.analyse_tros: TROs, Kanten, Befunde.""" + + model_config = _CONFIG + + tros: list[DerivedTro] = Field(default_factory=list) + edges: list[tuple[str, str]] = Field(default_factory=list) + # (von, nach) -> Kantenart (transfer / umlauf / weiche / unbestimmt / normal) + edge_kinds: dict[tuple[str, str], str] = Field(default_factory=dict) + isolated: list[DerivedTro] = Field(default_factory=list) + unknown_types: list[str] = Field(default_factory=list) + findings: list[str] = Field(default_factory=list) + # Wie Host/Fanout ermittelt wurden (Doku/Bericht) + method: str = "Materialfluss-Graph (Zuordnung/Nachbarn)" + # Systemgrenzen (BTMT Be-/Entladestationen) - siehe derive_boundaries; kein + # TRO, nur damit der tatsaechliche Materialein-/ausgang sichtbar wird. + boundaries: list[DerivedTro] = Field(default_factory=list) + + def by_id(self, tro_id: str) -> DerivedTro | None: + for tro in self.tros + self.boundaries: + if tro.tro_id == tro_id: + return tro + return None + + +class TroDocument(BaseModel): + """ + Die TRO-JSON einer Anlage (``_tro.json``). + + Geschrieben von ``tro_flow --json`` (:meth:`from_analysis`, :meth:`to_json`), + gelesen von scl_skeleton (:meth:`from_json`). Weil das Modell beide Seiten + definiert, faellt ein umbenannter oder falscher Schluessel beim Einlesen + als Fehler auf. + """ + + model_config = _CONFIG + + schema_id: str = Field(default=SCHEMA, alias="schema") + source: str + created: str = Field(default_factory=lambda: datetime.now().strftime("%Y-%m-%d %H:%M")) + note: str = "" + tros: list[DerivedTro] = Field(default_factory=list) + # Systemgrenzen (SYSn): Materialein-/ausgang, kein TRO - keine SCL + boundaries: list[DerivedTro] = Field(default_factory=list) + connections: list[Connection] = Field(default_factory=list) + open_points: list[OpenPoint] = Field(default_factory=list, alias="openPoints") + findings: list[str] = Field(default_factory=list) + + @classmethod + def from_analysis(cls, analysis: Analysis, source: str, note: str = "") -> TroDocument: + """Dokument aus dem Ergebnis von tro_flow.analyse_tros bilden.""" + return cls( + source=source, + note=note, + tros=analysis.tros, + boundaries=analysis.boundaries, + connections=[ + Connection(src=src, dst=dst, kind=analysis.edge_kinds.get((src, dst), "")) + for src, dst in analysis.edges + ], + findings=analysis.findings, + ) + + @classmethod + def from_json(cls, text: str) -> TroDocument: + """JSON-Text einlesen und pruefen (pydantic.ValidationError bei Fehlern).""" + return cls.model_validate_json(text) + + def to_json(self) -> str: + """Als JSON-Text mit den Schluesseln der TRO-JSON (id, type, fbType ...).""" + return self.model_dump_json(by_alias=True, indent=2) + "\n" + + def by_id(self, tro_id: str) -> DerivedTro | None: + for tro in self.tros + self.boundaries: + if tro.tro_id == tro_id: + return tro + return None diff --git a/lib/scl_skeleton.py b/lib/scl_skeleton.py index 1228846..c3e7258 100644 --- a/lib/scl_skeleton.py +++ b/lib/scl_skeleton.py @@ -7,11 +7,14 @@ scl_skeleton.py - FB_Main-Geruest als SCL aus der TRO-Liste erzeugen. Letzter Schritt der Kette: material_flow.py Objekte -> Materialfluss - tro_flow.py Fluss -> TROs - tro_annotate.py TROs -> Marker in der Zeichnung - tro_extract.py Zeichnung -> JSON + tro_flow.py Fluss -> TROs -> _tro.json (--json) scl_skeleton.py JSON -> FB_Main.scl <- hier +Die TRO-JSON wird ueber das gemeinsame Modell lib/model.py (TroDocument) +eingelesen und dabei geprueft: ein falscher oder umbenannter Schluessel bricht +mit einer Fehlermeldung ab, statt still einen Standardwert zu nehmen. Bauteile +(items) und FB-Typ kommen aus tro_catalog, ueber den TRO-Typ. + Erzeugt wird ein **Geruest**, kein fertiges Programm: je TRO eine REGION mit Instanzaufruf und allen Parameterzeilen in der richtigen Reihenfolge, so wie es die realen Aufrufstellen in doc/TRO_Katalog/scl_templates/*.scl zeigen. @@ -41,7 +44,11 @@ from collections import deque from datetime import datetime from pathlib import Path +from pydantic import ValidationError + from material_flow import env_dir +from model import SCHEMA, TroBase, TroDocument +from tro_catalog import TroItem TODO = "TODO(E-Planung)" @@ -55,11 +62,12 @@ DEFAULT_SEP_TIMES = ( ) -def load_tros(path: Path) -> dict: - return json.loads(path.read_text(encoding="utf-8")) +def load_document(path: Path) -> TroDocument: + """TRO-JSON einlesen und gegen das Modell pruefen.""" + return TroDocument.from_json(path.read_text(encoding="utf-8")) -def flow_order(tros: list[dict], start_id: str = "") -> tuple[list[dict], str]: +def flow_order(tros: list[TroBase], start_id: str = "") -> tuple[list[TroBase], str]: """TROs in Fliessrichtung sortieren. Startpunkt ist ein TRO ohne Vorgaenger - also die Stelle, an der Material @@ -70,14 +78,14 @@ def flow_order(tros: list[dict], start_id: str = "") -> tuple[list[dict], str]: --start ueberschreiben). Nicht erreichte TROs haengt die Funktion hinten an, damit nie einer verlorengeht. """ - nach_id = {t["id"]: t for t in tros} - ohne_vorgaenger = sorted(t["id"] for t in tros if not t.get("predecessors")) + nach_id = {t.tro_id: t for t in tros} + ohne_vorgaenger = sorted(t.tro_id for t in tros if not t.predecessors) # Materialeingang ueber eine Station (SYSn von tro_flow.py): der TRO, den # sie speist, ist der natuerliche Anfang - auch wenn er zusaetzlich einen # TRO-Vorgaenger hat (Kreisel-Umlauf). eingaenge = sorted({ - t["id"] for t in tros - for vor in t.get("predecessors", []) if vor not in nach_id + t.tro_id for t in tros + for vor in t.predecessors if vor not in nach_id }) if start_id and start_id in nach_id: @@ -86,7 +94,7 @@ def flow_order(tros: list[dict], start_id: str = "") -> tuple[list[dict], str]: start, grund = ohne_vorgaenger, "TRO ohne Vorgaenger (Materialeingang)" elif eingaenge: stationen = sorted({ - vor for i in eingaenge for vor in nach_id[i]["predecessors"] + vor for i in eingaenge for vor in nach_id[i].predecessors if vor not in nach_id }) start = eingaenge @@ -103,7 +111,7 @@ def flow_order(tros: list[dict], start_id: str = "") -> tuple[list[dict], str]: if tro_id in gesehen: continue gesehen.append(tro_id) - for nxt in nach_id.get(tro_id, {}).get("successors", []): + for nxt in nach_id[tro_id].successors if tro_id in nach_id else []: # SYSn (Materialausgang) ist kein TRO - keine REGION, keine Instanz. if nxt in nach_id and nxt not in gesehen: queue.append(nxt) @@ -113,7 +121,7 @@ def flow_order(tros: list[dict], start_id: str = "") -> tuple[list[dict], str]: return [nach_id[i] for i in gesehen], grund -def reachable_via(start: str, first: str, tros: dict[str, dict]) -> list[str]: +def reachable_via(start: str, first: str, tros: dict[str, TroBase]) -> list[str]: """Welche TROs sind ueber diesen Abgang erreichbar? Das ist die Zielliste eines Weichenabgangs: von `first` aus weiterlaufen, @@ -127,25 +135,26 @@ def reachable_via(start: str, first: str, tros: dict[str, dict]) -> list[str]: if node in gesehen or node == start: continue gesehen.add(node) - queue.extend(tros.get(node, {}).get("successors", [])) + if node in tros: + queue.extend(tros[node].successors) return sorted(gesehen) -def nachbar(node: str, alle: dict[str, dict]) -> str: +def nachbar(node: str, alle: dict[str, TroBase]) -> str: """Nachbar fuer Kommentare; SYSn als Station kenntlich machen.""" return node if node in alle else f"{node} (Station, kein TRO)" -def region_title(tro: dict) -> str: - seps = ", ".join(tro.get("separators", [])) +def region_title(tro: TroBase) -> str: + seps = ", ".join(tro.separators) teile = f" Sep {seps}" if seps else "" - return f"{tro['id']} ({tro['type']}){teile}" + return f"{tro.tro_id} ({tro.type_name}){teile}" -def emit_separators(add, tro: dict, anzahl: int, hat_scanner: bool) -> None: - seps = tro.get("separators", []) +def emit_separators(add, tro: TroBase, anzahl: int, hat_scanner: bool) -> None: + seps = tro.separators for i in range(1, anzahl + 1): - pfad = f"#{tro['id']}.stInSeparator{i}" + pfad = f"#{tro.tro_id}.stInSeparator{i}" layout = seps[i - 1] if i <= len(seps) else "?" add(f" // Separator {i} = Bauteil {layout} aus dem Layout") add(f' {pfad}.Settings.nSeparatorNo := "cSep_____"; // {TODO}') @@ -162,10 +171,10 @@ def emit_separators(add, tro: dict, anzahl: int, hat_scanner: bool) -> None: add("") -def emit_switches(add, tro: dict, anzahl: int, alle: dict[str, dict]) -> None: - abgaenge = tro.get("successors", []) +def emit_switches(add, tro: TroBase, anzahl: int, alle: dict[str, TroBase]) -> None: + abgaenge = tro.successors for j in range(1, anzahl + 1): - pfad = f"#{tro['id']}.stInSwitch{j}" + pfad = f"#{tro.tro_id}.stInSwitch{j}" add(f' {pfad}.Settings.nSwitchNo := "cSwi_____"; // {TODO}') add(f" {pfad}.Settings.xPluggedExSen3 := FALSE;") add(f" {pfad}.Settings.xPluggedExSen4 := FALSE;") @@ -185,13 +194,13 @@ def emit_switches(add, tro: dict, anzahl: int, alle: dict[str, dict]) -> None: add(" // Zielliste je Abgang (aus dem TRO-Graphen berechnet) -") add(" // Grundlage fuer die Zuordnung Ziel -> Richtung in FC_Direction:") for abgang in abgaenge: - ziele = reachable_via(tro["id"], abgang, alle) + ziele = reachable_via(tro.tro_id, abgang, alle) add(f" // ueber {abgang}: {', '.join(ziele) if ziele else '-'}") add("") -def emit_vario(add, tro: dict) -> None: - pfad = f"#{tro['id']}.stInVario" +def emit_vario(add, tro: TroBase) -> None: + pfad = f"#{tro.tro_id}.stInVario" add(f" {pfad}.Settings.xContinousEmptying := FALSE;") add(f" {pfad}.Settings.tWaitForJog := T#300ms; // {TODO}: pruefen") add(f" {pfad}.Settings.tConvEmpty := T#45000ms; // {TODO}: pruefen") @@ -208,10 +217,10 @@ def emit_vario(add, tro: dict) -> None: add("") -def emit_barcode(add, tro: dict, anzahl: int) -> None: +def emit_barcode(add, tro: TroBase, anzahl: int) -> None: for i in range(1, anzahl + 1): suffix = "" if anzahl == 1 else str(i) - pfad = f"#{tro['id']}.stInBarcodeReader{suffix}" + pfad = f"#{tro.tro_id}.stInBarcodeReader{suffix}" add(f" {pfad}.Settings.xActivateCheckASCI := TRUE;") add(f" {pfad}.Settings.tTimeout := T#3000ms;") add(f' // {TODO}: Cognex-Interface verdrahten ' @@ -219,21 +228,21 @@ def emit_barcode(add, tro: dict, anzahl: int) -> None: add("") -def emit_call(add, tro: dict, items: dict[str, int], alle: dict[str, dict]) -> None: +def emit_call(add, tro: TroBase, items: dict[str, int], alle: dict[str, TroBase]) -> None: """Aufrufstelle schreiben. Die Zeilen werden als (Code, Kommentar) gefuehrt und das Komma erst beim Zusammensetzen gesetzt - sonst landet es hinter dem "//" und ist Teil des Kommentars, womit die Parameterliste nicht mehr uebersetzt. """ - tro_id = tro["id"] + tro_id = tro.tro_id zeilen: list[tuple[str, str]] = [ ('nInMainTroNo := "cMainTro____"', TODO), ('xInSftyOk := "DB_InterfaceSafety".stFromSafety.EStopMain.Q', TODO), ("xInAllRdyToStart := TRUE", ""), (f'xInRelease := "DB_TRO_TEST".{tro_id}', ""), ] - if items.get("VarioDrive"): + if items.get(TroItem.VARIO_DRIVE): zeilen.append(('xInMotorProtection := "FC____"', TODO)) zeilen.append(("xInAutomatic := #t_automatik", "")) zeilen.append(("stInOutMachineState := #stInOutControlUnitCabinet", "")) @@ -242,20 +251,20 @@ def emit_call(add, tro: dict, items: dict[str, int], alle: dict[str, dict]) -> N # Staubereiche: je ankommender/abgehender Verbindung einer. Welche # DB-Struktur dahintersteht, ist eine Planungsentscheidung - hier steht # nur, WIE VIELE es sein muessen und zu welchem Nachbarn sie gehoeren. - for n, vor in enumerate(tro.get("predecessors", []), start=1): + for n, vor in enumerate(tro.predecessors, start=1): zeilen.append((f'arInOutJamEntr{n} := "DB_JamArea____".stJam____.arCarrier', f"{TODO} - Staubereich von {nachbar(vor, alle)}")) zeilen.append((f'stInOutJamEntr{n} := "DB_JamArea____".stJam____.stData', "")) - for n, nach in enumerate(tro.get("successors", []), start=2): + for n, nach in enumerate(tro.successors, start=2): zeilen.append((f'arInOutJamExit{n} := "DB_JamArea____".stJam____.arCarrier', f"{TODO} - Staubereich nach {nachbar(nach, alle)}")) zeilen.append((f'stInOutJamExit{n} := "DB_JamArea____".stJam____.stData', "")) - for i in range(1, items.get("Separator", 1) + 1): + for i in range(1, items.get(TroItem.SEPARATOR, 1) + 1): zeilen.append((f'xOutStopper{i} => "MB____"', TODO)) - if items.get("VarioDrive"): + if items.get(TroItem.VARIO_DRIVE): zeilen.append(('xOutVarioMotor => "MA____"', TODO)) - for j in range(1, items.get("Switch", 0) + 1): + for j in range(1, items.get(TroItem.SWITCH, 0) + 1): zeilen.append((f'xOutSw{j}ExTo3 => "MB____"', TODO)) zeilen.append((f'stInOutHMI := "DB_Interface_HMI".stTRO.{tro_id}', "")) zeilen.append(('nOutStateLast => "DB_GLOB_TroState".arTroState["cMainTro____"]', @@ -270,13 +279,13 @@ def emit_call(add, tro: dict, items: dict[str, int], alle: dict[str, dict]) -> N add(f"{text} // {kommentar}" if kommentar else text) -def emit_direction(add, tro: dict, anzahl: int) -> None: +def emit_direction(add, tro: TroBase, anzahl: int) -> None: """Weichenlogik - je Weiche ein FC_Direction-Aufruf.""" add("") add(" REGION Direction") for j in range(1, anzahl + 1): - ziel = (f"#{tro['id']}.nInDirection" if anzahl == 1 - else f"#{tro['id']}.nInDirection{j}") + ziel = (f"#{tro.tro_id}.nInDirection" if anzahl == 1 + else f"#{tro.tro_id}.nInDirection{j}") add(f' "FC_Direction"(nInSwitchNo := "cSwi_____", // {TODO}') add(f" nOutDirection => {ziel},") add(f' InOutCarrier := "DB_JamArea____".stJam____' @@ -286,9 +295,9 @@ def emit_direction(add, tro: dict, anzahl: int) -> None: add(" END_REGION") -def render(daten: dict, quelle: str, start_id: str = "") -> str: - tros = daten["tros"] - alle = {t["id"]: t for t in tros} +def render(document: TroDocument, quelle: str, start_id: str = "") -> str: + tros = document.tros + alle = {t.tro_id: t for t in tros} geordnet, startgrund = flow_order(tros, start_id) out: list[str] = [] @@ -310,7 +319,7 @@ def render(daten: dict, quelle: str, start_id: str = "") -> str: add("// (DB_JamArea...) und die Zeiten, soweit als 'pruefen' vermerkt.") add("//") add("// Reihenfolge der REGIONs = Materialfluss.") - add(f"// Anfang bei {geordnet[0]['id']}: {startgrund}.") + add(f"// Anfang bei {geordnet[0].tro_id}: {startgrund}.") add("// ============================================================") add("") add('FUNCTION_BLOCK "FB_Main"') @@ -318,45 +327,44 @@ def render(daten: dict, quelle: str, start_id: str = "") -> str: add("VERSION : 0.1") add(" VAR ") for tro in geordnet: - fb = tro.get("fbType") or "FB_UNBEKANNT" - kommentar = f" // {tro['type']}" - if tro.get("separators"): - kommentar += f", Sep {', '.join(tro['separators'])}" - add(f' {tro["id"]} : "{fb}";{kommentar}') + fb = tro.fb_block or "FB_UNBEKANNT" + kommentar = f" // {tro.type_name}" + if tro.separators: + kommentar += f", Sep {', '.join(tro.separators)}" + add(f' {tro.tro_id} : "{fb}";{kommentar}') add(" END_VAR") add("") add("BEGIN") for tro in geordnet: - items = tro.get("items", {}) - n_sep = items.get("Separator", 1) - n_swi = items.get("Switch", 0) - n_scan = items.get("Scanner", 0) + items = tro.items + n_sep = items.get(TroItem.SEPARATOR, 1) + n_swi = items.get(TroItem.SWITCH, 0) + n_scan = items.get(TroItem.SCANNER, 0) add("") add(f"REGION {region_title(tro)}") - pos = tro.get("position") - if pos: - add(f" // Anlagenkoordinate x/y: {pos['x']:.0f} / {pos['y']:.0f} mm") - add(f" // Vorgaenger: {', '.join(tro.get('predecessors', [])) or '-'}") - add(f" // Nachfolger: {', '.join(tro.get('successors', [])) or '-'}") - if tro.get("confidence"): - add(f" // Einstufung: {tro['confidence']}") + if tro.position: + add(f" // Anlagenkoordinate x/y: {tro.x:.0f} / {tro.y:.0f} mm") + add(f" // Vorgaenger: {', '.join(tro.predecessors) or '-'}") + add(f" // Nachfolger: {', '.join(tro.successors) or '-'}") + if tro.confidence: + add(f" // Einstufung: {tro.confidence}") add("") emit_separators(add, tro, n_sep, hat_scanner=bool(n_scan)) if n_swi: emit_switches(add, tro, n_swi, alle) - if items.get("VarioDrive"): + if items.get(TroItem.VARIO_DRIVE): emit_vario(add, tro) if n_scan: emit_barcode(add, tro, n_scan) - if items.get("StorageLine"): - add(f" // {TODO}: Linienspeicher mit {items['StorageLine']} Linien - " + if items.get(TroItem.STORAGE_LINE): + add(f" // {TODO}: Linienspeicher mit {items[TroItem.STORAGE_LINE]} Linien - " f"Parametrierung je Linie") add("") - add(" #%s.stInPriorityManager.xReleaseOk := TRUE;" % tro["id"]) + add(" #%s.stInPriorityManager.xReleaseOk := TRUE;" % tro.tro_id) add("") emit_call(add, tro, items, alle) if n_swi: @@ -378,32 +386,47 @@ def main(argv: list[str] | None = None) -> int: json_file = results / args.json if not json_file.is_file(): print(f"FEHLER: JSON nicht gefunden: {json_file}", file=sys.stderr) - print(" Zuerst bin/tro_extract.bat laufen lassen.", file=sys.stderr) + print(" Zuerst bin/tro_flow.bat --file .csv --json laufen lassen.", + file=sys.stderr) return 1 try: - daten = load_tros(json_file) - except (OSError, ValueError) as exc: + document = load_document(json_file) + except OSError as exc: print(f"FEHLER: {json_file} nicht lesbar: {exc}", file=sys.stderr) return 1 - if not daten.get("tros"): + except ValidationError as exc: + schema = _schema_of(json_file) + if schema and schema != SCHEMA: + # Anderes Format (z. B. die alte tro_extract-JSON) - die einzelnen + # Feldfehler helfen dann nicht weiter. + print(f"FEHLER: {json_file} hat das Schema '{schema}', erwartet wird " + f"'{SCHEMA}' ({exc.error_count()} Abweichungen).", file=sys.stderr) + print(" Mit bin/tro_flow.bat --file .csv --json neu erzeugen.", + file=sys.stderr) + return 1 + print(f"FEHLER: {json_file} passt nicht zum TRO-Modell (lib/model.py):", + file=sys.stderr) + print(exc, file=sys.stderr) + return 1 + if not document.tros: print(f"FEHLER: {json_file} enthaelt keine TROs.", file=sys.stderr) return 1 - text = render(daten, json_file.name, args.start) + text = render(document, json_file.name, args.start) out_file = results / (args.out or f"{json_file.stem.replace('_tro', '')}_FB_Main.scl") out_file.write_text(text, encoding="utf-8") offen = text.count(TODO) - geordnet, startgrund = flow_order(daten["tros"], args.start) + geordnet, startgrund = flow_order(document.tros, args.start) print("") print("================================================================") print("FB_MAIN-GERUEST") print("================================================================") print(f"Quelle = {json_file}") - print(f"TROs = {len(daten['tros'])}") - print(f"Reihenfolge = {' -> '.join(t['id'] for t in geordnet)}") - print(f"Anfang = {geordnet[0]['id']} ({startgrund})") + print(f"TROs = {len(document.tros)}") + print(f"Reihenfolge = {' -> '.join(t.tro_id for t in geordnet)}") + print(f"Anfang = {geordnet[0].tro_id} ({startgrund})") print(f"Offene Stellen = {offen} (Marker '{TODO}')") print(f"SCL = {out_file}") print("================================================================") @@ -411,10 +434,18 @@ def main(argv: list[str] | None = None) -> int: return 0 +def _schema_of(path: Path) -> str: + """Schema-Kennung einer JSON-Datei, nur fuer die Fehlermeldung ("" wenn unlesbar).""" + try: + return str(json.loads(path.read_text(encoding="utf-8")).get("schema", "")) + except (OSError, ValueError, AttributeError): + return "" + + def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser( prog="scl_skeleton", - description="Erzeugt aus der extrahierten TRO-Liste ein FB_Main-Geruest " + description="Erzeugt aus der TRO-JSON (tro_flow --json) ein FB_Main-Geruest " "als SCL zum Import in TIA Portal.", formatter_class=argparse.RawDescriptionHelpFormatter, epilog="Beispiele:\n" @@ -429,7 +460,7 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace: "--json", default="export_tro.json", metavar="NAME", - help="TRO-JSON aus tro_extract.py in %%SKEL_RESULTS%% " + help="TRO-JSON aus tro_flow.py --json in %%SKEL_RESULTS%% " "(oder ein vollstaendiger Pfad). Standard: %(default)s", ) parser.add_argument( diff --git a/lib/tro_annotate.py b/lib/tro_annotate.py index d2eaa1c..13f9c75 100644 --- a/lib/tro_annotate.py +++ b/lib/tro_annotate.py @@ -12,6 +12,12 @@ TROs und schreibt eine **Kopie** der Zeichnung mit zusaetzlicher Beschriftung: * optional Pfeile fuer den gerichteten Materialfluss zwischen den TROs * optional eine Legende +Die Zeichnung ist eine **Ansicht** zur Sichtkontrolle, keine Datenquelle: die +TRO-JSON fuer scl_skeleton.py schreibt tro_flow.py --json direkt aus dem CSV. +tro_annotate leitet die TROs mit genau demselben Aufruf ab wie tro_flow, damit +Zeichnung und JSON uebereinstimmen. (tro_extract.py, der Rueckweg Zeichnung -> +JSON, ist geparkt.) + Die Originalzeichnung wird nie veraendert. Alle Ergaenzungen liegen auf eigenen Layern (Vorgabe: Praefix "TRO_") und tragen XDATA der Anwendungskennung SPS_SKEL_TRO. Ein erneuter Lauf entfernt genau diese Objekte und baut sie neu - @@ -101,13 +107,15 @@ from dxf_registration import ( ) from material_flow import build_graph, env_dir, read_elements, resolve_input from tro_catalog import STYLES, TRO_CATALOG, TroSymbol, get_tro +from model import Analysis from tro_flow import ( BOUNDARY_ENTRY, BOUNDARY_EXIT, STATION_FARBE, - Analysis, _station_positionen, analyse_tros, + find_sibling_raw_json, + load_sepliste_counts, ) # --------------------------------------------------------------------------- @@ -898,9 +906,13 @@ def main(argv: list[str] | None = None) -> int: print(f"FEHLER: {csv_file} nicht lesbar: {exc}", file=sys.stderr) return 1 graph = build_graph(elements, csv_file.name, read_warnings) - # Ohne cfg/tro_overrides.ini - wie tro_flow, damit Marker und - # Flussdiagramm zusammenpassen. - analysis = analyse_tros(graph, elements) + # Derselbe Aufruf wie in tro_flow.main (ohne cfg/tro_overrides.ini, mit + # der Separator-Gegenprobe aus _raw.json), 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) 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 c29b02b..5c2fa63 100644 --- a/lib/tro_flow.py +++ b/lib/tro_flow.py @@ -86,6 +86,8 @@ Ausgabe (alles in SKEL_RESULTS) _tro_flow.dot immer (mit --use-cords lagerichtige pos-Attribute) _tro_flow.svg mit --tosvg (dot, mit --use-cords neato -n) _tro_doc.md mit --doc + _tro.json mit --json: TRO-JSON (model.TroDocument), Eingabe + fuer scl_skeleton.py _tro_errors.md nur wenn TRO-Fehler gefunden wurden Aufruf ueber bin/tro_flow.bat bzw. bin/tro_flow.sh. @@ -98,7 +100,6 @@ import csv import json import math import sys -from dataclasses import dataclass, field from datetime import datetime from pathlib import Path @@ -124,13 +125,13 @@ from material_flow import ( read_elements, resolve_input, ) +from model import Analysis, DerivedTro, TroDocument from tro_catalog import ( CATALOG_AS_OF, ITEM_FB, STYLE_EXT, STYLES, TRO_CATALOG, - TroDefinition, TroItem, get_tro, mermaid_classdefs, @@ -177,82 +178,6 @@ COORD_MAX_LUECKE = 1500 LEGEND_MARGIN = 3000 -# --------------------------------------------------------------------------- -# Datenmodell -# --------------------------------------------------------------------------- - - -@dataclass -class Tro: - """Eine ermittelte Uebergabestelle.""" - - tro_id: str - type_name: str - label: str - hosts: list[str] - separators: list[str] - scanners: list[str] - reason: str - confidence: str - fb_block: str = "" - items: dict[str, int] = field(default_factory=dict) - successors: list[str] = field(default_factory=list) - predecessors: list[str] = field(default_factory=list) - colocated: list[str] = field(default_factory=list) - # Mittelpunkt der Bauteile, aus denen dieser TRO gebildet wurde - position: tuple[float, float, float] | None = None - position_from: str = "" - position_points: int = 0 - - @property - def x(self) -> float | None: - return self.position[0] if self.position else None - - @property - def y(self) -> float | None: - return self.position[1] if self.position else None - - @property - def z(self) -> float | None: - return self.position[2] if self.position else None - - def describe_position(self) -> str: - """Koordinate als Text, z. B. '1030, 4700'.""" - if not self.position: - return "-" - return f"{self.position[0]:.0f}, {self.position[1]:.0f}" - - @property - def connected(self) -> bool: - return bool(self.successors or self.predecessors or self.colocated) - - def definition(self) -> TroDefinition | None: - """Typdefinition aus tro_catalog (None wenn Typ unbekannt).""" - return get_tro(self.type_name) - - -@dataclass -class Analysis: - tros: list[Tro] = field(default_factory=list) - edges: list[tuple[str, str]] = field(default_factory=list) - # (von, nach) -> Kantenart (transfer / umlauf / weiche / unbestimmt / normal) - edge_kinds: dict[tuple[str, str], str] = field(default_factory=dict) - isolated: list[Tro] = field(default_factory=list) - unknown_types: list[str] = field(default_factory=list) - findings: list[str] = field(default_factory=list) - # Wie Host/Fanout ermittelt wurden (Doku/Bericht) - method: str = "Materialfluss-Graph (Zuordnung/Nachbarn)" - # Systemgrenzen (BTMT Be-/Entladestationen) - siehe derive_boundaries; kein - # TRO, nur damit der tatsaechliche Materialein-/ausgang sichtbar wird. - boundaries: list[Tro] = field(default_factory=list) - - def by_id(self, tro_id: str) -> Tro | None: - for tro in self.tros + self.boundaries: - if tro.tro_id == tro_id: - return tro - return None - - # --------------------------------------------------------------------------- # TRO-Ermittlung # --------------------------------------------------------------------------- @@ -730,7 +655,7 @@ def derive_tros( findings: list[str], *, sepliste_counts: dict[str, int] | None = None, -) -> list[Tro]: +) -> list[DerivedTro]: """TRO-Liste aus den Separatoren und ihren angeschlossenen Objekten bilden. Host eines Separators aus der CSV-Spalte "Zuordnung" (_default_resolve_host), @@ -862,7 +787,7 @@ def derive_tros( f"sepliste (Roh-JSON) - Gegenprobe weicht ab, bitte pruefen" ) - tros: list[Tro] = [] + tros: list[DerivedTro] = [] # 1) Schwerkraftlinien (Gefaellestrecke mit Separator) -> Speicherblock lanes = sorted( @@ -887,7 +812,7 @@ def derive_tros( speicher_steuer.update(steuer) alle_seps = steuer + seps tros.append( - Tro( + DerivedTro( tro_id="", type_name="PinStore_Auto", label=f"Speicher ({len(group)} Schwerkraftlinien)", @@ -907,7 +832,7 @@ def derive_tros( for lane in group: lane_seps = seps_on([lane]) tros.append( - Tro( + DerivedTro( tro_id="", type_name="1Sep", label=graph.nodes[lane].element.name, @@ -973,7 +898,7 @@ def derive_tros( # Streckengruppe genau 1 echter Kettenseparator, alle # uebrigen Zuordnung-Treffer sind eigenstaendige Vereinzeler). tros.append( - Tro( + DerivedTro( tro_id="", type_name="1Sep", label=element.name, @@ -1034,7 +959,7 @@ def derive_tros( f"Vario eingestuft (widerspricht alternierendem Muster)" ) tros.append( - Tro( + DerivedTro( tro_id="", type_name=type_name, label=element.name, @@ -1055,7 +980,7 @@ def derive_tros( node = graph.nodes[host] lane_name = _merkmal(node.element.merkmale, "Name") or node.element.name tros.append( - Tro( + DerivedTro( tro_id="", type_name="1Sep", label=f"Separator {sep_id} auf {lane_name} Bahn {node.lane}", @@ -1079,7 +1004,7 @@ def derive_tros( # vor), und die Strecke hat bereits reale, gerichtete Kanten zum Kreisel # (_apply_transfers) - darueber haengt sich die Verknuepfung (link_tros) # von selbst ein, ohne einen weiteren synthetischen Host zu erfinden. - as_switch_tro: dict[str, Tro] = {} # Ausschleuselement-Id -> 1Sep1Swi + as_switch_tro: dict[str, DerivedTro] = {} # Ausschleuselement-Id -> 1Sep1Swi for sep_id, (transfer, distance) in sorted(switch_hits.items()): if sep_id in speicher_steuer: continue # Teil seines Speichers (Schritt 1) @@ -1088,7 +1013,7 @@ def derive_tros( own_scanners = sorted(sep_scanners.get(sep_id, [])) art = "Ausschleuselement" as_switch_tro[transfer.teile_id] = tro = ( - Tro( + DerivedTro( tro_id="", type_name="1Sep1Swi", label=f"Separator {sep_id} an {art} {transfer.teile_id} " @@ -1161,10 +1086,10 @@ def derive_tros( # 3c) Separatoren an einer echten ILS Weiche -> 1Sep1Swi (siehe # _find_weiche_separators). Host wie bei 3b: der aufgeloeste Zuordnung-Host # des Separators, sonst die Weiche selbst. - weiche_tro: dict[str, Tro] = {} # Weichen-Id -> 1Sep1Swi + weiche_tro: dict[str, DerivedTro] = {} # Weichen-Id -> 1Sep1Swi for sep_id, weiche in sorted(weiche_hits.items()): host = resolve_host(elements_by_id[sep_id], findings) - weiche_tro[weiche.teile_id] = tro = Tro( + weiche_tro[weiche.teile_id] = tro = DerivedTro( tro_id="", type_name="1Sep1Swi", label=f"Separator {sep_id} an ILS Weiche {weiche.teile_id}", @@ -1194,7 +1119,7 @@ def derive_tros( continue tros.remove(as_tro) tros.remove(w_tro) - tros.append(Tro( + tros.append(DerivedTro( tro_id="", type_name="2Sep2Swi", label=(f"Uebergabestelle {kreisel.name}: Ausschleuselement " @@ -1258,7 +1183,7 @@ BOUNDARY_EXIT = "Materialausgang" BOUNDARY_TYPES = (BOUNDARY_ENTRY, BOUNDARY_EXIT) -def derive_boundaries(elements: list[Element], graph: Graph) -> list[Tro]: +def derive_boundaries(elements: list[Element], graph: Graph) -> list[DerivedTro]: """Be-/Entladestationen als sichtbare Systemgrenz-Knoten. Ohne diese Knoten "verschwindet" der Materialfluss stillschweigend an der @@ -1286,14 +1211,14 @@ def derive_boundaries(elements: list[Element], graph: Graph) -> list[Tro]: getrennt von den TRO-IDs, damit auf den ersten Blick klar bleibt: das sind keine Steuerobjekte. """ - boundaries: list[Tro] = [] + boundaries: list[DerivedTro] = [] for element in elements: if not element.is_station: continue if element.teile_id not in graph.nodes: continue boundaries.append( - Tro( + DerivedTro( tro_id="", type_name=BOUNDARY_ENTRY if element.is_entry else BOUNDARY_EXIT, label=element.name, @@ -1349,7 +1274,7 @@ def _bahn_richtung(node) -> tuple[float, float] | None: return achse if mit_achse else (-achse[0], -achse[1]) -def _tro_bahn(tro: Tro, graph: Graph, sep_assignment: dict[str, str]) -> str | None: +def _tro_bahn(tro: DerivedTro, graph: Graph, sep_assignment: dict[str, str]) -> str | None: """Kreisel-Bahn, auf der ein TRO physisch sitzt - fuer die Positions- Verkettung in link_tros(). Meist einfach der TRO-Host (wenn er direkt auf der Bahn liegt, z. B. ein gewoehnlicher Kreisel-Separator). Bei einem @@ -1430,7 +1355,7 @@ def _chain_mit_ungelisteten_separatoren( def link_tros( - graph: Graph, tros: list[Tro], elements: list[Element] + graph: Graph, tros: list[DerivedTro], elements: list[Element] ) -> tuple[list[tuple[str, str]], dict[tuple[str, str], str]]: """ TROs verknuepfen. @@ -1692,7 +1617,7 @@ def link_tros( # vorbei (Review 2026-09-21). sep_assignment = {e.teile_id: e.assignment for e in elements if e.kind == "Separator" and e.assignment} - bahn_gruppen: dict[str, list[Tro]] = {} + bahn_gruppen: dict[str, list[DerivedTro]] = {} for tro in tros: bahn = _tro_bahn(tro, graph, sep_assignment) if bahn and bahn.split("-")[0] not in covered_kreisel_bases: @@ -1814,7 +1739,7 @@ def _centroid( ) -def locate_tros(tros: list[Tro], elements: list[Element], graph: Graph) -> None: +def locate_tros(tros: list[DerivedTro], elements: list[Element], graph: Graph) -> None: """ Jedem TRO eine Koordinate geben. @@ -1872,15 +1797,8 @@ def analyse_tros( locate_tros(linked, elements, graph) analysis.edges, analysis.edge_kinds = link_tros(graph, linked, elements) - for tro in analysis.tros: - definition = tro.definition() - if definition: - tro.fb_block = definition.get_fb_block() - tro.items = definition.get_items() - else: - analysis.unknown_types.append(tro.type_name) - - analysis.unknown_types = sorted(set(analysis.unknown_types)) + analysis.unknown_types = sorted( + {tro.type_name for tro in analysis.tros if tro.definition() is None}) analysis.isolated = [tro for tro in analysis.tros if not tro.connected] analysis.findings.sort() return analysis @@ -1891,7 +1809,7 @@ def analyse_tros( # --------------------------------------------------------------------------- -def _tro_label(tro: Tro) -> str: +def _tro_label(tro: DerivedTro) -> str: lines = [f"{tro.tro_id} ({tro.type_name})", tro.label] if tro.fb_block: lines.append(tro.fb_block) @@ -1968,7 +1886,7 @@ def _station_positionen( echten Station entscheidet nur bei Gleichstand. Station ohne angehaengten TRO: echte Koordinate.""" tro_by_id = {t.tro_id: t for t in analysis.tros} - je_tro: dict[str, list[Tro]] = {} + je_tro: dict[str, list[DerivedTro]] = {} ergebnis: dict[str, tuple[float, float]] = {} for station in analysis.boundaries: nachbarn = (station.successors if station.type_name == BOUNDARY_ENTRY @@ -2017,7 +1935,7 @@ def _station_positionen( return ergebnis -def _tro_kurzinfo(tro: Tro) -> str: +def _tro_kurzinfo(tro: DerivedTro) -> str: """Zweizeilige Beschriftung neben dem Symbol: Typ und Bauteile.""" if tro.type_name in BOUNDARY_TYPES: return _dot_label([tro.type_name, tro.label]) @@ -2030,12 +1948,12 @@ def _tro_kurzinfo(tro: Tro) -> str: return _dot_label([tro.type_name, teile]) -def _tro_tooltip(tro: Tro) -> str: +def _tro_tooltip(tro: DerivedTro) -> str: """Vollstaendige Angaben als SVG-Tooltip (Zeilenumbruch per ).""" return _tro_label(tro).replace("\\n", " ") -def _coord_node_attrs(tro: Tro, base: dict[str, str]) -> str: +def _coord_node_attrs(tro: DerivedTro, base: dict[str, str]) -> str: """Knotenattribute fuer --use-cords: feste Groesse, kurzes Label im Symbol, Typ/Bauteile als xlabel daneben, alle Details als Tooltip.""" attrs = dict(base) @@ -2069,7 +1987,7 @@ def _zeichen_positionen( """Zeichenposition (mm) je TRO/Systemgrenze fuer --use-cords: TROs auf ihrer Anlagenkoordinate, mit compact die Leerstrecken gekuerzt (_kompakte_positionen); die Stationsbeschriftungen danach relativ zum - angehaengten TRO (_station_positionen). Reine Darstellung - Tro.position + angehaengten TRO (_station_positionen). Reine Darstellung - DerivedTro.position und die Tooltip-Koordinaten bleiben echt.""" roh: dict[str, tuple[float, float]] = {} for tro in analysis.tros: @@ -2117,7 +2035,7 @@ def render_dot(analysis: Analysis, source: str, use_coords: bool = False, add = out.append zeichen = _zeichen_positionen(analysis, compact) if use_coords else {} - def _pos(node: Tro) -> str: + def _pos(node: DerivedTro) -> str: if use_coords: p = zeichen.get(node.tro_id) return (f', pos="{p[0] * COORD_SCALE:.2f},{p[1] * COORD_SCALE:.2f}"' @@ -2719,6 +2637,7 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace: " tro_flow.bat --file mubea.csv --tosvg\n" " 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" "\n" "Exit-Codes:\n" " 0 ok\n" @@ -2761,6 +2680,12 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace: "schreiben: Knoten (TRO = Typ), Verbindungen, Systemgrenzen und je " "TRO Vorgaenger/Nachfolger.", ) + parser.add_argument( + "--json", + action="store_true", + help="TRO-JSON als _tro.json nach %%SKEL_RESULTS%% schreiben " + "(Modell lib/model.py, TroDocument) - Eingabe fuer scl_skeleton.py.", + ) parser.add_argument( "--true-scale", dest="true_scale", @@ -2812,6 +2737,7 @@ def main(argv: list[str] | None = None) -> int: stem = csv_file.stem connect_target = results / f"{stem}_tro_connect.ini" + json_target = results / f"{stem}_tro.json" analysis = analyse_tros(graph, elements, sepliste_counts=sepliste_counts) if not analysis.tros: @@ -2847,6 +2773,16 @@ def main(argv: list[str] | None = None) -> int: ) connect_file = connect_target + json_file = None + if args.json: + document = TroDocument.from_analysis( + analysis, source=csv_file.name, + note="Aus dem CSV-Export abgeleitet (lib/tro_flow.py) - die " + "Zeichnung ist nur Ansicht, keine Datenquelle.", + ) + json_target.write_text(document.to_json(), encoding="utf-8") + json_file = json_target + has_errors = write_error_file(analysis, csv_file, error_target) error_file = error_target if has_errors else None @@ -2872,6 +2808,8 @@ def main(argv: list[str] | None = None) -> int: warnings, engine=engine) if connect_file: print(f"connect.ini = {connect_file}") + if json_file: + print(f"TRO-JSON = {json_file}") if svg_error: print(f"FEHLER: SVG nicht erzeugt: {svg_error}", file=sys.stderr) diff --git a/requirements.txt b/requirements.txt index b014d57..67bcdde 100644 --- a/requirements.txt +++ b/requirements.txt @@ -9,5 +9,5 @@ ezdxf >= 1.4.0 # Nur fuer Vorschaubilder eines DXF (ezdxf.addons.drawing) - optional: # matplotlib >= 3.8 -# Noch nicht benoetigt: -# pytest >= 9.0.0 +# Tests (tests/, Aufruf: python -m pytest tests/) +pytest >= 9.0.0 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..23490c0 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,9 @@ +# -*- coding: utf-8 -*- +"""lib/ importierbar machen, auch ohne vorher bin/setenv aufzurufen.""" + +import sys +from pathlib import Path + +LIB = Path(__file__).resolve().parent.parent / "lib" +if str(LIB) not in sys.path: + sys.path.insert(0, str(LIB)) diff --git a/tests/test_model.py b/tests/test_model.py new file mode 100644 index 0000000..d389741 --- /dev/null +++ b/tests/test_model.py @@ -0,0 +1,127 @@ +# -*- coding: utf-8 -*- +"""Tests fuer lib/model.py - die TRO-JSON zwischen tro_flow und scl_skeleton.""" + +import json + +import pytest +from pydantic import ValidationError + +from model import SCHEMA, Analysis, Connection, DerivedTro, TroDocument +from scl_skeleton import render + + +def _tro(tro_id: str, type_name: str, **extra) -> DerivedTro: + data = dict( + tro_id=tro_id, + type_name=type_name, + label=f"Label {tro_id}", + hosts=["0001"], + separators=["0040"], + scanners=[], + reason="Test", + confidence="mittel", + ) + data.update(extra) + return DerivedTro(**data) + + +@pytest.fixture +def document() -> TroDocument: + """Kleine Anlage: SYS1 -> TRO01 (Weiche) -> TRO02 / TRO03 -> SYS2.""" + analysis = Analysis( + tros=[ + _tro("TRO01", "1Sep1Swi", predecessors=["SYS1"], successors=["TRO02", "TRO03"], + position=(1030.0, 4700.0, 1500.0)), + _tro("TRO02", "1Sep", separators=["0041"], predecessors=["TRO01"], + successors=["SYS2"]), + _tro("TRO03", "Vario", separators=["0042"], predecessors=["TRO01"]), + ], + boundaries=[ + _tro("SYS1", "Materialeingang", separators=[], successors=["TRO01"]), + _tro("SYS2", "Materialausgang", separators=[], predecessors=["TRO02"]), + ], + edges=[("SYS1", "TRO01"), ("TRO01", "TRO02"), ("TRO01", "TRO03"), ("TRO02", "SYS2")], + edge_kinds={("TRO01", "TRO02"): "weiche"}, + findings=["ein Befund"], + ) + return TroDocument.from_analysis(analysis, source="test.csv", note="Test") + + +def test_round_trip_is_lossless(document): + text = document.to_json() + again = TroDocument.from_json(text) + assert again.to_json() == text + assert again.by_id("TRO01").successors == ["TRO02", "TRO03"] + + +def test_json_uses_tro_json_keys(document): + data = json.loads(document.to_json()) + assert data["schema"] == SCHEMA + first = data["tros"][0] + assert first["id"] == "TRO01" + assert first["type"] == "1Sep1Swi" + assert first["fbType"] == "FB_ILS_MTRO_1Sep1Swi" + assert first["items"] == {"Separator": 1, "Switch": 1} + assert first["position"] == {"x": 1030.0, "y": 4700.0, "z": 1500.0} + assert data["connections"][1] == {"from": "TRO01", "to": "TRO02", "kind": "weiche"} + assert "openPoints" in data + + +def test_items_and_fb_block_follow_the_type(): + tro = _tro("TRO01", "1Sep") + assert tro.items == {"Separator": 1} + tro.type_name = "1Sep2Swi" + assert tro.items == {"Separator": 1, "Switch": 2, "Scanner": 1} + assert tro.fb_block == "FB_ILS_MTRO_1Sep2Swi" + + +def test_unknown_type_has_no_catalog_data(): + station = _tro("SYS1", "Materialeingang") + assert station.definition() is None + assert station.items == {} + assert station.fb_block == "" + + +def test_catalog_fields_cannot_be_assigned(): + tro = _tro("TRO01", "1Sep") + with pytest.raises((AttributeError, ValidationError)): + tro.fb_block = "FB_Anders" + + +def test_unknown_key_is_rejected(document): + data = json.loads(document.to_json()) + data["tros"][0]["separatorz"] = data["tros"][0].pop("separators") + with pytest.raises(ValidationError, match="separatorz"): + TroDocument.model_validate(data) + + +def test_items_not_matching_catalog_are_rejected(document): + data = json.loads(document.to_json()) + data["tros"][1]["items"] = {"Separator": 2} + with pytest.raises(ValidationError, match="passt nicht zum Typ"): + TroDocument.model_validate(data) + + +def test_fb_type_not_matching_catalog_is_rejected(document): + data = json.loads(document.to_json()) + data["tros"][1]["fbType"] = "FB_ILS_MTRO_Vario" + with pytest.raises(ValidationError, match="fbType"): + TroDocument.model_validate(data) + + +def test_connection_accepts_alias_and_field_name(): + assert Connection(**{"from": "A", "to": "B"}) == Connection(src="A", dst="B") + + +def test_scl_skeleton_renders_from_document(document): + scl = render(document, "test_tro.json") + assert 'TRO01 : "FB_ILS_MTRO_1Sep1Swi";' in scl + assert 'TRO03 : "FB_ILS_MTRO_Vario";' in scl + # Weiche: FC_Direction und eine Zielliste je Abgang + assert '"FC_Direction"' in scl + assert "// ueber TRO02: SYS2, TRO02" in scl + # Vario-TRO bekommt den Vario-Parameterblock + assert "#TRO03.stInVario.Settings" in scl + # Systemgrenzen sind keine Instanzen + assert "SYS1 :" not in scl + assert "Anfang bei TRO01" in scl