From 39b445a7c423de47927f71f022ff292c3596a26e Mon Sep 17 00:00:00 2001 From: Samer Ayadi Date: Fri, 16 Oct 2026 10:53:52 +0200 Subject: [PATCH] scl_skeleton: Anlagen-JSON als Eingabe (Stufe 4, Namen) --json erkennt _anlage.json am Schema. scl_context.HwSource setzt Namen aus bindings (nur "zugeordnet") und conveyors ein: Separator-, Lagerlinien-, Weichen- und Vario-Antriebsnamen, #fbCarousel am Kreisel, Foerderernummer im DB_JamArea-Platzhalter ( bleibt offen). Eingesetzte Namen werden gegen _hw.json geprueft (--hw, --strict bricht ab). --eplan bleibt fuer die TRO-JSON; Templates und Golden-Files unveraendert. Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 8 +- doc/Konzept_Layout_Hardware_SCL.md | 7 + doc/Python_Scripts.md | 49 ++++- lib/scl_context.py | 282 ++++++++++++++++++++++------- lib/scl_skeleton.py | 155 +++++++++++++--- tests/test_scl_anlage.py | 183 +++++++++++++++++++ 6 files changed, 588 insertions(+), 96 deletions(-) create mode 100644 tests/test_scl_anlage.py diff --git a/CLAUDE.md b/CLAUDE.md index 8864834..33220f4 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -227,8 +227,12 @@ byte — rewrite them (`SKEL_UPDATE_GOLDEN=1`) only for a reviewed, intended cha TRO-ID + slot + role, `conveyors[]`). Reads no CSV: the TRO-JSON carries `parts` (positions/bounding boxes, written by `tro_flow --json`). Conveyors are matched by majority vote of the devices naming them; corrections in `cfg/hw_overrides.ini`, - numbers in `cfg/parameter.ini [merge_hw_layout]`. `scl_skeleton` does not read the plant JSON - yet (stage 4). See `doc/Python_Scripts.md` §5e. + numbers in `cfg/parameter.ini [merge_hw_layout]`. See `doc/Python_Scripts.md` §5e. +8. **`scl_skeleton.py --json _anlage.json`** (stage 4, names) — reads the plant JSON + (schema detected) and fills sensors, valves, drives and conveyor numbers from `bindings`/ + `conveyors` via `scl_context.HwSource`; 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. What is **not** built out yet, relative to the original two-tool design: diff --git a/doc/Konzept_Layout_Hardware_SCL.md b/doc/Konzept_Layout_Hardware_SCL.md index f006f8f..df0c8b3 100644 --- a/doc/Konzept_Layout_Hardware_SCL.md +++ b/doc/Konzept_Layout_Hardware_SCL.md @@ -188,6 +188,13 @@ Weichenventile (`doc/E-Planung_Bruecke.md`, "Offen"). ### 4. `scl_skeleton.py` / `scl_context.py` (Stufe 4) +**Stand 2026-10-16, umgesetzt (Namen):** `scl_skeleton --json _anlage.json` +(Schema erkannt), Namen über `scl_context.HwSource` aus `bindings` (nur +`zugeordnet`) und `conveyors` (`#fbCarousel`, `DB_JamArea` mit +offenem ``), Gegenprüfung gegen `_hw.json` (`--strict` bricht ab). +`--eplan` bleibt für die TRO-JSON, bis `e_planning.py` entfällt. Offen: neue +Bausteine (unten), Stausensoren der Ausgänge, Scanner. + - Liest **eine** Eingabe: `_anlage.json`; ohne Hardwareteil wie heute `_tro.json`. - `--eplan` und die Zusammenführung in `scl_context._separator` entfallen; der diff --git a/doc/Python_Scripts.md b/doc/Python_Scripts.md index e73c0ac..8ec6d05 100644 --- a/doc/Python_Scripts.md +++ b/doc/Python_Scripts.md @@ -46,11 +46,13 @@ tro_flow.py ──► _tro_flow.dot / .svg, _tro_doc.md, _tro_error │ (assigns BG/MB of the E-planning drawing to the layout │ parts by position + numbering rule; see §5c) │ - │ NEW PATH (stages 2/3, scl_skeleton does not read it yet): + │ NEW PATH (stages 2-4): │ _tro.json ──┐ │ ├─► merge_hw_layout.py ──► _anlage.json + .md (§5e) │ hw_import.py ─────┘ (TRO slots <- BMK, conveyors K0102 -> Kreisel3) │ ▲ *_TIA.xlsx + *_positionsconv.json (§5d) + │ _anlage.json ──► scl_skeleton.py --json ──► _FB_Main.scl + │ (names from the bindings; replaces --eplan) │ └── same derivation ──► tro_annotate.py ──► _annotated.dxf (VIEWER ONLY) (also needs the BricsCAD DXF; uses dxf_registration.py for the CSV↔DXF @@ -114,6 +116,7 @@ bin\tro_flow.bat --file mubea.csv --tosvg --use-cords --doc --json bin\scl_skeleton.bat --json mubea_tro.json bin\e_planning.bat --file mubea.csv --eplan Mubea_positionsdraw.json bin\scl_skeleton.bat --json mubea_tro.json --eplan mubea_eplan.json +bin\scl_skeleton.bat --json mubea_anlage.json --templates bin\tro_annotate.bat --file mubea.csv --dxf 500573_60_1.dxf --flow --fb --legend ``` @@ -532,8 +535,10 @@ 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`, looked up in -`%SKEL_RESULTS%`; written by `tro_flow.py --json`), `--out NAME` (default -`_FB_Main.scl`), `--eplan NAME` (see below), `--start TRO`. The last one matters while the plant has +`%SKEL_RESULTS%`; either the TRO-JSON from `tro_flow.py --json` or the plant +JSON from `merge_hw_layout.py`, told apart by its `schema`), `--out NAME` +(default `_FB_Main.scl`, `_tro`/`_anlage` dropped from the source +name), `--eplan NAME` and `--hw NAME` (see below), `--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. @@ -551,8 +556,9 @@ shows up as a destination in the per-switch lists. (`build_context` → `SclFile` / `SclTro`: flow order, separator/switch/scanner blocks, jam entries/exits, destination lists, names), and `scl_skeleton.py` only turns that context into text (`render_context`). Every name or address is an -`SclName` (value + `todo` flag); today all of them are placeholders with -`todo=True`. This split prepares the move to parameterised templates and +`SclName` (value + `todo` flag); without hardware planning all of them are +placeholders with `todo=True`. Real names come from `scl_context.HwSource` +(plant JSON or, phasing out, the eplan assignment). This split prepares the move to parameterised templates and per-project naming rules described in [`SCL_Templates.md`](SCL_Templates.md). **Templates (opt-in, `--templates [DIR]`).** With this switch every TRO whose FB @@ -567,7 +573,30 @@ by a PLC engineer (`doc/SCL_Templates.md` §6). `2Sep1Swi`, `EmptyCarrBuffer`, `LoadingBoom` and `2Sep2Swi` have no call site yet and stay generic. `--strict` stops (exit `1`, no SCL) if any FB type has no template. Without `--templates` the output is unchanged. -**Names from the E-planning (opt-in, `--eplan NAME`).** Reads the assignment +**Names from the plant JSON (stage 4, `--json _anlage.json`).** The +layout is `anlage.layout`; names come from `bindings[]` by TRO-ID + slot + role +(`AnlageDocument.binding`), and only status `zugeordnet` counts — `offen` and +`widerspruch` stay `TODO(E-Planung)`. Sensors are written as the TIA constant +(`Binding.constant`, e.g. `cInBG0642`), outputs as the BMK (`MB0643`). + +| Slot / role | SCL | +|---|---| +| `separator` `sensor` / `stopper` / `partSensor` | `stSenInSep`, `xOutStopper`, `stSenPart` | +| `line` `sensor` / `stopper` / `jamSensor` | PinStore storage lines | +| `switch` `valve` | `xOutSwExTo3` | +| `drive` `motor` / `motorProtection` / `sen*` | Vario motor, `xInMotorProtection`, `stSenCarrInPos` … (motor/protection fall back to the conveyor of the TRO's host) | +| `conveyors[]` on the TRO's host | `xInCarouselRun` = `#fbCarousel.xOutConveyorFullSpeed` (Kreisel only); jam DB `"DB_JamArea".stJam_____` — still `TODO`, the running number `` is not fixed yet | + +Not filled yet: jam sensors at the exits (`stInSenJam`, the `Ausschleus.stau` +bindings have no slot), scanners, separator/switch/MainTro numbers. The +templates are unchanged — a known name simply drops its marker. Afterwards +every inserted name is checked against the hardware planning (`hwFile` of the +plant JSON in `%SKEL_RESULTS%`, or `--hw NAME`): names missing there are listed +as a warning, with `--strict` the run stops (exit `1`, no SCL). `--eplan` +together with a plant JSON is an error. Mubea: 46 names inserted, 47 fewer +`TODO`s (482 → 435), 0 unknown names. Tests: `tests/test_scl_anlage.py`. + +**Names from the E-planning (phasing out, `--eplan NAME`, TRO-JSON only).** Reads the assignment `_eplan.json` written by `e_planning.py` (§5c; bare name looked up in `%SKEL_RESULTS%`). For every separator it assigns, the separator sensor (`stSenInSep := "DB_Inputs".Sensors["cInBG0642"]`) and the stopper valve @@ -790,9 +819,9 @@ K0105 = Kreisel1 (2/3)**, each with motor and motor protection. The other conveyors (Aufgabestation #100/#101, Vario #106–#109) are outside the CSV or have a single vote and stay open. -**Not yet:** `scl_skeleton` still reads `_eplan.json` (`--eplan`); -switching it to `_anlage.json` is stage 4 of the concept, and only then -`e_planning.py` becomes the thin wrapper and is removed. +**Stage 4:** `scl_skeleton --json _anlage.json` reads the plant JSON +directly (§5b). `--eplan` still works for a TRO-JSON and is phased out together +with `e_planning.py`. Tests: `tests/test_merge_hw_layout.py` (sensor spatial / stopper by number, contradiction, open roles, part without slot, conveyor majority with @@ -855,7 +884,7 @@ tro_flow.py (writes) · scl_skeleton.py (reads) · tro_annotate.py (draws) | `LayoutPart` | A layout part with position (`layoutId`, `kind`, `name`, `position`, `bbox`) — the target of the spatial assignment in `merge_hw_layout.py`, so that it never reads the CSV. `distance_to_box()`. Optional in v4; a file without `parts` still reads. | | `Binding` | One assignment hardware → slot role: `tro`, `slot`, `role`, `layoutId`, `bmk`, `constant`, `address`, `source`, `rule` (`raeumlich`/`nummer`/`mehrheit`/`hand`), `distance`, `runnerUp`, `status` (`zugeordnet`/`offen`/`widerspruch`/`ohne_slot`), `note`. | | `ConveyorBinding` | A conveyor of the E-planning (`K0102`) at a layout conveyor: `eplanName`, `layoutId`, `layoutName`, `votes`/`total`, `motor`, `protection`, `encoder`, `rule`, `status`, `note`. | -| `AnlageDocument` | The plant JSON `_anlage.json` (`sps_skel-anlage-v1`): `layout` (the TRO-JSON unchanged), `registration`, `bindings`, `conveyors`, `unassigned`, `warnings`. `binding(tro, slot, role)`, `conveyor_for(layout_id)`. Written by `merge_hw_layout.py`; the future single input of `scl_skeleton`. | +| `AnlageDocument` | The plant JSON `_anlage.json` (`sps_skel-anlage-v1`): `layout` (the TRO-JSON unchanged), `registration`, `bindings`, `conveyors`, `unassigned`, `warnings`. `binding(tro, slot, role)`, `conveyor_for(layout_id)`. Written by `merge_hw_layout.py`; input of `scl_skeleton` (stage 4). | `TroBase` and `TroCandidate` share the `position` handling (JSON `{x,y,z}`, `x`/`y`/`z`, `describe_position()`) through the private mixin `_Positioned`. diff --git a/lib/scl_context.py b/lib/scl_context.py index eec37cb..c100597 100644 --- a/lib/scl_context.py +++ b/lib/scl_context.py @@ -16,11 +16,19 @@ Schichten: Namen (Separator-, Weichen-, MainTro-Nummer, Adressen ...) sind ein :class:`SclName`: Wert plus Merker, ob er noch aus der E-Planung kommen muss. -Ohne E-Planung sind alle Platzhalter ``todo=True``. Mit einer Zuordnung aus -lib/e_planning.py (``eplan``) werden Separator-Sensor und Stopperventil der -zugeordneten Separatoren eingesetzt und ``todo`` geloescht; alles andere -bleibt offen. Die Namensregeln aus Schritt 4 (cfg/scl_naming.ini) arbeiten -spaeter genauso: Wert ersetzen, Merker loeschen. +Ohne Hardwareplanung sind alle Platzhalter ``todo=True``. Quelle der echten +Namen ist :class:`HwSource`: + + - die Anlagen-JSON aus merge_hw_layout (``anlage``, Stufe 4 in + doc/Konzept_Layout_Hardware_SCL.md): Zuordnungen je TRO-ID + Slot + Rolle + (Separator, Lagerlinie, Weiche, Antrieb) und die Foerderernummer je + Layout-Foerderer (Kreisel-Freigabe, Staubereichs-DB); + - oder, auslaufend, die Zuordnung aus lib/e_planning.py (``eplan``): + nur Separator-Sensor und Stopperventil, je Layout-ID. + +Ein eingesetzter Name loescht ``todo``; alles andere bleibt offen. Die +Namensregeln aus Schritt 4 (cfg/scl_naming.ini) arbeiten spaeter genauso: +Wert ersetzen, Merker loeschen. """ from __future__ import annotations @@ -32,7 +40,7 @@ from typing import TYPE_CHECKING from pydantic import BaseModel, ConfigDict, Field -from model import JamArea, TroBase, TroDocument +from model import AnlageDocument, ConveyorBinding, JamArea, TroBase, TroDocument from tro_catalog import TroItem if TYPE_CHECKING: @@ -52,6 +60,7 @@ __all__ = [ "SclStore", "SclTro", "SclFile", + "HwSource", "build_context", "flow_order", "reachable_via", @@ -81,10 +90,91 @@ def _bekannt(wert: str) -> SclName: # Rolle in e_planning.ini -> wie der Name in der SCL geschrieben wird -# (Sensor als Konstante der DB_Inputs-Sensorliste, Ventil als Variable) +# (Sensor als Konstante der DB_Inputs-Sensorliste, Ventil als Variable). +# Die Anlagen-JSON bringt die Konstante aus dem TIA-Import mit; die Vorlage +# gilt nur, wenn sie fehlt (alte eplan-Zuordnung). _SENSOR = "cIn{bmk}" _VENTIL = "{bmk}" +# Antriebs-Slot (model._DRIVE_ROLES) -> Parameter des Vario-Aufrufs +_VARIO_SENSOR_ROLES = { + "stSenCarrInPos": "senCarrInPos", + "stSenFinger": "senFinger", + "stSenLastPos": "senLastPos", + "stSenJam": "senJam", +} + + +class HwSource: + """Namen aus der Hardwareplanung - eine Abfrage fuer beide Quellen. + + ``anlage`` (merge_hw_layout): Schluessel TRO-ID + Slot + Rolle, nur + Zuordnungen mit Status ``zugeordnet`` zaehlen; ``offen``/``widerspruch`` + bleiben ``TODO(E-Planung)``. ``eplan`` (e_planning.py, auslaufend): + Schluessel Layout-ID, kennt nur Separator-Sensor und -Stopper. Ohne beide + liefert jede Abfrage None. + """ + + def __init__(self, anlage: AnlageDocument | None = None, + eplan: EplanDocument | None = None) -> None: + self.anlage = anlage + self.eplan = eplan + self._kinds = ({p.layout_id: p.kind for p in anlage.layout.parts} + if anlage is not None else {}) + + def device(self, tro_id: str, slot: str, role: str, + layout: str = "") -> tuple[str, str] | None: + """(BMK, Konstante) einer Slot-Rolle; Konstante "" wenn unbekannt.""" + if self.anlage is not None: + b = self.anlage.binding(tro_id, slot, role) + if b is None or b.status != "zugeordnet" or not b.bmk: + return None + return b.bmk, b.constant + if self.eplan is not None and layout and layout != "?": + a = self.eplan.for_layout(layout).get(role) + return (a.bmk, "") if a is not None else None + return None + + def sensor(self, tro_id: str, slot: str, role: str, layout: str = "", + offen: str = "cInBG____") -> SclName: + """Sensor als Konstante der DB_Inputs-Sensorliste (cInBG0532).""" + geraet = self.device(tro_id, slot, role, layout) + if geraet is None: + return _offen(offen) + bmk, konstante = geraet + return _bekannt(konstante or _SENSOR.format(bmk=bmk)) + + def output(self, tro_id: str, slot: str, role: str, layout: str = "", + offen: str = "MB____", fmt: str = _VENTIL) -> SclName: + """Ausgang/Variable ueber den BMK (MB0533, "FC0202" ...).""" + geraet = self.device(tro_id, slot, role, layout) + return _bekannt(fmt.format(bmk=geraet[0])) if geraet else _offen(offen) + + def conveyor(self, host: str) -> ConveyorBinding | None: + """Foerderer der Hardwareplanung (K0102) am Layout-Foerderer ("0010-R").""" + if self.anlage is None or not host: + return None + return self.anlage.conveyor_for(host) + + def is_carousel(self, host: str) -> bool: + return self._kinds.get(host.split("-", 1)[0]) == "Kreisel" + + @property + def source(self) -> str: + """Fuer den Dateikopf; "" ohne Hardwareplanung.""" + if self.anlage is not None: + return (f"{self.anlage.hw_file or '-'} + {self.anlage.layout_file or '-'} " + f"(Anlagen-JSON vom {self.anlage.created})") + if self.eplan is not None: + return f"{self.eplan.eplan_file} (Zuordnung vom {self.eplan.created})" + return "" + + +def _foerderer_nr(conveyor: ConveyorBinding) -> str: + """K0102 / #106 -> "0102" bzw. "0106" (Schreibweise der Referenz-DBs).""" + ziffern = re.sub(r"\D", "", conveyor.eplan_name) + return ziffern.zfill(4) if ziffern else "" + class SclSeparator(BaseModel): """Ein stInSeparator-Block.""" @@ -318,6 +408,8 @@ class SclFile(BaseModel): tros: list[SclTro] = Field(default_factory=list) # Zuordnung zur E-Planung, aus der Namen eingesetzt wurden ("" = keine) eplan_source: str = "" + # Anlagen-JSON (merge_hw_layout), aus der Namen eingesetzt wurden ("" = keine) + hw_source: str = "" # --------------------------------------------------------------------------- @@ -432,27 +524,32 @@ def _jam(n: int, neighbour: str, jam_db: str, sensor: SclName | None = None, ) -def _store(tro: TroBase, eplan: EplanDocument | None = None) -> SclStore: +def _store(tro: TroBase, hw: HwSource) -> SclStore: """PinStore_Auto: eine Linie je Speicherbahn (Host) des Layouts. tro_flow legt die Separatoren als [steuernde ..., Linien ...] ab; die Linien sind die letzten len(hosts). Fehlen Hosts (fremde JSON), zaehlt - jeder Separator als Linie. Sensor und Stopper jeder Linie (und des - Pin-Separators, falls tro_flow einen gefunden hat) kommen aus der - E-Planung wie bei den Separatorbloecken. + jeder Separator als Linie. Sensor und Stopper jeder Linie (Slot + ``line``) und des Pin-Separators (Slot ``separator1``, falls tro_flow + einen steuernden Separator gefunden hat) kommen aus der Hardwareplanung + wie bei den Separatorbloecken. """ hosts = list(getattr(tro, "hosts", []) or []) seps = list(tro.separators) anzahl = len(hosts) if 0 < len(hosts) <= len(seps) else len(seps) steuer, linien = seps[:len(seps) - anzahl], seps[len(seps) - anzahl:] - pin_sensor, pin_stopper = _sep_namen(steuer[0], eplan) if steuer else ( - _offen("cInBG____"), _offen("MB____")) + if steuer: + pin_sensor, pin_stopper = _sep_namen(tro.tro_id, "separator1", steuer[0], hw) + else: + pin_sensor, pin_stopper = _offen("cInBG____"), _offen("MB____") zeilen = [] for n, sep in enumerate(linien, start=1): - sensor, stopper = _sep_namen(sep, eplan) + slot = f"line{n}" + sensor, stopper = _sep_namen(tro.tro_id, slot, sep, hw) zeilen.append(SclStoreLine(n=n, layout=sep, jam_length=_offen('"cStorage____ExitJam"'), - sensor=sensor, jam_sensor=_offen("cInBG____"), + sensor=sensor, + jam_sensor=hw.sensor(tro.tro_id, slot, "jamSensor"), stopper=stopper)) return SclStore( encoder=_offen("#fbEncoder____"), @@ -477,35 +574,51 @@ def _store(tro: TroBase, eplan: EplanDocument | None = None) -> SclStore: ) -def _sep_namen(layout: str, eplan: EplanDocument | None) -> tuple[SclName, SclName]: - """Sensor und Stopperventil eines Layout-Separators aus der E-Planung. +def _sep_namen(tro_id: str, slot: str, layout: str, + hw: HwSource) -> tuple[SclName, SclName]: + """Sensor und Stopperventil eines Separators aus der Hardwareplanung. - Gilt fuer die Separatorbloecke (stInSeparator) und genauso fuer die - Linien eines PinStore_Auto - beide sind Separatoren im Layout. Ohne - Zuordnung bleiben beide offen. + Gilt fuer die Separatorbloecke (stInSeparator, Slot ``separator``) + und genauso fuer die Linien eines PinStore_Auto (Slot ``line``) - beide + sind Separatoren im Layout. Ohne Zuordnung bleiben beide offen. """ - rollen = eplan.for_layout(layout) if eplan is not None and layout != "?" else {} - sensor = rollen.get("sensor") - stopper = rollen.get("stopper") - return (_bekannt(_SENSOR.format(bmk=sensor.bmk)) if sensor else _offen("cInBG____"), - _bekannt(_VENTIL.format(bmk=stopper.bmk)) if stopper else _offen("MB____")) + return (hw.sensor(tro_id, slot, "sensor", layout), + hw.output(tro_id, slot, "stopper", layout)) -def _separator(i: int, layout: str, eplan: EplanDocument | None) -> SclSeparator: - sensor, stopper = _sep_namen(layout, eplan) +def _separator(i: int, tro_id: str, layout: str, hw: HwSource) -> SclSeparator: + slot = f"separator{i}" + sensor, stopper = _sep_namen(tro_id, slot, layout, hw) return SclSeparator( n=i, layout=layout, no=_offen("cSep_____"), sensor=sensor, stopper=stopper, - # Teilesensor kennt die E-Planung-Bruecke noch nicht - offen - part_sensor=_offen("cInBG____"), + # Teilesensor: nur die Anlagen-JSON kennt ihn (Rolle partSensor) + part_sensor=hw.sensor(tro_id, slot, "partSensor"), ) +_JAM_DB = '"DB_JamArea____".stJam____' + + +def _jam_db(area: JamArea | None, hw: HwSource) -> str: + """Platzhalter des Staubereichs-DB, mit Foerderernummer soweit bekannt. + + Schema der Referenzen: "DB_JamArea".stJam_. ist der + Foerderer, auf dem der Staubereich liegt (Anlagen-JSON, conveyors[]); + - die laufende Nummer entlang des Flusses auf diesem Foerderer - ist + noch nicht festgelegt (doc/Konzept_Layout_Hardware_SCL.md "Staubereiche"). + Der Name bleibt deshalb immer offen. + """ + conveyor = hw.conveyor(area.host) if area is not None else None + nr = _foerderer_nr(conveyor) if conveyor is not None else "" + return f'"DB_JamArea{nr}".stJam{nr}_____' if nr else _JAM_DB + + def _entries(tro: TroBase, eigene: list[JamArea], - alle: dict[str, TroBase], jam_db: str) -> list[SclJam]: + alle: dict[str, TroBase], hw: HwSource) -> list[SclJam]: """Ein JamEntr je eigenem Staubereich - nicht je Vorgaenger: mehrere Vorgaenger fahren in denselben Staubereich ein (Referenzen UH01-UH05).""" eingaenge = [] @@ -513,65 +626,106 @@ def _entries(tro: TroBase, eigene: list[JamArea], von = ", ".join(_nachbar(v, alle) for v in area.predecessors) or "-" if area.note: von = f"{von} ({area.note})" - eingaenge.append(_jam(area.entry, von, jam_db, jam_id=area.jam_id)) + eingaenge.append(_jam(area.entry, von, _jam_db(area, hw), jam_id=area.jam_id)) return eingaenge def _exit_area(tro_id: str, nach: str, - bereiche: dict[str, list[JamArea]]) -> str: - """Kennung des Staubereichs, in den tro_id beim Nachfolger einfaehrt.""" + bereiche: dict[str, list[JamArea]]) -> JamArea | None: + """Staubereich, in den tro_id beim Nachfolger einfaehrt.""" ziel = bereiche.get(nach, []) passend = [a for a in ziel if tro_id in a.predecessors] or ziel - return passend[0].jam_id if passend else "" + return passend[0] if passend else None + + +def _drive(tro: TroBase, hw: HwSource) -> tuple[SclName, SclName]: + """Motor und Motorschutz eines Vario-Antriebs. + + Erst der Antriebs-Slot (Rollen motor/motorProtection); fehlt er, der + Foerderer der Hardwareplanung am Host des TRO - der Vario-Antrieb ist + dessen Motor (MA/FC-Paar aus hw_import). + """ + motor = hw.output(tro.tro_id, "drive", "motor", offen="MA____") + schutz = hw.output(tro.tro_id, "drive", "motorProtection", + offen='"FC____"', fmt='"{bmk}"') + hosts = list(getattr(tro, "hosts", []) or []) + conveyor = hw.conveyor(hosts[0]) if hosts else None + if conveyor is not None: + if motor.todo and conveyor.motor: + motor = _bekannt(conveyor.motor) + if schutz.todo and conveyor.protection: + schutz = _bekannt(f'"{conveyor.protection}"') + return motor, schutz + + +def _carousel_run(tro: TroBase, hw: HwSource) -> SclName: + """xInCarouselRun1: Freigabe des Kreisels, auf dem der TRO sitzt. + + Bekannt nur, wenn der Host ein Kreisel ist und die Anlagen-JSON dessen + Foerderernummer kennt (Referenz: #fbCarousel0102.xOutConveyorFullSpeed). + """ + hosts = list(getattr(tro, "hosts", []) or []) + host = hosts[0] if hosts else "" + conveyor = hw.conveyor(host) if hw.is_carousel(host) else None + nr = _foerderer_nr(conveyor) if conveyor is not None else "" + if not nr: + return _offen("#fbCarousel____.xOutConveyorFullSpeed") + return _bekannt(f"#fbCarousel{nr}.xOutConveyorFullSpeed") def _tro_context(tro: TroBase, alle: dict[str, TroBase], - eplan: EplanDocument | None = None, + hw: HwSource | None = None, bereiche: dict[str, list[JamArea]] | None = None) -> SclTro: items = tro.items n_sep = items.get(TroItem.SEPARATOR, 1) n_swi = items.get(TroItem.SWITCH, 0) n_scan = items.get(TroItem.SCANNER, 0) tro_id = tro.tro_id - jam_db = '"DB_JamArea____".stJam____' + hw = hw or HwSource() bereiche = bereiche or {} separators = [ - _separator(i, tro.separators[i - 1] if i <= len(tro.separators) else "?", eplan) + _separator(i, tro_id, + tro.separators[i - 1] if i <= len(tro.separators) else "?", hw) for i in range(1, n_sep + 1) ] switches = [ SclSwitch( n=j, no=_offen("cSwi_____"), - valve=_offen("MB____"), + valve=hw.output(tro_id, f"switch{j}", "valve"), # Eine Weiche bedient alle Abgaenge - direkt ableitbar. to1_destinations=len(tro.successors) if n_swi == 1 else None, direction_target=(f"#{tro_id}.nInDirection" if n_swi == 1 else f"#{tro_id}.nInDirection{j}"), - jam_db=_offen(jam_db), + jam_db=_offen(_JAM_DB), ) for j in range(1, n_swi + 1) ] has_vario = bool(items.get(TroItem.VARIO_DRIVE)) vario_sensors = { - name: _offen("cInBG____") - for name in ("stSenCarrInPos", "stSenFinger", "stSenLastPos", "stSenJam") + name: hw.sensor(tro_id, "drive", rolle) + for name, rolle in _VARIO_SENSOR_ROLES.items() } if has_vario else {} + # Nur ein Vario hat einen eigenen Antrieb; sonst bleiben die (ungenutzten) + # Platzhalter offen, statt den Motor des Kreisels zu tragen. + vario_motor, motor_protection = _drive(tro, hw) if has_vario else ( + _offen("MA____"), _offen('"FC____"')) switch_exits = [ SclSwitchExit(via=abgang, targets=reachable_via(tro_id, abgang, alle)) for abgang in tro.successors ] if n_swi else [] # Staubereiche aus der TRO-JSON (jamAreas): Eingaenge sind die eigenen # (ein JamEntr je eigenem Staubereich, nicht je Vorgaenger), Ausgaenge die - # Eingaenge der Nachfolger. Der DB-Name braucht die Foerderernummer aus der - # Hardwareplanung und bleibt offen. - jam_entries = _entries(tro, bereiche.get(tro_id, []), alle, jam_db) - jam_exits = [ - _jam(n, _nachbar(nach, alle), jam_db, sensor=_offen("cInBG____"), - jam_id=_exit_area(tro_id, nach, bereiche)) - for n, nach in enumerate(tro.successors, start=2) - ] + # Eingaenge der Nachfolger. Der DB-Name braucht Foerderernummer (aus der + # Anlagen-JSON) und laufende Nummer (nicht festgelegt) und bleibt offen. + jam_entries = _entries(tro, bereiche.get(tro_id, []), alle, hw) + jam_exits = [] + for n, nach in enumerate(tro.successors, start=2): + area = _exit_area(tro_id, nach, bereiche) + jam_exits.append(_jam(n, _nachbar(nach, alle), _jam_db(area, hw), + sensor=_offen("cInBG____"), + jam_id=area.jam_id if area is not None else "")) return SclTro( id=tro_id, @@ -599,18 +753,18 @@ def _tro_context(tro: TroBase, alle: dict[str, TroBase], main_tro=_offen("cMainTro____"), safety=_offen('"DB_InterfaceSafety".stFromSafety.EStopMain.Q'), release=f'"DB_TRO_TEST".{tro_id}', - motor_protection=_offen('"FC____"'), - vario_motor=_offen("MA____"), + motor_protection=motor_protection, + vario_motor=vario_motor, # Schaltschrank ist anlagenspezifisch ("...OF1" in der Referenz) machine_state=_offen("#stInOutControlUnitCabinet"), priority_udt=_offen("udt_____"), # Vorschlag aus der Referenz TRO107, Prioritaet bleibt Planungsentscheidung prio_dir2=_offen("TRUE"), - carousel_run=_offen("#fbCarousel____.xOutConveyorFullSpeed"), + carousel_run=_carousel_run(tro, hw), # dasselbe Objekt wie der erste Ein-/Ausgang: ein Name, den die # Namensregeln dort setzen, gilt damit auch hier - jam_in=jam_entries[0] if jam_entries else _jam(1, "-", jam_db), - jam_out=jam_exits[0] if jam_exits else _jam(2, "-", jam_db), + jam_in=jam_entries[0] if jam_entries else _jam(1, "-", _JAM_DB), + jam_out=jam_exits[0] if jam_exits else _jam(2, "-", _JAM_DB), error_quit=_offen('"SF____"'), automatic=_offen("#t_automatik"), is_border=_offen("TRUE"), @@ -622,18 +776,24 @@ def _tro_context(tro: TroBase, alle: dict[str, TroBase], station_release=_offen('#"rTrig_Workstation ____".CLK'), loaded_error_sensor=_offen("cInBG____")) if items.get(TroItem.WORK_STATION) else None), - store=_store(tro, eplan) if items.get(TroItem.STORAGE_LINE) else None, + store=_store(tro, hw) if items.get(TroItem.STORAGE_LINE) else None, ) def build_context(document: TroDocument, source: str, start_id: str = "", created: str | None = None, - eplan: EplanDocument | None = None) -> SclFile: + eplan: EplanDocument | None = None, + anlage: AnlageDocument | None = None) -> SclFile: """TroDocument -> SclFile (TROs in Fliessrichtung). - eplan: Zuordnung Layout -> E-Planung (e_planning.py); ohne sie bleiben - alle Namen offen und die Ausgabe ist die bisherige. + anlage: Anlagen-JSON (merge_hw_layout) - ihr Layout ist ``document``, + ihre Zuordnungen liefern die Namen. eplan: alte Zuordnung Layout -> + E-Planung (e_planning.py), nur ohne ``anlage``. Ohne beide bleiben alle + Namen offen und die Ausgabe ist die bisherige. """ + if anlage is not None and eplan is not None: + raise ValueError("anlage und eplan schliessen sich aus") + hw = HwSource(anlage=anlage, eplan=eplan) alle = {t.tro_id: t for t in document.tros} bereiche: dict[str, list[JamArea]] = {} for area in document.jam_areas_or_derived(): @@ -643,7 +803,7 @@ def build_context(document: TroDocument, source: str, start_id: str = "", source=source, created=created or datetime.now().strftime("%Y-%m-%d %H:%M"), start_reason=startgrund, - tros=[_tro_context(t, alle, eplan, bereiche) for t in geordnet], - eplan_source=(f"{eplan.eplan_file} (Zuordnung vom {eplan.created})" - if eplan is not None else ""), + tros=[_tro_context(t, alle, hw, bereiche) for t in geordnet], + eplan_source=hw.source if eplan is not None else "", + hw_source=hw.source if anlage is not None else "", ) diff --git a/lib/scl_skeleton.py b/lib/scl_skeleton.py index 93d7629..1e7c901 100644 --- a/lib/scl_skeleton.py +++ b/lib/scl_skeleton.py @@ -6,14 +6,19 @@ 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.json (--json) - scl_skeleton.py JSON -> FB_Main.scl <- hier + material_flow.py Objekte -> Materialfluss + tro_flow.py Fluss -> TROs -> _tro.json (--json) + merge_hw_layout.py TRO-JSON + Hardwareplanung -> _anlage.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. +Eingabe ist die TRO-JSON (nur Layout) oder die Anlagen-JSON (Layout plus +Zuordnungen der Hardwareplanung); erkannt am Schema. Beide werden ueber das +gemeinsame Modell lib/model.py (TroDocument / AnlageDocument) 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. Aus der Anlagen-JSON +werden zugeordnete Sensoren, Ventile, Antriebe und Foerderernummern +eingesetzt (scl_context.HwSource); offene Zuordnungen bleiben TODO. Erzeugt wird ein **Geruest**, kein fertiges Programm: je TRO eine REGION mit Instanzaufruf und allen Parameterzeilen in der richtigen Reihenfolge, so wie es @@ -49,11 +54,12 @@ from jinja2 import ( TemplateError, TemplateNotFound, ) -from pydantic import ValidationError +from pydantic import BaseModel, ValidationError from e_planning import EPLAN_SCHEMA, EplanDocument +from hw_import import HardwareDocument from material_flow import env_dir -from model import SCHEMA, TroDocument +from model import ANLAGE_SCHEMA, SCHEMA, AnlageDocument, TroDocument from scl_context import TODO, SclFile, SclName, SclTro, build_context # Parametrisierte Aufrufstellen je FB (doc/SCL_Templates.md); Dateiname = fbType. @@ -336,6 +342,13 @@ def render_context(context: SclFile, templates: SclTemplates | None = None) -> s add("// Separator-Sensoren und Stopperventile aus der Zuordnung sind") add("// eingesetzt (ohne Marker); was die Zuordnung nicht sicher fand,") add(f"// bleibt '{TODO}'. Details: _eplan.md (lib/e_planning.py).") + if context.hw_source: + add("//") + add(f"// Hardwareplanung: {context.hw_source}") + add("// Zugeordnete Sensoren, Ventile, Antriebe und Foerderernummern") + add("// sind eingesetzt (ohne Marker); offene Zuordnungen und") + add(f"// Widersprueche bleiben '{TODO}'. Details: _anlage.md") + add("// (lib/merge_hw_layout.py).") if templates is not None: mit, ohne = template_usage(context, templates) add("//") @@ -386,13 +399,23 @@ def main(argv: list[str] | None = None) -> int: file=sys.stderr) return 1 + schema = _schema_of(json_file) + anlage = None try: - document = load_document(json_file) + if schema == ANLAGE_SCHEMA: + anlage = AnlageDocument.from_json(json_file.read_text(encoding="utf-8")) + document = anlage.layout + else: + document = load_document(json_file) except OSError as exc: print(f"FEHLER: {json_file} nicht lesbar: {exc}", file=sys.stderr) return 1 except ValidationError as exc: - schema = _schema_of(json_file) + if schema == ANLAGE_SCHEMA: + print(f"FEHLER: {json_file} passt nicht zum Anlagen-Modell (lib/model.py):", + file=sys.stderr) + print(exc, file=sys.stderr) + return 1 if schema and schema != SCHEMA: # Anderes Format (z. B. die alte tro_extract-JSON) - die einzelnen # Feldfehler helfen dann nicht weiter. @@ -417,6 +440,11 @@ def main(argv: list[str] | None = None) -> int: return 1 templates = SclTemplates(ordner) + if anlage is not None and args.eplan: + print("FEHLER: --eplan geht nicht zusammen mit einer Anlagen-JSON - die " + "Zuordnungen stehen schon darin (merge_hw_layout).", file=sys.stderr) + return 1 + eplan = None if args.eplan: eplan_file = _in_results(args.eplan, results) @@ -438,12 +466,39 @@ def main(argv: list[str] | None = None) -> int: f"TRO-JSON zu '{document.source}'.", file=sys.stderr) return 1 - context = build_context(document, json_file.name, args.start, eplan=eplan) + context = build_context(document, json_file.name, args.start, eplan=eplan, + anlage=anlage) mit, ohne = template_usage(context, templates) if args.strict and ohne: print(f"FEHLER: kein Template fuer {', '.join(ohne)} (--strict) - " f"keine SCL geschrieben.", file=sys.stderr) return 1 + + # Gegenpruefung: jeder eingesetzte Name muss in der Hardwareplanung stehen + # (doc/Konzept_Layout_Hardware_SCL.md, Stufe 4). Die Namen kommen selbst aus + # ihr - ein Fehler hier heisst: Handkorrektur (hw_overrides.ini) mit + # Tippfehler oder Anlagen-JSON zu einem anderen Stand der Hardwareplanung. + hw_file = None + unbekannt: list[str] = [] + if anlage is not None and (args.hw or anlage.hw_file): + hw_file = _in_results(args.hw or Path(anlage.hw_file).name, results) + try: + hw = HardwareDocument.from_json(hw_file.read_text(encoding="utf-8")) + except (OSError, ValidationError) as exc: + if args.hw: + print(f"FEHLER: Hardwareplanung nicht lesbar: {exc}", file=sys.stderr) + return 1 + print(f"Hinweis: {hw_file} nicht lesbar - Namen nicht gegengeprueft.") + hw_file = None + else: + unbekannt = unknown_names(context, hw) + if unbekannt: + stufe = "FEHLER" if args.strict else "WARNUNG" + print(f"{stufe}: {len(unbekannt)} eingesetzte Namen fehlen in {hw_file.name}: " + f"{', '.join(unbekannt)}", file=sys.stderr) + if args.strict: + print(" (--strict) - keine SCL geschrieben.", file=sys.stderr) + return 1 try: text = render_context(context, templates) except TemplateError as exc: @@ -451,7 +506,8 @@ def main(argv: list[str] | None = None) -> int: # als eine mit stillen Luecken. print(f"FEHLER im Template: {exc}", file=sys.stderr) return 1 - out_file = results / (args.out or f"{json_file.stem.replace('_tro', '')}_FB_Main.scl") + stamm = json_file.stem.replace("_anlage", "").replace("_tro", "") + out_file = results / (args.out or f"{stamm}_FB_Main.scl") out_file.write_text(text, encoding="utf-8") offen = text.count(TODO) @@ -469,6 +525,11 @@ def main(argv: list[str] | None = None) -> int: if eplan is not None: print(f"E-Planung = {eplan_file}") print(f"Eingesetzt = {_eingesetzt(context)} Namen aus der E-Planung") + if anlage is not None: + print(f"Hardwareplanung = {anlage.hw_file or '-'}") + print(f"Eingesetzt = {len(known_names(context))} Namen aus der Anlagen-JSON") + if hw_file is not None: + print(f"Gegenpruefung = {len(unbekannt)} unbekannte Namen ({hw_file.name})") print(f"Offene Stellen = {offen} (Marker '{TODO}')") print(f"SCL = {out_file}") print("================================================================") @@ -498,6 +559,41 @@ def _eingesetzt(context: SclFile) -> int: return sum(not name.todo for name in namen) +def known_names(context: SclFile) -> list[str]: + """Alle eingesetzten (nicht offenen) Namen des Kontexts, ohne Doppelte.""" + namen: list[str] = [] + + def walk(obj) -> None: + if isinstance(obj, SclName): + if not obj.todo and obj.value not in namen: + namen.append(obj.value) + elif isinstance(obj, BaseModel): + for feld in type(obj).model_fields: + walk(getattr(obj, feld)) + elif isinstance(obj, (list, tuple)): + for x in obj: + walk(x) + elif isinstance(obj, dict): + for x in obj.values(): + walk(x) + + walk(context) + return namen + + +def unknown_names(context: SclFile, hw: HardwareDocument) -> list[str]: + """Eingesetzte Namen, die die Hardwareplanung nicht kennt. + + Geprueft werden Symbole und Konstanten (cInBG0532, MB0533, "FC0202"); + Instanzzugriffe wie #fbCarousel0102.xOut... sind keine globalen Namen. + """ + bekannt = {n for d in hw.devices for n in (d.bmk, d.constant) if n} + bekannt |= {c.name for c in hw.constants} + bekannt |= {n for d in hw.drives for n in (d.motor, d.protection, d.encoder) if n} + return [name for name in known_names(context) + if not name.startswith("#") and name.strip('"') not in bekannt] + + def _schema_of(path: Path) -> str: """Schema-Kennung einer JSON-Datei, nur fuer die Fehlermeldung ("" wenn unlesbar).""" try: @@ -509,26 +605,30 @@ def _schema_of(path: Path) -> str: def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser( prog="scl_skeleton", - description="Erzeugt aus der TRO-JSON (tro_flow --json) ein FB_Main-Geruest " - "als SCL zum Import in TIA Portal.", + description="Erzeugt aus der TRO-JSON (tro_flow --json) oder der " + "Anlagen-JSON (merge_hw_layout) ein FB_Main-Geruest als SCL " + "zum Import in TIA Portal.", formatter_class=argparse.RawDescriptionHelpFormatter, epilog="Beispiele:\n" " scl_skeleton.bat\n" " scl_skeleton.bat --json export_tro.json --out FB_Main.scl\n" " scl_skeleton.bat --json mubea_tro.json --templates\n" + " scl_skeleton.bat --json mubea_anlage.json --templates\n" " scl_skeleton.bat --json mubea_tro.json --eplan mubea_eplan.json\n" "\n" "Exit-Codes:\n" " 0 ok (auch wenn TROs ohne Template generisch erzeugt wurden)\n" " 1 Eingabe-, Aufruf- oder Templatefehler; mit --strict auch\n" - " ein FB-Typ ohne Template\n", + " ein FB-Typ ohne Template oder ein eingesetzter Name, der\n" + " in der Hardwareplanung fehlt\n", ) parser.add_argument( "--json", default="export_tro.json", metavar="NAME", - help="TRO-JSON aus tro_flow.py --json in %%SKEL_RESULTS%% " - "(oder ein vollstaendiger Pfad). Standard: %(default)s", + help="TRO-JSON aus tro_flow.py --json oder Anlagen-JSON aus " + "merge_hw_layout.py in %%SKEL_RESULTS%% (oder ein vollstaendiger " + "Pfad); erkannt am Schema. Standard: %(default)s", ) parser.add_argument( "--out", @@ -555,15 +655,24 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser.add_argument( "--eplan", metavar="NAME", - help="Zuordnung zur E-Planung aus e_planning.py (_eplan.json in " - "%%SKEL_RESULTS%% oder ein Pfad): Separator-Sensoren und " - "Stopperventile werden eingesetzt statt als TODO markiert.", + help="Auslaufend, nur mit einer TRO-JSON: Zuordnung zur E-Planung aus " + "e_planning.py (_eplan.json in %%SKEL_RESULTS%% oder ein " + "Pfad): Separator-Sensoren und Stopperventile werden eingesetzt " + "statt als TODO markiert. Mit der Anlagen-JSON nicht noetig.", + ) + parser.add_argument( + "--hw", + metavar="NAME", + help="Nur mit einer Anlagen-JSON: Hardwareplanung (_hw.json) " + "fuer die Gegenpruefung der eingesetzten Namen. Standard: hwFile " + "aus der Anlagen-JSON, in %%SKEL_RESULTS%%.", ) parser.add_argument( "--strict", action="store_true", - help="Nur mit --templates: abbrechen (Exit 1), wenn ein FB-Typ der " - "Anlage kein Template hat.", + help="Abbrechen (Exit 1), wenn ein FB-Typ der Anlage kein Template " + "hat (mit --templates) oder ein eingesetzter Name in der " + "Hardwareplanung fehlt (mit Anlagen-JSON).", ) return parser.parse_args(argv) diff --git a/tests/test_scl_anlage.py b/tests/test_scl_anlage.py new file mode 100644 index 0000000..ae9af06 --- /dev/null +++ b/tests/test_scl_anlage.py @@ -0,0 +1,183 @@ +# -*- coding: utf-8 -*- +"""Tests fuer scl_skeleton mit der Anlagen-JSON (Stufe 4). + +Die Anlagen-JSON (merge_hw_layout) traegt das Layout und die Zuordnungen der +Hardwareplanung je TRO-ID + Slot + Rolle. scl_context.HwSource setzt daraus +die Namen ein; was offen oder widerspruechlich ist, bleibt TODO(E-Planung). +Die Zuordnungen sind hier von Hand gebaut - wie merge_hw_layout sie findet, +pruefen die Tests in test_merge_hw_layout.py. +""" + +import scl_skeleton +from hw_import import HardwareDocument, HwDevice +from model import ( + AnlageDocument, + Analysis, + Binding, + ConveyorBinding, + DerivedTro, + LayoutPart, + TroDocument, +) +from scl_context import TODO, build_context + + +def _tro(tro_id, type_name, seps, hosts, **extra): + return DerivedTro(tro_id=tro_id, type_name=type_name, label=tro_id, + hosts=list(hosts), separators=list(seps), scanners=[], + reason="Test", **extra) + + +def _part(layout_id, kind, name): + return LayoutPart(layout_id=layout_id, kind=kind, name=name, + position=(0.0, 0.0, 0.0)) + + +def _bind(tro, slot, role, bmk="", constant="", status="zugeordnet"): + return Binding(tro=tro, slot=slot, role=role, bmk=bmk, constant=constant, + status=status) + + +def anlage() -> AnlageDocument: + """SYS1 -> TRO01 (1Sep1Swi, Kreisel3) -> TRO02 (1Sep, Kreisel3) / TRO03 (Vario, VF_1).""" + tros = [ + _tro("TRO01", "1Sep1Swi", ["0042"], ["0010-R"], predecessors=["SYS1"], + successors=["TRO02", "TRO03"]), + _tro("TRO02", "1Sep", ["0043"], ["0010-L"], predecessors=["TRO01"], + successors=["SYS2"]), + _tro("TRO03", "Vario", ["0062"], ["0015"], predecessors=["TRO01"]), + ] + parts = [_part("0010", "Kreisel", "Kreisel3"), _part("0015", "Strecke", "VF_1")] + layout = TroDocument.from_analysis(Analysis(tros=tros), source="anlage.csv", + parts=parts) + return AnlageDocument( + layout_file="anlage_tro.json", + hw_file="Anlage_hw.json", + layout=layout, + bindings=[ + _bind("TRO01", "separator1", "sensor", "BG0532", "cInBG0532"), + _bind("TRO01", "separator1", "stopper", "MB0533", "cOutMB0533"), + _bind("TRO01", "switch1", "valve", "MB0540", "cOutMB0540"), + # Widerspruch: Name ist da, aber nicht verlaesslich -> bleibt offen + _bind("TRO02", "separator1", "sensor", "BG0600", "cInBG0600", + status="widerspruch"), + _bind("TRO02", "separator1", "stopper", status="offen"), + _bind("TRO03", "drive", "senJam", "BG0900", "cInBG0900"), + ], + conveyors=[ + ConveyorBinding(eplan_name="K0102", layout_id="0010", layout_name="Kreisel3", + motor="MA0202", protection="FC0202"), + ConveyorBinding(eplan_name="K0106", layout_id="0015", layout_name="VF_1", + motor="MA0210", protection="FC0210"), + ], + ) + + +def _by_id(context, tro_id): + return next(t for t in context.tros if t.id == tro_id) + + +def _kontext(doc: AnlageDocument): + return build_context(doc.layout, "anlage.json", created="2026-01-01 00:00", + anlage=doc) + + +def test_separator_und_weiche_aus_den_zuordnungen(): + tro = _by_id(_kontext(anlage()), "TRO01") + sep = tro.separators[0] + assert (sep.sensor.value, sep.sensor.todo) == ("cInBG0532", False) + assert (sep.stopper.value, sep.stopper.todo) == ("MB0533", False) + assert (tro.switches[0].valve.value, tro.switches[0].valve.todo) == ("MB0540", False) + # Rolle ohne Zuordnung bleibt offen + assert sep.part_sensor.todo + + +def test_offen_und_widerspruch_bleiben_todo(): + sep = _by_id(_kontext(anlage()), "TRO02").separators[0] + assert (sep.sensor.value, sep.sensor.todo) == ("cInBG____", True) + assert (sep.stopper.value, sep.stopper.todo) == ("MB____", True) + + +def test_kreisel_freigabe_und_staubereich_aus_der_foerderernummer(): + ctx = _kontext(anlage()) + tro01 = _by_id(ctx, "TRO01") + assert tro01.carousel_run.value == "#fbCarousel0102.xOutConveyorFullSpeed" + assert not tro01.carousel_run.todo + # Staubereich: Foerderer bekannt, laufende Nummer nicht -> weiter offen + assert tro01.jam_entries[0].db.value == '"DB_JamArea0102".stJam0102_____' + assert tro01.jam_entries[0].db.todo + # Ausgang zu TRO03 liegt auf VF_1 (K0106) + assert tro01.jam_exits[1].db.value == '"DB_JamArea0106".stJam0106_____' + # TRO03 sitzt auf einer Strecke, nicht am Kreisel -> Freigabe bleibt offen + assert _by_id(ctx, "TRO03").carousel_run.todo + + +def test_vario_antrieb_ueber_den_foerderer(): + tro = _by_id(_kontext(anlage()), "TRO03") + assert (tro.vario_motor.value, tro.vario_motor.todo) == ("MA0210", False) + assert (tro.motor_protection.value, tro.motor_protection.todo) == ('"FC0210"', False) + assert tro.vario_sensors["stSenJam"].value == "cInBG0900" + assert tro.vario_sensors["stSenFinger"].todo + + +def test_ohne_hardwareplanung_alles_offen(): + doc = anlage() + ctx = build_context(doc.layout, "anlage_tro.json") + tro = _by_id(ctx, "TRO01") + assert tro.separators[0].sensor.todo and tro.carousel_run.todo + assert ctx.hw_source == "" + + +def _hw(*ohne: str) -> HardwareDocument: + namen = [("BG0532", "cInBG0532"), ("MB0533", "cOutMB0533"), + ("MB0540", "cOutMB0540"), ("BG0900", "cInBG0900"), + ("MA0210", ""), ("FC0210", "")] + return HardwareDocument(devices=[HwDevice(bmk=b, constant=c) + for b, c in namen if b not in ohne]) + + +def _lauf(tmp_path, monkeypatch, *weitere: str, hw: HardwareDocument | None = None): + (tmp_path / "anlage_anlage.json").write_text(anlage().to_json(), encoding="utf-8") + (tmp_path / "Anlage_hw.json").write_text((hw or _hw()).to_json(), encoding="utf-8") + monkeypatch.setenv("SKEL_RESULTS", str(tmp_path)) + rc = scl_skeleton.main(["--json", "anlage_anlage.json", *weitere]) + return rc, tmp_path / "anlage_FB_Main.scl" + + +def test_scl_skeleton_liest_die_anlagen_json(tmp_path, monkeypatch, capsys): + rc, scl = _lauf(tmp_path, monkeypatch, "--templates") + assert rc == 0 + text = scl.read_text(encoding="utf-8") + assert "// Hardwareplanung: Anlage_hw.json + anlage_tro.json" in text + assert '#TRO01.stInSeparator1.stSenInSep := "DB_Inputs".Sensors["cInBG0532"];\n' in text + assert 'xOutStopper1 => "MB0533",\n' in text + assert "xInCarouselRun1 := #fbCarousel0102.xOutConveyorFullSpeed," in text + assert "0 unbekannte Namen" in capsys.readouterr().out + + +def test_weniger_offene_stellen_als_mit_der_tro_json(tmp_path, monkeypatch): + rc, scl = _lauf(tmp_path, monkeypatch, "--templates") + mit = scl.read_text(encoding="utf-8").count(TODO) + (tmp_path / "anlage_tro.json").write_text(anlage().layout.to_json(), encoding="utf-8") + assert scl_skeleton.main(["--json", "anlage_tro.json", "--templates", + "--out", "ohne.scl"]) == 0 + ohne = (tmp_path / "ohne.scl").read_text(encoding="utf-8").count(TODO) + assert rc == 0 and mit < ohne + + +def test_eplan_und_anlagen_json_schliessen_sich_aus(tmp_path, monkeypatch): + rc, scl = _lauf(tmp_path, monkeypatch, "--eplan", "x_eplan.json") + assert rc == 1 + assert not scl.exists() + + +def test_unbekannter_name_warnt_und_bricht_mit_strict_ab(tmp_path, monkeypatch, capsys): + rc, scl = _lauf(tmp_path, monkeypatch, hw=_hw("MB0540")) + assert rc == 0 and scl.exists() + assert "WARNUNG: 1 eingesetzte Namen fehlen in Anlage_hw.json: MB0540" in \ + capsys.readouterr().err + scl.unlink() + rc, scl = _lauf(tmp_path, monkeypatch, "--strict", hw=_hw("MB0540")) + assert rc == 1 + assert not scl.exists() +