diff --git a/docs/design/agentic-harness.md b/docs/design/agentic-harness.md index 7bb9c1a8..b7132267 100644 --- a/docs/design/agentic-harness.md +++ b/docs/design/agentic-harness.md @@ -232,7 +232,7 @@ What is left, after the scar tissue, is five ideas and one posture. | **`handoffs/*.md`** | 116 files, 100 done; `Goal / Why / Constraints / Definition of done / Out of scope / Result` | **the draft rows that already exist** — a document revision, a trigger, an adjustment, a reason, a severity | `Why` cites its provenance; `Result` carries *Verified* and *Left out, and why*. A proposal's `rationale` and `evidence` are the same two fields | | **`journal/`** | 479 stanzas, append-only, several concurrent writers, no locking — timestamps run backwards and three files carry a duplicate heading | **`ai_turns` and `audit_log`** | Already there and already better: indexed, pruned to a stated horizon, gated on `audit.read`, and one writer per row | | **`reports/fitness-*.md`** | the crew measuring itself weekly | **the eval suite and `fsmes assist coverage`** | Both exist. Neither is on the AI tab yet (§8) | -| **`bin/crew-guard`** | seven mechanical refusals | **the capability checks, and `shadow.REGISTER`'s 36 entries** | The lesson, not the mechanism: *every rule in the guard exists because prose failed first.* The role file said "you do not loop" and the agent looped for four days | +| **`bin/crew-guard`** | seven mechanical refusals | **the capability checks, and `shadow.REGISTER`'s 38 entries** | The lesson, not the mechanism: *every rule in the guard exists because prose failed first.* The role file said "you do not loop" and the agent looped for four days | | **panes, sessions, `--resume`, permission modes, `clear_trust_dialog`** | ~80 of `crew-up`'s 200 lines | **nothing** | This is the "not exactly". A server process has one long-lived loop already (`api/app.py`'s lifespan) and needs none of it | ### The five ideas worth taking, in order diff --git a/docs/operate/scenarios.md b/docs/operate/scenarios.md new file mode 100644 index 00000000..5e3abfa5 --- /dev/null +++ b/docs/operate/scenarios.md @@ -0,0 +1,183 @@ +# Run a product through a line by writing tags + +A scenario is a file of "at this time, this tag takes this value". `fsmes +scenario play` writes it into an OPC UA server — KEPServerEX, or any other — +so every system that reads that server sees the same thing happen: this MES, +the MES you run today, a script you are testing. One run, two MESs, the same +cans. + +```bash +fsmes scenario check line-trial.json # is the file right? +fsmes scenario check line-trial.json --endpoint opc.tcp://kepware:49320 \ + --security Basic256Sha256,SignAndEncrypt --user mes --password ... +fsmes scenario play line-trial.json --endpoint opc.tcp://kepware:49320 ... +``` + +`play` prints a line per write as the clock reaches it and stops at the end or +on Ctrl-C. `--speed 4` plays four times as fast; `--from 02:00` starts two +minutes in. The connection flags default to the agent's own settings +(`MES_OPC_ENDPOINT`, `MES_OPC_SECURITY`, `MES_OPC_USER`, `MES_OPC_PASSWORD`) +and the same client certificate, so a server that already trusts the agent +trusts this. + +## The safety rule + +Only tags on a **Simulator** or **Memory Based** channel are ever written. A +write to a tag on a PLC driver reaches the PLC, so before writing anything the +player reads every tag it will write and refuses the whole run if any one of +them fails: + +- **Refused** — the tag's channel is not declared in the file's `drivers:` + block as `Simulator` or `Memory Based`; or it is declared as anything else + (a PLC driver, `Advanced Simulator`); or Kepware shows signs of real + equipment behind it (see below), whatever the file says. The refusal names + the channel and the reason. There is no flag to override it. +- **Read-only** — the server does not let this client write the tag. Reported, + never written. +- **Missing** or **type** — the tag is not on the server, or a value in the + file cannot be written as the tag's type (text into a number, 40000 into a + Short, three values into an array of four). + +`fsmes scenario check --endpoint` runs exactly these checks, writes nothing, +and prints one line per tag: type, access, node. + +In shadow mode (`MES_SHADOW=true`) `play` writes nothing at all: shadow mode +promises this MES changes nothing outside its own database, and a scenario +changes tags on another server. Run the player from a shell without it. + +Why the file has to name the driver: Kepware's address space does not say which +driver a channel uses — its System Tags list names none +([Kepware Server help, System Tags](https://support.ptc.com/help/kepware/kepware_server/en/kepware/server/system-tags.html)). +What it does show is the difference between driver classes: a serial driver's +channel has `_System._ComId`, an Ethernet driver's channel has +`_System._NetworkAdapter`, and an Ethernet device's `_System._DeviceId` is its +IP address. The player looks for those three on every Kepware channel it writes, +and any one of them refuses the channel even if the file says `Simulator`. +(A Kepware server is recognised by `KEPServerEX` in its namespace table.) + +## The file + +JSON, one object. Any key starting with `_` is a note and is ignored; any other +unknown key is a mistake and `check` names its line. + +```json +{ + "name": "three cans through the line", + "drivers": {"SimScratch": "Simulator"}, + "tags": { + "ld.good": "ns=2;s=SimScratch.LD.GoodCount", + "ld.serial": "ns=2;s=SimScratch.LD.Serial", + "qi.pass": {"node": "ns=2;s=SimScratch.QI.Pass", "type": "Boolean"} + }, + "steps": [ + {"at": "00:00", "tag": "ld.serial", "value": "TRIAL-1"}, + {"at": "00:10", "tag": "ld.good", "ramp": {"from": 0, "to": 400, "over": 120, "every": 10}} + ], + "product": { + "id": "CAN-{n:04d}", "units": 3, "every_s": 20, "advance_s": 8, "start": "00:02", + "stations": [ + {"station": "LD", "count": "ld.good", "id": "ld.serial"}, + {"station": "QI", "set": {"qi.pass": true}} + ] + } +} +``` + +**`tags`** — a short name for each tag, so the rest of the file reads +`ld.good` and not a node id. A tag is a node id (`ns=2;s=..` +on Kepware) or a browse path from Objects (`LD/FeedRateSP`, in the file's +`namespace`). `type` is read only by `scenario serve`; against a real server +the tag's own type is read from the server. + +**`drivers`** — each channel the scenario writes, and its driver. The channel is +the first part of the address: `SimScratch` in `ns=2;s=SimScratch.LD.GoodCount`, +`LD` in `LD/FeedRateSP`. + +**`steps`** — `at` (seconds, `"mm:ss"` or `"h:mm:ss"` from the start), `tag`, +and either a `value` (a number, true/false, text, or a list for an array tag) +or a `ramp`: `from`, `to`, `over` seconds, a write `every` seconds (default 1). +A ramp between two whole numbers writes whole numbers. + +**`product`** — one block, or a list of them. A product goes through +`stations` in order, `advance_s` seconds at each (a station can give its own +`advance_s`). `units` products enter `every_s` apart, the first at `start`. +At each station, on arrival: + +| key | what is written | +|---|---| +| `count` | the tag goes up by `by` (default 1) **from what it holds now** — read from the server at that moment, never reset | +| `state` | set to `running` (default 1); set to `after`, if given, when the product leaves | +| `id` | set to the product's id: `id` with `{n}` filled in (`CAN-{n:04d}` → `CAN-0001`), numbered from `first` (default 1) | +| `set` | each tag: value given | + +Two writes at the same second go in the order the file gives them. `fsmes +scenario check --timeline` prints every write in order before anything +is played. + +## Rehearse without Kepware + +```bash +fsmes scenario serve labs/kepsim/scenarios/product-through-the-line.json +fsmes scenario play labs/kepsim/scenarios/product-through-the-line.json \ + --endpoint opc.tcp://127.0.0.1:4945/fsmes/scenario +``` + +`serve` holds every tag of the file as a writable memory tag at the file's own +addresses, like a Kepware Memory Based channel, on loopback, anonymous and +unencrypted. Anything that reads OPC UA can watch it. + +Two examples live in [`labs/kepsim/scenarios/`](https://github.com/factorysemantics/factorysemantics-mes/tree/main/labs/kepsim/scenarios): +`product-through-the-line.json` (three cans through the six KepSim stations, +for `serve` or a scratch Kepware channel) and `setpoint-drill.json` (steps and +a ramp on the setpoints of the KepSim replay server, `fsmes run-opc-sim +--replay` — the only tags that server lets a client write; its station tags +are replayed from a file and read-only, and `check` says so). + +## Kepware: what can be written + +From Kepware's own documentation: + +- **Simulator driver** — register (`R`), constant (`K`) and string (`S`) + addresses are Read/Write and hold what is written. **But an `R` register + read as an integer type goes up by one on every read** — it is the driver's + "live data". Put counts and states on `K` addresses (`K0000`, Short or + DWord), text on `S` addresses, and arrays as `K0010 [4]`. The simulation + functions — `RAMP`, `SINE`, `RANDOM`, `USER` — are read-only and keep + generating; the player cannot write them. (Simulator Driver Help, "Address + Descriptions" and "Simulation Functions are Read Only Objects".) +- **Memory Based driver** — a simulated device inside the server that keeps + its tag values, even between server runs (Memory Based Driver Help, + overview). Its help lists a "read only" address error but the access table + was not found; `check --endpoint` reads each tag's real access level, so + your server answers this. +- **Advanced Simulator driver** — tags come from rows of a database over ODBC + (its product sheet). Whether they accept a write was not found in its + documentation; refused here regardless, because the next row overwrites + whatever a scenario writes. +- **Any PLC driver** (Siemens, Allen-Bradley, Modbus, …) — a write reaches the + PLC. Refused here, always. + +Two Kepware habits worth knowing: + +- **The first connection is rejected** until the client certificate is trusted: + OPC UA Configuration Manager → Trusted Clients → select the rejected + certificate → Trust, then reconnect (some versions want the runtime + reinitialised). The certificate is the agent's own + (`certs/mes_twin_client.der`), so trusting it once covers both. +- **Writes carry no timestamp.** Kepware, like many servers, refuses a write + with a source timestamp; the player sends value only. + +## A scratch channel with your own tag names + +So both MESs read the scenario where they already read the line, make a +channel that only the scenario writes: + +1. In the Kepware configuration, **Add Channel → Simulator** (or Memory + Based), named for what it is — `Scratch`, say. +2. Add a device under it with the device name your MESs expect, then a tag per + tag the trial needs, on `K` and `S` addresses, with the data type the real + tag has. +3. Point each MES at the scratch tags for the trial (a tag map in this MES; the + equivalent in the other), and list the channel in the file's `drivers:`. +4. `fsmes scenario check --endpoint ...` until every tag says `write`. +5. `fsmes scenario play --endpoint ...`. diff --git a/labs/kepsim/README.md b/labs/kepsim/README.md index 6921a5e7..3d8b18fc 100644 --- a/labs/kepsim/README.md +++ b/labs/kepsim/README.md @@ -168,6 +168,7 @@ generated at. | `kepware_setup.py` | Config API: channel + devices (`--discover` / `--delete`) | | `verify_readback.py` | OPC UA read-back proof, driven by the MES tag map | | `scenario.md` | what breaks, when, and why it's interesting | +| `scenarios/` | files for `fsmes scenario play`: a product through the six stations, a setpoint drill — see [docs/operate/scenarios.md](../../docs/operate/scenarios.md) | The MES-side pieces live in the main package: `src/fsmes/integrations/opc/csv_replay.py` (the replay server), diff --git a/labs/kepsim/scenarios/product-through-the-line.json b/labs/kepsim/scenarios/product-through-the-line.json new file mode 100644 index 00000000..55d5b179 --- /dev/null +++ b/labs/kepsim/scenarios/product-through-the-line.json @@ -0,0 +1,30 @@ +{ + "_note": "Three cans through the KepSim line's six stations, LD to Palletiser, 20 s apart. The tags sit on a scratch Simulator channel, SimScratch, so the replayed line is never touched. Rehearse with `fsmes scenario serve` (docs/operate/scenarios.md).", + "name": "three cans through the line", + "drivers": {"SimScratch": "Simulator"}, + "tags": { + "ld.state": "ns=2;s=SimScratch.LD.State", + "ld.good": "ns=2;s=SimScratch.LD.GoodCount", + "ld.serial": "ns=2;s=SimScratch.LD.Serial", + "rd.state": "ns=2;s=SimScratch.RD.State", + "rd.good": "ns=2;s=SimScratch.RD.GoodCount", + "wash.state": "ns=2;s=SimScratch.Washer.State", + "wash.good": "ns=2;s=SimScratch.Washer.GoodCount", + "qi.good": "ns=2;s=SimScratch.QI.GoodCount", + "qi.serial": "ns=2;s=SimScratch.QI.Serial", + "qi.pass": {"node": "ns=2;s=SimScratch.QI.Pass", "type": "Boolean"}, + "fill.good": "ns=2;s=SimScratch.Refill.GoodCount", + "pal.good": "ns=2;s=SimScratch.Palletiser.GoodCount" + }, + "product": { + "id": "CAN-{n:04d}", "units": 3, "every_s": 20, "advance_s": 8, "start": "00:02", + "stations": [ + {"station": "LD", "state": "ld.state", "running": 1, "after": 0, "count": "ld.good", "id": "ld.serial"}, + {"station": "RD", "state": "rd.state", "running": 1, "after": 0, "count": "rd.good"}, + {"station": "Washer", "state": "wash.state", "running": 1, "after": 0, "count": "wash.good"}, + {"station": "QI", "count": "qi.good", "id": "qi.serial", "set": {"qi.pass": true}}, + {"station": "Refill", "count": "fill.good"}, + {"station": "Palletiser", "count": "pal.good"} + ] + } +} diff --git a/labs/kepsim/scenarios/setpoint-drill.json b/labs/kepsim/scenarios/setpoint-drill.json new file mode 100644 index 00000000..d325a352 --- /dev/null +++ b/labs/kepsim/scenarios/setpoint-drill.json @@ -0,0 +1,17 @@ +{ + "_note": "A set-and-ramp drill against the KepSim replay server (`fsmes run-opc-sim --replay`): its setpoints are the tags it lets a client write. Feed rate steps down, the washer is ramped up three degrees over a minute and set back. The process values that follow them move on the replay's own lag.", + "name": "setpoint drill", + "namespace": "urn:mes-twin:sim", + "drivers": {"LD": "Simulator", "Washer": "Simulator"}, + "tags": { + "ld.feed": "LD/FeedRateSP", + "wash.temp": "Washer/WashTempSP" + }, + "steps": [ + {"at": "00:00", "tag": "ld.feed", "value": 110.0}, + {"at": "00:05", "tag": "ld.feed", "value": 90.0}, + {"at": "00:10", "tag": "wash.temp", "ramp": {"from": 71.0, "to": 74.0, "over": 60, "every": 10}}, + {"at": "01:20", "tag": "wash.temp", "value": 71.0}, + {"at": "01:30", "tag": "ld.feed", "value": 110.0} + ] +} diff --git a/mkdocs.yml b/mkdocs.yml index 99dcfd23..7672e3bc 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -103,6 +103,7 @@ nav: - Tutorial — deploy: operate/deploy.md - How-to — one plant on two boxes: operate/redundancy.md - How-to — read-only OPC UA commissioning: operate/opc-readonly.md + - How-to — run a product through a line by writing tags: operate/scenarios.md - Explanation — losing sight of a machine: operate/opc-disconnections.md - Explanation — how much of the window did the MES see: operate/coverage.md - Reference — what the floor assistant can do for you: operate/assistant-coverage.md diff --git a/src/fsmes/cli.py b/src/fsmes/cli.py index c846c7ad..f9a099ff 100644 --- a/src/fsmes/cli.py +++ b/src/fsmes/cli.py @@ -2309,6 +2309,209 @@ def maintenance_rules() -> None: f"{_people(trade['people'])}") +scenario_app = typer.Typer( + help="Scenarios: timed tag values written into an OPC UA server (Kepware or any other).") +app.add_typer(scenario_app, name="scenario") + +SCENARIO_ENDPOINT = typer.Option(None, "--endpoint", help="The server, opc.tcp://host:port. " + "Default: MES_OPC_ENDPOINT.") +SCENARIO_SECURITY = typer.Option(None, "--security", help="Policy,mode as for the agent, e.g. " + "Basic256Sha256,SignAndEncrypt. Default: MES_OPC_SECURITY.") +SCENARIO_USER = typer.Option(None, "--user", help="Default: MES_OPC_USER.") +SCENARIO_PASSWORD = typer.Option(None, "--password", help="Default: MES_OPC_PASSWORD.") + + +def _scenario_settings(endpoint, security, user, password) -> Settings: + """The agent's own connection settings, with whatever the command line + overrides: the same certificate, so a server that trusts the agent + trusts this.""" + settings = get_settings() + setup_logging("ERROR", settings.log_dir, "scenario") + update = {key: value for key, value in (("opc_endpoint", endpoint), ("opc_security", security), + ("opc_user", user), ("opc_password", password)) + if value is not None} + return settings.model_copy(update=update) + + +def _scenario_file(file: Path): + from fsmes.integrations.opc import scenario + + try: + return scenario.load(file) + except scenario.ScenarioError as exc: + typer.echo(f"NOT OK {file}: {len(exc.problems)} problem(s)") + for problem in exc.problems: + typer.echo(f" {problem}") + raise typer.Exit(2) from None + + +async def _scenario_connect(settings: Settings): + import logging + + from asyncua import Client + + from fsmes.integrations.opc.security import apply_security, explain_connection_error + + # The library's own chatter (the session timeout it negotiated) is not + # what a person playing a scenario needs to read between the writes. + logging.getLogger("asyncua").setLevel(logging.ERROR) + client = Client(settings.opc_endpoint) + await apply_security(client, settings) + try: + await client.connect() + except Exception as exc: + typer.echo(f"Could not connect to {settings.opc_endpoint}:\n{explain_connection_error(exc)}") + raise typer.Exit(2) from None + return client + + +@scenario_app.command("check") +def scenario_check( + file: Path = typer.Argument(..., help="The scenario file (JSON)."), + endpoint: str = typer.Option(None, "--endpoint", help="Also check against this server: every " + "tag found, its type and access, and the driver rule. Without it, " + "only the file is checked."), + security: str = SCENARIO_SECURITY, + user: str = SCENARIO_USER, + password: str = SCENARIO_PASSWORD, + timeline: bool = typer.Option(False, "--timeline", help="Also print every write, in order."), +) -> None: + """Say what is wrong with a scenario file before it is played. + + Offline, it reads the file and names the line of each mistake. With + --endpoint it connects (reading only, writing nothing) and reports, for + every tag the scenario writes: whether the server has it, its type, whether + it may be written, and whether its channel passes the driver rule — + Simulator and Memory Based channels only, never a PLC driver. + Exit 0 = every tag can be written; 1 = some cannot; 2 = the file is wrong + or the server cannot be reached. + """ + from fsmes.integrations.opc import scenario + + plan = _scenario_file(file) + counts = sum(1 for w in plan.writes if w.increment is not None) + typer.echo(f"OK {file}: {plan.name} — {len(plan.writes)} writes to {len(plan.used())} of " + f"{len(plan.tags)} tags over {scenario.clock(plan.end)}" + + (f" ({counts} are counts going up from what the tag holds)" if counts else "")) + if timeline: + scenario.timeline(plan, typer.echo) + if endpoint is None: + return + + async def go() -> int: + client = await _scenario_connect(_scenario_settings(endpoint, security, user, password)) + try: + targets = await scenario.inspect(client, plan) + finally: + await client.disconnect() + typer.echo(f"Connected to {endpoint}") + return scenario.report(targets, typer.echo) + + raise typer.Exit(1 if asyncio.run(go()) else 0) + + +@scenario_app.command("play") +def scenario_play( + file: Path = typer.Argument(..., help="The scenario file (JSON)."), + endpoint: str = SCENARIO_ENDPOINT, + security: str = SCENARIO_SECURITY, + user: str = SCENARIO_USER, + password: str = SCENARIO_PASSWORD, + speed: float = typer.Option(1.0, "--speed", help="2 plays twice as fast; 0.5 half as fast."), + start: str = typer.Option("0", "--from", help="Start part way in, minutes:seconds (\"02:00\"); " + "writes before it are skipped."), +) -> None: + """Write a scenario's values into the server, each at its time. + + Every tag is checked first, exactly as `fsmes scenario check --endpoint` + does, and nothing is written unless all of them pass: a tag on a channel + not declared Simulator or Memory Based is refused, a read-only tag is + reported. Values are written as the tag's own type. Prints a line per + write; stops at the end or on Ctrl-C. + """ + from fsmes.integrations.opc import scenario + + plan = _scenario_file(file) + if speed <= 0: + typer.echo("--speed must be more than 0.") + raise typer.Exit(2) + try: + offset = scenario.parse_time(start) + except ValueError as exc: + typer.echo(f"--from {exc}") + raise typer.Exit(2) from None + from fsmes import shadow + + if shadow.enabled(): + typer.echo(f"Nothing written: {shadow.BANNER} A scenario writes tags on another " + f"server, so it does not run in shadow mode. {shadow.HOW_TO_LEAVE}") + raise typer.Exit(1) + settings = _scenario_settings(endpoint, security, user, password) + + async def go() -> int: + client = await _scenario_connect(settings) + try: + targets = await scenario.inspect(client, plan) + if any(t.problem for t in targets.values()): + typer.echo(f"Nothing written. {settings.opc_endpoint}:") + scenario.report(targets, typer.echo) + return 1 + typer.echo(f"Playing {plan.name} into {settings.opc_endpoint} at {speed:g}x" + + (f" from {scenario.clock(offset)}" if offset else "") + + f" — {sum(1 for w in plan.writes if w.at >= offset)} writes, " + f"{scenario.clock(max(plan.end - offset, 0) / speed)}. Ctrl-C stops.") + try: + done = await scenario.play(client, plan, targets, speed=speed, start=offset, + out=typer.echo) + except scenario.ScenarioError as exc: + for problem in exc.problems: + typer.echo(problem) + return 1 + typer.echo(f"Done: {done} writes.") + return 0 + finally: + with contextlib.suppress(Exception): + await client.disconnect() + + try: + raise typer.Exit(asyncio.run(go())) + except KeyboardInterrupt: + typer.echo("Stopped (Ctrl-C). The tags keep the last values written.") + raise typer.Exit(130) from None + + +@scenario_app.command("serve") +def scenario_serve( + file: Path = typer.Argument(..., help="The scenario file (JSON)."), + endpoint: str = typer.Option("opc.tcp://127.0.0.1:4945/fsmes/scenario", "--endpoint", + help="Where to listen. Loopback by default."), +) -> None: + """Rehearse without Kepware: serve the scenario's tags as writable memory tags. + + Every tag in the file is served at its own address, writable, like a + Kepware Memory Based channel — so `fsmes scenario play` and anything that + reads the tags (this MES's agent, another MES) can be tried at home. A + tag's type comes from its `type` in the file, else from its first value. + Anonymous and unencrypted: for one machine, not a network. Ctrl-C stops. + """ + from fsmes.integrations.opc import scenario + + plan = _scenario_file(file) + setup_logging("ERROR", get_settings().log_dir, "scenario") + + async def go() -> None: + server = await scenario.rehearsal_server(plan, endpoint) + async with server: + typer.echo(f"Serving {len(plan.tags)} tags of {plan.name} at {endpoint}. Ctrl-C stops.") + while True: + await asyncio.sleep(3600) + + try: + asyncio.run(go()) + except KeyboardInterrupt: + typer.echo("Stopped.") + + layout_app = typer.Typer( help="The 3D line view's layout file: where each machine stands.") app.add_typer(layout_app, name="layout") diff --git a/src/fsmes/integrations/opc/scenario.py b/src/fsmes/integrations/opc/scenario.py new file mode 100644 index 00000000..7468e0fa --- /dev/null +++ b/src/fsmes/integrations/opc/scenario.py @@ -0,0 +1,777 @@ +"""Scenarios: timed tag values written into an OPC UA server. + +A scenario file says "at this time, this tag takes this value", with two +shorthands: a ramp ("climbs from A to B over N seconds") and a product ("goes +through these stations in order, each N seconds apart"). `fsmes scenario play` +writes it into any OPC UA server — Kepware first — so every system reading that +server, this MES or another one, sees the same product go through. + +The file is JSON, read here with the line of every object kept, so a mistake is +reported as "line 14: ..." rather than as a path into a tree. + +The safety rule is in this module, not in the docs. A write to a tag on a real +PLC driver reaches the PLC, so before anything is written every target is read +and judged: + +- its own `UserAccessLevel` must allow a write, or it is reported read-only; +- its channel (the first segment of its address) must be declared in the file's + `drivers:` block as `Simulator` or `Memory Based`, because Kepware's address + space does not name a channel's driver (its System Tags list none) — and + any sign the server gives of a network or serial device behind that channel + refuses it whatever the file says. There is no override. + +`play` writes nothing unless every target passes. +""" + +from __future__ import annotations + +import asyncio +import json +import json.decoder +import json.scanner +import re +import time +from collections.abc import Callable +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from asyncua import Client, Server, ua + +from fsmes import shadow + +SAFE_DRIVERS = {"simulator": "Simulator", "memory based": "Memory Based"} +NODE_ID = re.compile(r"^(ns=\d+;|nsu=[^;]+;)?[isgb]=") + + +class ScenarioError(Exception): + """A scenario file that cannot be played, with every problem found.""" + + def __init__(self, problems: list[str]): + super().__init__("\n".join(problems)) + self.problems = problems + + +# ------------------------------------------------------------------ the file + + +class _Obj(dict): + """A JSON object that remembers the line it started on.""" + + line = 0 + + +def _decoder() -> json.JSONDecoder: + decoder = json.JSONDecoder() + + def parse_object(s_and_end, strict, scan_once, object_hook, object_pairs_hook, memo=None): + s, end = s_and_end + line = s.count("\n", 0, end) + 1 + pairs, end = json.decoder.JSONObject(s_and_end, strict, scan_once, None, list, memo) + obj = _Obj() + for key, value in pairs: + if key in obj: + raise ScenarioError([f"line {line}: {key!r} is given twice in the same block"]) + obj[key] = value + obj.line = line + return obj, end + + decoder.parse_object = parse_object + decoder.scan_once = json.scanner.py_make_scanner(decoder) + return decoder + + +@dataclass +class Tag: + alias: str + address: str + line: int + type: str | None = None # only `scenario serve` reads it + + @property + def is_node_id(self) -> bool: + return bool(NODE_ID.match(self.address)) + + @property + def channel(self) -> str | None: + """The first segment of the address: Kepware's channel name for + `ns=2;s=..`, the top object for a browse path. + A numeric or GUID node id has none.""" + if self.is_node_id: + _, _, ident = self.address.rpartition(";") + if not ident.startswith("s="): + return None + return ident[2:].split(".", 1)[0] or None + return self.address.split("/", 1)[0].split(":", 1)[-1] or None + + +@dataclass +class Write: + at: float + alias: str + line: int + value: Any = None + increment: float | None = None # a count that goes up by this much + why: str = "" # "ramp", "CAN-0001 at LD", ... + + def describe(self) -> str: + return f"+{_num(self.increment)}" if self.increment is not None else _show(self.value) + + +@dataclass +class Scenario: + source: str + name: str + namespace: str | None + tags: dict[str, Tag] + drivers: dict[str, str] + writes: list[Write] = field(default_factory=list) + + @property + def end(self) -> float: + return max((w.at for w in self.writes), default=0.0) + + def used(self) -> list[Tag]: + seen = {w.alias for w in self.writes} + return [tag for alias, tag in self.tags.items() if alias in seen] + + +def load(path: Path) -> Scenario: + """Read and check a scenario file. Raises ScenarioError naming every + problem with its line.""" + path = Path(path) + try: + text = path.read_text(encoding="utf-8") + except OSError as exc: + raise ScenarioError([f"cannot read {path}: {exc.strerror or exc}"]) from None + return parse(text, str(path)) + + +def parse(text: str, source: str = "") -> Scenario: + try: + doc = _decoder().decode(text) + except json.JSONDecodeError as exc: + raise ScenarioError([f"line {exc.lineno}: not valid JSON — {exc.msg}"]) from None + if not isinstance(doc, _Obj): + raise ScenarioError(["line 1: a scenario is one JSON object, { ... }"]) + problems: list[str] = [] + _known(doc, {"name", "namespace", "tags", "drivers", "steps", "product"}, problems) + + tags: dict[str, Tag] = {} + raw_tags = doc.get("tags") + if not isinstance(raw_tags, _Obj) or not raw_tags: + problems.append(f"line {doc.line}: `tags` is missing — name each tag once, " + f'"alias": "ns=2;s=Channel.Device.Tag"') + raw_tags = _Obj() + for alias, spec in raw_tags.items(): + if alias.startswith("_"): + continue + if isinstance(spec, str): + tags[alias] = Tag(alias, spec.strip(), raw_tags.line) + elif isinstance(spec, _Obj) and isinstance(spec.get("node"), str): + _known(spec, {"node", "type"}, problems) + kind = spec.get("type") + if kind is not None and kind not in SERVE_TYPES: + problems.append(f"line {spec.line}: tag {alias!r} type {kind!r} is not one of " + f"{', '.join(SERVE_TYPES)}") + tags[alias] = Tag(alias, spec["node"].strip(), spec.line, kind) + else: + problems.append(f"line {raw_tags.line}: tag {alias!r} must be a node id or a " + f'browse path ("LD/GoodCount"), or {{"node": ..., "type": ...}}') + + drivers: dict[str, str] = {} + raw_drivers = doc.get("drivers", _Obj()) + if not isinstance(raw_drivers, _Obj): + problems.append(f"line {doc.line}: `drivers` is a block of channel: driver name") + else: + for channel, driver in raw_drivers.items(): + if channel.startswith("_"): + continue + if not isinstance(driver, str): + problems.append(f"line {raw_drivers.line}: driver for channel {channel!r} must be text") + continue + drivers[channel] = driver + + scenario = Scenario(source, str(doc.get("name") or Path(source).stem), + doc.get("namespace"), tags, drivers) + + steps = doc.get("steps", []) + if not isinstance(steps, list): + problems.append(f"line {doc.line}: `steps` is a list, [ ... ]") + steps = [] + for step in steps: + _step(step, scenario, problems) + + products = doc.get("product", []) + if isinstance(products, _Obj): + products = [products] + if not isinstance(products, list): + problems.append(f"line {doc.line}: `product` is a block or a list of blocks") + products = [] + for product in products: + _product(product, scenario, problems) + + if not scenario.writes and not problems: + problems.append(f"line {doc.line}: nothing to write — add `steps` or a `product`") + if problems: + raise ScenarioError(problems) + # Stable sort: two writes at the same second go in the order the file gives. + scenario.writes.sort(key=lambda w: w.at) + return scenario + + +def _known(obj: _Obj, allowed: set[str], problems: list[str]) -> None: + for key in obj: + if key not in allowed and not key.startswith("_"): + problems.append(f"line {obj.line}: unknown key {key!r} (expected one of " + f"{', '.join(sorted(allowed))}; a key starting with _ is a note)") + + +def _alias(obj: _Obj, key: str, scenario: Scenario, problems: list[str], required: bool = True): + alias = obj.get(key) + if alias is None: + if required: + problems.append(f"line {obj.line}: `{key}` is missing") + return None + if alias not in scenario.tags: + problems.append(f"line {obj.line}: tag {alias!r} is not in `tags`") + return None + return alias + + +def _value_ok(value: Any) -> bool: + if isinstance(value, bool | int | float | str): + return True + return isinstance(value, list) and bool(value) and all( + isinstance(v, bool | int | float) for v in value) + + +def _step(step: Any, scenario: Scenario, problems: list[str]) -> None: + if not isinstance(step, _Obj): + problems.append("a step is a block: {\"at\": \"00:10\", \"tag\": ..., \"value\": ...}") + return + _known(step, {"at", "tag", "value", "ramp"}, problems) + at = _time(step, "at", problems) + alias = _alias(step, "tag", scenario, problems) + if ("value" in step) == ("ramp" in step): + problems.append(f"line {step.line}: a step has a `value` or a `ramp`, one of the two") + return + if at is None or alias is None: + return + if "value" in step: + if not _value_ok(step["value"]): + problems.append(f"line {step.line}: value {step['value']!r} is not a number, " + f"true/false, text, or a list of numbers") + return + scenario.writes.append(Write(at, alias, step.line, value=step["value"])) + return + ramp = step["ramp"] + if not isinstance(ramp, _Obj): + problems.append(f"line {step.line}: `ramp` is a block: {{\"from\": 0, \"to\": 400, \"over\": 120}}") + return + _known(ramp, {"from", "to", "over", "every"}, problems) + start, stop = ramp.get("from"), ramp.get("to") + if not all(isinstance(v, int | float) and not isinstance(v, bool) for v in (start, stop)): + problems.append(f"line {ramp.line}: a ramp needs numbers for `from` and `to`") + return + over = _time(ramp, "over", problems) + every = _time(ramp, "every", problems, default=1.0) + if over is None or every is None: + return + if over <= 0 or every <= 0: + problems.append(f"line {ramp.line}: `over` and `every` must be more than zero seconds") + return + whole = isinstance(start, int) and isinstance(stop, int) + n = max(1, round(over / every)) + for i in range(n + 1): + value = start + (stop - start) * i / n + value = round(value) if whole else round(value, 6) + scenario.writes.append(Write(at + over * i / n, alias, ramp.line, value=value, why="ramp")) + + +def _product(product: Any, scenario: Scenario, problems: list[str]) -> None: + if not isinstance(product, _Obj): + problems.append("a product is a block with `stations`") + return + _known(product, {"start", "advance_s", "units", "every_s", "id", "first", "stations"}, problems) + start = _time(product, "start", problems, default=0.0) + advance = _time(product, "advance_s", problems) + every = _time(product, "every_s", problems) if "every_s" in product else advance + units, first = product.get("units", 1), product.get("first", 1) + template = product.get("id", "") + stations = product.get("stations") + if not isinstance(units, int) or isinstance(units, bool) or units < 1: + problems.append(f"line {product.line}: `units` is a whole number, 1 or more") + return + if not isinstance(first, int) or isinstance(first, bool): + problems.append(f"line {product.line}: `first` is a whole number") + return + if not isinstance(template, str): + problems.append(f"line {product.line}: `id` is text, like \"CAN-{{n:04d}}\"") + return + try: + template.format(n=first) + except (KeyError, IndexError, ValueError) as exc: + problems.append(f"line {product.line}: `id` {template!r} cannot be filled in ({exc}); " + f"{{n}} is the unit's number") + return + if not isinstance(stations, list) or not stations: + problems.append(f"line {product.line}: a product needs `stations`, in the order it visits them") + return + if start is None or advance is None or every is None: + return + if advance < 0 or every < 0: + problems.append(f"line {product.line}: `advance_s` and `every_s` cannot be negative") + return + + plan = [] + before = len(problems) + for station in stations: + if not isinstance(station, _Obj): + problems.append(f"line {product.line}: each station is a block with a `station` name") + return + _known(station, {"station", "count", "by", "state", "running", "after", "id", "set", + "advance_s"}, problems) + name = station.get("station") + if not isinstance(name, str) or not name: + problems.append(f"line {station.line}: a station needs a `station` name") + continue + count = _alias(station, "count", scenario, problems, required=False) + state = _alias(station, "state", scenario, problems, required=False) + id_tag = _alias(station, "id", scenario, problems, required=False) + by = station.get("by", 1) + if not isinstance(by, int | float) or isinstance(by, bool): + problems.append(f"line {station.line}: `by` is a number") + continue + extra = station.get("set", _Obj()) + if not isinstance(extra, _Obj): + problems.append(f"line {station.line}: `set` is a block of tag: value") + continue + sets = [] + for alias, value in extra.items(): + if alias.startswith("_"): + continue + if alias not in scenario.tags: + problems.append(f"line {extra.line}: tag {alias!r} is not in `tags`") + elif not _value_ok(value): + problems.append(f"line {extra.line}: value {value!r} for {alias!r} is not a number, " + f"true/false, text, or a list of numbers") + else: + sets.append((alias, value)) + for key in ("running", "after"): + if key in station and not _value_ok(station[key]): + problems.append(f"line {station.line}: `{key}` {station[key]!r} is not a value") + if ("running" in station or "after" in station) and "state" not in station: + problems.append(f"line {station.line}: `running`/`after` need a `state` tag") + dwell = _time(station, "advance_s", problems, default=advance) + if not (count or state or id_tag or sets): + problems.append(f"line {station.line}: station {name!r} writes nothing — give it a " + f"`count`, `state`, `id` or `set`") + plan.append((station, name, count, by, state, id_tag, sets, dwell)) + if len(problems) > before: + return + + for k in range(units): + n = first + k + label = template.format(n=n) if template else f"unit {n}" + t = start + k * every + for station, name, count, by, state, id_tag, sets, dwell in plan: + why = f"{label} at {name}" + line = station.line + if id_tag: + value = label if template else str(n) + scenario.writes.append(Write(t, id_tag, line, value=value, why=why)) + if state: + scenario.writes.append(Write(t, state, line, value=station.get("running", 1), why=why)) + for alias, value in sets: + scenario.writes.append(Write(t, alias, line, value=value, why=why)) + if count: + scenario.writes.append(Write(t, count, line, increment=by, why=why)) + if state and "after" in station: + scenario.writes.append(Write(t + (dwell or 0), state, line, value=station["after"], + why=f"{label} leaves {name}")) + t += dwell or 0 + + +def _time(obj: _Obj, key: str, problems: list[str], default: float | None = None) -> float | None: + if key not in obj: + if default is None: + problems.append(f"line {obj.line}: `{key}` is missing") + return default + try: + return parse_time(obj[key]) + except ValueError as exc: + problems.append(f"line {obj.line}: `{key}` {exc}") + return None + + +def parse_time(value: Any) -> float: + """Seconds, or "mm:ss", or "h:mm:ss".""" + if isinstance(value, int | float) and not isinstance(value, bool): + if value < 0: + raise ValueError(f"{value!r} is before the start") + return float(value) + if isinstance(value, str) and re.fullmatch(r"\d+(:\d{1,2}){0,2}(\.\d+)?", value.strip()): + seconds = 0.0 + for part in value.strip().split(":"): + seconds = seconds * 60 + float(part) + return seconds + raise ValueError(f"{value!r} is not a time: seconds (90) or minutes:seconds (\"01:30\")") + + +def clock(seconds: float) -> str: + whole = int(seconds) + text = f"{whole // 60:02d}:{whole % 60:02d}" if whole < 3600 else ( + f"{whole // 3600}:{whole % 3600 // 60:02d}:{whole % 60:02d}") + frac = seconds - whole + return text + (f".{round(frac * 10)}" if frac >= 0.05 else "") + + +def _num(value) -> str: + return str(int(value)) if isinstance(value, float) and value.is_integer() else str(value) + + +def _show(value) -> str: + return json.dumps(value) if isinstance(value, str | bool | list) else _num(value) + + +# ------------------------------------------------------------- the server side + + +INT_RANGES = { + ua.VariantType.SByte: (-2**7, 2**7 - 1), ua.VariantType.Byte: (0, 2**8 - 1), + ua.VariantType.Int16: (-2**15, 2**15 - 1), ua.VariantType.UInt16: (0, 2**16 - 1), + ua.VariantType.Int32: (-2**31, 2**31 - 1), ua.VariantType.UInt32: (0, 2**32 - 1), + ua.VariantType.Int64: (-2**63, 2**63 - 1), ua.VariantType.UInt64: (0, 2**64 - 1), +} +FLOATS = {ua.VariantType.Float, ua.VariantType.Double} + + +@dataclass +class Target: + """What the server says about one tag the scenario writes.""" + + tag: Tag + node: Any = None + node_id: str = "" + variant: ua.VariantType | None = None + array_length: int | None = None # None: a scalar + writable: bool = False + problem: str = "" # empty when the tag may be written + verdict: str = "ok" # ok | missing | read-only | refused | type + + def type_name(self) -> str: + if self.variant is None: + return "?" + name = self.variant.name + return f"{name}[{self.array_length}]" if self.array_length is not None else name + + +def coerce(value: Any, target: Target) -> ua.Variant: + """The value as the node's own type, or ValueError saying why it cannot be. + Kepware answers a wrong variant with BadTypeMismatch, so nothing is left to + the library's guess.""" + vt = target.variant + if vt is None: + raise ValueError("the server did not say this tag's type") + if target.array_length is not None: + if not isinstance(value, list): + raise ValueError(f"this tag is an array of {target.array_length}; give a list") + if target.array_length and len(value) != target.array_length: + raise ValueError(f"this tag holds {target.array_length} values; the list has {len(value)}") + return ua.Variant([_scalar(v, vt) for v in value], vt) + if isinstance(value, list): + raise ValueError(f"this tag holds one {vt.name}, not a list") + return ua.Variant(_scalar(value, vt), vt) + + +def _scalar(value: Any, vt: ua.VariantType) -> Any: + if vt == ua.VariantType.Boolean: + if isinstance(value, bool): + return value + if value in (0, 1): + return bool(value) + raise ValueError(f"{value!r} is not true/false") + if vt == ua.VariantType.String: + if isinstance(value, bool): + raise ValueError(f"{value!r} is not text") + return value if isinstance(value, str) else _num(value) + if vt in INT_RANGES: + if isinstance(value, bool) or not isinstance(value, int | float) or float(value) != int(value): + raise ValueError(f"{value!r} is not a whole number, and this tag is {vt.name}") + low, high = INT_RANGES[vt] + if not low <= int(value) <= high: + raise ValueError(f"{value!r} does not fit {vt.name} ({low} to {high})") + return int(value) + if vt in FLOATS: + if isinstance(value, bool) or not isinstance(value, int | float): + raise ValueError(f"{value!r} is not a number") + return float(value) + raise ValueError(f"this tag's type, {vt.name}, is not one a scenario writes") + + +async def _node_for(client: Client, tag: Tag, namespace: str | None): + if tag.is_node_id: + return client.get_node(tag.address) + idx = None + path = [] + for part in tag.address.split("/"): + if re.match(r"^\d+:", part): + path.append(part) + continue + if idx is None: + if namespace: + idx = await client.get_namespace_index(namespace) + else: + idx = 2 + path.append(f"{idx}:{part}") + return await client.nodes.objects.get_child(path) + + +async def _is_kepware(client: Client) -> bool: + try: + names = await client.get_namespace_array() + except Exception: + return False + return any("kepserver" in str(name).lower() or "kepware" in str(name).lower() for name in names) + + +async def _real_device_sign(client: Client, tag: Tag) -> str | None: + """Ask Kepware for evidence of real equipment behind this tag's channel. + + Kepware names no driver in its address space, but its `_System` tags differ + by driver class: a serial driver's channel has `_ComId`, an Ethernet + driver's has `_NetworkAdapter`, and an Ethernet device's `_DeviceId` is an + IP address. A Simulator or Memory Based channel is neither serial nor + Ethernet. Any such sign refuses the channel, whatever the file declares. + """ + head, _, ident = tag.address.rpartition(";") + parts = ident[2:].split(".") + ns = head + ";" if head else "" + probes = [f"{ns}s={parts[0]}._System._ComId", f"{ns}s={parts[0]}._System._NetworkAdapter"] + for probe in probes: + try: + await client.get_node(probe).read_value() + except Exception: + continue + return f"the server shows {probe.split('s=', 1)[1]}, which only a serial or Ethernet driver has" + if len(parts) >= 3: + probe = f"{ns}s={parts[0]}.{parts[1]}._System._DeviceId" + try: + device_id = await client.get_node(probe).read_value() + except Exception: + return None + if isinstance(device_id, str) and re.search(r"\d+\.\d+\.\d+\.\d+|[A-Za-z].*\.", device_id): + return f"device {parts[1]} is addressed as {device_id!r}, a network address" + return None + + +def judge_driver(tag: Tag, drivers: dict[str, str]) -> str: + """Empty when the file says this tag's channel is a simulator; else why not.""" + channel = tag.channel + if channel is None: + return ("its address has no channel name (a numeric node id), so nobody can say " + "what is behind it; address it by name") + declared = drivers.get(channel) + if declared is None: + return (f"channel {channel!r} is not declared in `drivers`; say which driver it uses — " + f"only Simulator or Memory Based channels are written") + normal = " ".join(declared.lower().replace("-", " ").split()) + if normal in SAFE_DRIVERS: + return "" + if normal == "advanced simulator": + return (f"channel {channel!r} is an Advanced Simulator: its tags are read from a database " + f"over ODBC and cannot be written") + return f"channel {channel!r} uses {declared!r}, which reaches real equipment" + + +async def inspect(client: Client, scenario: Scenario) -> dict[str, Target]: + """Resolve and judge every tag the scenario writes. Reads only.""" + kepware = await _is_kepware(client) + targets: dict[str, Target] = {} + for tag in scenario.used(): + target = Target(tag) + targets[tag.alias] = target + try: + node = await _node_for(client, tag, scenario.namespace) + node_class = await node.read_node_class() + except Exception as exc: + target.verdict, target.problem = "missing", f"not on the server ({_short(exc)})" + continue + target.node, target.node_id = node, node.nodeid.to_string() + if node_class != ua.NodeClass.Variable: + target.verdict, target.problem = "missing", f"is a {node_class.name}, not a tag" + continue + try: + data = await node.read_data_value() + access = (await node.read_attribute(ua.AttributeIds.UserAccessLevel)).Value.Value + data_type = await node.read_data_type() + except Exception as exc: + target.verdict, target.problem = "missing", f"cannot be read ({_short(exc)})" + continue + current = data.Value.Value if data.Value is not None else None + if data.Value is not None and current is not None: + target.variant = data.Value.VariantType + elif data_type.NamespaceIndex == 0 and isinstance(data_type.Identifier, int): + try: + target.variant = ua.VariantType(data_type.Identifier) + except ValueError: + target.variant = None + if isinstance(current, list): + target.array_length = len(current) + target.writable = bool((int(access or 0) >> ua.AccessLevel.CurrentWrite) & 1) + + reason = judge_driver(tag, scenario.drivers) + if not reason and kepware and tag.is_node_id: + reason = await _real_device_sign(client, tag) or "" + if reason: + target.verdict, target.problem = "refused", reason + elif not target.writable: + target.verdict, target.problem = "read-only", "the server allows reading it, not writing" + else: + for write in scenario.writes: + if write.alias != tag.alias: + continue + try: + if write.increment is not None: + if target.variant not in INT_RANGES and target.variant not in FLOATS: + raise ValueError(f"a count must be a number tag; this one is {target.type_name()}") + else: + coerce(write.value, target) + except ValueError as exc: + target.verdict, target.problem = "type", f"line {write.line}: {exc}" + break + return targets + + +def _short(exc: Exception) -> str: + text = str(exc) or type(exc).__name__ + return text.splitlines()[0][:120] + + +def report(targets: dict[str, Target], out: Callable[[str], None] = print) -> int: + """Print one line per tag; return how many cannot be written.""" + bad = 0 + out(f"{'tag':<18} {'type':<12} {'access':<10} node") + for alias, target in targets.items(): + access = "write" if target.writable else ("read" if target.node is not None else "-") + out(f"{alias:<18} {target.type_name():<12} {access:<10} {target.node_id or target.tag.address}") + if target.problem: + bad += 1 + out(f" {target.verdict.upper()}: {target.problem}") + good = len(targets) - bad + out(f"{len(targets)} tags: {good} can be written, {bad} cannot.") + return bad + + +async def play(client: Client, scenario: Scenario, targets: dict[str, Target], *, + speed: float = 1.0, start: float = 0.0, out: Callable[[str], None] = print, + now: Callable[[], float] = time.monotonic, sleep=asyncio.sleep) -> int: + """Write every value at its time. Returns how many writes were made. + + Refuses to start unless every target was judged writable: a scenario half + played is a test of nothing. Refused outright in shadow mode, which + promises that this MES changes nothing outside its own database.""" + shadow.guard("opc.scenario_play", detail=f"Nothing written from {scenario.source}.") + blocked = [t for t in targets.values() if t.problem] + if blocked: + raise ScenarioError([f"{t.tag.alias}: {t.verdict}: {t.problem}" for t in blocked]) + writes = [w for w in scenario.writes if w.at >= start] + began = now() + done = 0 + for write in writes: + due = (write.at - start) / speed + wait = due - (now() - began) + if wait > 0: + await sleep(wait) + target = targets[write.alias] + if write.increment is not None: + current = await target.node.read_value() + if isinstance(current, bool) or not isinstance(current, int | float): + raise ScenarioError([f"line {write.line}: {write.alias} reads {current!r}, not a count"]) + value = current + write.increment + shown = f"{_show(value)} (+{_num(write.increment)})" + else: + value = write.value + shown = _show(value) + variant = coerce(value, target) + # No timestamps on the write: Kepware, like many servers, refuses a + # write that carries a SourceTimestamp. + try: + await target.node.write_attribute(ua.AttributeIds.Value, ua.DataValue(variant)) + except Exception as exc: + raise ScenarioError([f"{clock(write.at)} {write.alias}: the server refused the write " + f"({_short(exc)}); stopped after {done} writes"]) from None + done += 1 + why = f" ({write.why})" if write.why else "" + out(f"{clock(write.at)} set {write.alias} = {shown}{why}") + return done + + +def timeline(scenario: Scenario, out: Callable[[str], None] = print) -> None: + for write in scenario.writes: + why = f" ({write.why})" if write.why else "" + out(f"{clock(write.at)} set {write.alias} = {write.describe()}{why}") + + +# --------------------------------------------------- a server to rehearse on + +SERVE_TYPES = { + "Boolean": ua.VariantType.Boolean, "Int16": ua.VariantType.Int16, "UInt16": ua.VariantType.UInt16, + "Int32": ua.VariantType.Int32, "UInt32": ua.VariantType.UInt32, "Int64": ua.VariantType.Int64, + "Float": ua.VariantType.Float, "Double": ua.VariantType.Double, "String": ua.VariantType.String, +} + + +def _guess_type(scenario: Scenario, tag: Tag) -> tuple[ua.VariantType, Any]: + if tag.type: + vt = SERVE_TYPES[tag.type] + else: + first = next((w for w in scenario.writes if w.alias == tag.alias), None) + value = first.value if first and first.increment is None else 0 + sample = value[0] if isinstance(value, list) else value + vt = (ua.VariantType.Boolean if isinstance(sample, bool) else ua.VariantType.String + if isinstance(sample, str) else ua.VariantType.Double + if isinstance(sample, float) else ua.VariantType.Int32) + first = next((w for w in scenario.writes if w.alias == tag.alias and w.increment is None), None) + if first is not None and isinstance(first.value, list): + return vt, [_zero(vt)] * len(first.value) + return vt, _zero(vt) + + +def _zero(vt: ua.VariantType) -> Any: + return {ua.VariantType.Boolean: False, ua.VariantType.String: ""}.get( + vt, 0.0 if vt in FLOATS else 0) + + +async def rehearsal_server(scenario: Scenario, endpoint: str) -> Server: + """A server holding every tag of the scenario as a writable memory tag at + the scenario's own addresses, like a Kepware Memory Based channel. Not + started; the caller starts and stops it.""" + server = Server() + await server.init() + server.set_endpoint(endpoint) + server.set_server_name("fsmes scenario rehearsal") + namespace = scenario.namespace or "urn:fsmes:scenario" + index = await server.register_namespace(namespace) + folders: dict[str, Any] = {} + for tag in scenario.tags.values(): + vt, initial = _guess_type(scenario, tag) + if tag.is_node_id: + node_id = ua.NodeId.from_string(tag.address) + while node_id.NamespaceIndex > len(await server.get_namespace_array()) - 1: + await server.register_namespace(f"urn:fsmes:scenario:{len(await server.get_namespace_array())}") + node = await server.nodes.objects.add_variable( + node_id, tag.address.rpartition("=")[2], ua.Variant(initial, vt)) + else: + *parents, leaf = [p.split(":", 1)[-1] for p in tag.address.split("/")] + parent = server.nodes.objects + key = "" + for name in parents: + key += "/" + name + if key not in folders: + folders[key] = await parent.add_object(index, name) + parent = folders[key] + node = await parent.add_variable(index, leaf, ua.Variant(initial, vt)) + await node.set_writable() + return server diff --git a/src/fsmes/shadow.py b/src/fsmes/shadow.py index 4b4311b1..e7d68288 100644 --- a/src/fsmes/shadow.py +++ b/src/fsmes/shadow.py @@ -91,7 +91,7 @@ class Outbound: #: `allowed` — it runs, because nothing about the plant leaves this box #: by it, and nothing outside this MES changes. #: -#: **36 entries.** The count is stated because a register that quietly loses +#: **38 entries.** The count is stated because a register that quietly loses #: a row is worse than no register, and `tests/test_shadow_mode.py` scans the #: source for outbound primitives and fails on any call site not covered by #: an entry here. @@ -153,6 +153,16 @@ class Outbound: note="`fsmes opc-verify` and `fsmes opc-browse` connect, browse and " "read. Commissioning a shadow plant is exactly this", ), + Outbound( + name="opc.scenario_play", + where="fsmes.integrations.opc.scenario:play", + reaches="tags on an OPC UA server, Kepware first, written by `fsmes scenario play`", + verdict="refused", + note="the player writes only tags on channels its file declares " + "Simulator or Memory Based, but it still changes another " + "server, so shadow mode stops it before it connects and again " + "at the first write", + ), # ------------------------------------------------------------- the ERP Outbound( name="erp.adapter", @@ -422,6 +432,14 @@ class Outbound: verdict="allowed", note="replays recorded tables into that same bundled server", ), + Outbound( + name="sim.scenario_rehearsal", + where="fsmes.integrations.opc.scenario:rehearsal_server", + reaches="its own OPC UA server, on loopback, holding a scenario's tags", + verdict="allowed", + note="`fsmes scenario serve` makes the tags writable so a scenario can " + "be rehearsed without Kepware. It is not a plant", + ), Outbound( name="sim.harness", where="fsmes.sim", diff --git a/tests/test_scenario_player.py b/tests/test_scenario_player.py new file mode 100644 index 00000000..67e6bfff --- /dev/null +++ b/tests/test_scenario_player.py @@ -0,0 +1,346 @@ +"""`fsmes scenario`: timed tag values written into an OPC UA server. + +The file is read with its line numbers, so a mistake is named where it is. +The safety rule is code: a tag on a channel not declared Simulator or Memory +Based is refused, as is one the server shows has real equipment behind it, and +a read-only tag is reported and never written. Values go in as each tag's own +type, because Kepware refuses a wrong variant. +""" + +import asyncio +import json +import socket +from pathlib import Path + +import pytest +import structlog +from asyncua import Client, Server, ua +from typer.testing import CliRunner + +from fsmes.cli import app +from fsmes.integrations.opc import scenario as sc + +EXAMPLES = Path(__file__).resolve().parents[1] / "labs" / "kepsim" / "scenarios" + + +def _parse(doc: dict) -> sc.Scenario: + return sc.parse(json.dumps(doc, indent=2)) + + +def _problems(text: str) -> list[str]: + with pytest.raises(sc.ScenarioError) as caught: + sc.parse(text) + return caught.value.problems + + +# ------------------------------------------------------------------ the file + + +def test_a_broken_file_names_the_line_of_each_mistake(): + text = """{ + "tags": {"good": "ns=2;s=Sim.Dev.Good"}, + "drivers": {"Sim": "Simulator"}, + "steps": [ + {"at": "00:10", "tag": "good", "value": 1}, + {"at": "ten", "tag": "good", "value": 2}, + {"at": 20, "tag": "nope", "value": 3}, + {"at": 30, "tag": "good", "valu": 4} + ] +}""" + problems = _problems(text) + assert any(p.startswith("line 6:") and "'ten' is not a time" in p for p in problems) + assert any(p.startswith("line 7:") and "'nope' is not in `tags`" in p for p in problems) + assert any(p.startswith("line 8:") and "unknown key 'valu'" in p for p in problems) + assert not any(p.startswith("line 5:") for p in problems), "the good step is not blamed" + + +def test_a_file_that_is_not_json_names_the_line_it_broke_on(): + problems = _problems('{\n "tags": {"a": "ns=2;s=X.Y.Z"},\n "steps": [\n {"at": 1,}\n ]\n}') + assert problems[0].startswith("line 4:") + + +def test_a_key_given_twice_is_a_mistake_not_a_silent_overwrite(): + problems = _problems('{\n "tags": {"a": "ns=2;s=X.Y.Z"},\n "steps": [\n' + ' {"at": 1, "tag": "a", "value": 1, "value": 2}\n ]\n}') + assert problems == ["line 4: 'value' is given twice in the same block"] + + +def test_a_ramp_climbs_from_its_start_to_its_end_in_whole_numbers_on_a_count(): + plan = _parse({"tags": {"good": "ns=2;s=Sim.D.Good"}, + "steps": [{"at": "01:00", "tag": "good", + "ramp": {"from": 0, "to": 400, "over": 120, "every": 30}}]}) + assert [(w.at, w.value) for w in plan.writes] == [ + (60, 0), (90, 100), (120, 200), (150, 300), (180, 400)] + assert all(isinstance(w.value, int) for w in plan.writes) + + +def test_a_product_visits_each_station_in_order_advance_seconds_apart(): + plan = _parse({ + "tags": {"a.good": "ns=2;s=S.A.Good", "a.id": "ns=2;s=S.A.Id", "b.good": "ns=2;s=S.B.Good", + "b.state": "ns=2;s=S.B.State"}, + "product": {"id": "CAN-{n:03d}", "units": 2, "every_s": 30, "advance_s": 10, "start": 5, + "stations": [{"station": "A", "count": "a.good", "id": "a.id"}, + {"station": "B", "count": "b.good", "state": "b.state", + "running": 1, "after": 0}]}}) + rows = [(w.at, w.alias, w.increment if w.increment is not None else w.value) for w in plan.writes] + assert rows == [ + (5, "a.id", "CAN-001"), (5, "a.good", 1), + (15, "b.state", 1), (15, "b.good", 1), (25, "b.state", 0), + (35, "a.id", "CAN-002"), (35, "a.good", 1), + (45, "b.state", 1), (45, "b.good", 1), (55, "b.state", 0)] + + +def test_both_example_scenarios_in_the_lab_read_cleanly(): + files = sorted(EXAMPLES.glob("*.json")) + assert len(files) == 2, f"two examples expected, found {len(files)}: {files}" + for path in files: + assert sc.load(path).writes, path + + +def test_check_without_a_server_says_what_is_wrong_and_exits_2(tmp_path): + broken = tmp_path / "broken.json" + broken.write_text('{\n "tags": {"a": "ns=2;s=X.Y.Z"},\n "steps": [\n' + ' {"at": "1:xx", "tag": "a", "value": 1}\n ]\n}', encoding="utf-8") + result = CliRunner().invoke(app, ["scenario", "check", str(broken)]) + assert result.exit_code == 2 + assert "line 4:" in result.output and "not a time" in result.output + + +def test_play_in_shadow_mode_writes_nothing_and_does_not_connect(monkeypatch): + from fsmes.config import get_settings + + monkeypatch.setenv("MES_SHADOW", "true") + monkeypatch.setenv("MES_ERP_MODE", "off") + monkeypatch.setenv("MES_UNS_MODE", "off") + get_settings.cache_clear() + logging_before = structlog.get_config() + try: + # Port 1 answers nobody: had the player tried to connect, it would + # say so instead of naming shadow mode. + result = CliRunner().invoke(app, ["scenario", "play", str(EXAMPLES / "setpoint-drill.json"), + "--endpoint", "opc.tcp://127.0.0.1:1/none"]) + finally: + get_settings.cache_clear() + structlog.configure(**logging_before) + assert result.exit_code == 1 + assert "Nothing written" in result.output and "Shadow mode" in result.output + + +# --------------------------------------------------------------- the safety + + +def _tag(address: str) -> sc.Tag: + return sc.Tag("t", address, 1) + + +def test_a_channel_declared_simulator_or_memory_based_may_be_written(): + assert sc.judge_driver(_tag("ns=2;s=Scratch.D.T"), {"Scratch": "Simulator"}) == "" + assert sc.judge_driver(_tag("ns=2;s=Mem.D.T"), {"Mem": "memory-based"}) == "" + + +def test_a_channel_on_a_plc_driver_is_refused_with_its_name(): + reason = sc.judge_driver(_tag("ns=2;s=Press.PLC1.Speed"), {"Press": "Siemens TCP/IP Ethernet"}) + assert "'Press'" in reason and "Siemens TCP/IP Ethernet" in reason and "real equipment" in reason + + +def test_a_channel_nobody_declared_is_refused_rather_than_guessed(): + assert "not declared" in sc.judge_driver(_tag("ns=2;s=Unknown.D.T"), {}) + + +def test_an_advanced_simulator_channel_is_refused_because_odbc_tags_cannot_be_written(): + assert "ODBC" in sc.judge_driver(_tag("ns=2;s=SimLine.LD.LD_csv_State"), {"SimLine": "Advanced Simulator"}) + + +def test_a_numeric_node_id_has_no_channel_and_is_refused(): + assert "no channel" in sc.judge_driver(_tag("ns=2;i=13"), {"x": "Simulator"}) + + +def test_a_value_becomes_the_tags_own_type_or_says_why_not(): + def target(vt, length=None): + return sc.Target(_tag("ns=2;s=a.b.c"), variant=vt, array_length=length) + + assert sc.coerce(7, target(ua.VariantType.Int16)) == ua.Variant(7, ua.VariantType.Int16) + assert sc.coerce(12, target(ua.VariantType.String)) == ua.Variant("12", ua.VariantType.String) + assert sc.coerce(1, target(ua.VariantType.Boolean)) == ua.Variant(True, ua.VariantType.Boolean) + assert sc.coerce([1, 2], target(ua.VariantType.Int16, 2)).Value == [1, 2] + for value, vt, length, says in [(40000, ua.VariantType.Int16, None, "does not fit"), + (-1, ua.VariantType.UInt32, None, "does not fit"), + (1.5, ua.VariantType.Int32, None, "whole number"), + ([1, 2, 3], ua.VariantType.Int16, 2, "holds 2 values"), + (5, ua.VariantType.Int16, 4, "give a list")]: + with pytest.raises(ValueError, match=says): + sc.coerce(value, target(vt, length)) + + +# ---------------------------------------------------- against a real server + + +def _free_port() -> int: + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +async def _server(endpoint: str, *, kepware: bool = False) -> tuple[Server, dict]: + server = Server() + await server.init() + server.set_endpoint(endpoint) + await server.register_namespace("KEPServerEX" if kepware else "urn:test:scenario") + objects = server.nodes.objects + nodes = {} + for name, variant, writable in [ + ("Scratch.Dev.Lot", ua.Variant("", ua.VariantType.String), True), + ("Scratch.Dev.Running", ua.Variant(False, ua.VariantType.Boolean), True), + ("Scratch.Dev.State", ua.Variant(0, ua.VariantType.Int16), True), + ("Scratch.Dev.Good", ua.Variant(100, ua.VariantType.UInt32), True), + ("Scratch.Dev.Heads", ua.Variant([0, 0, 0, 0], ua.VariantType.Int16), True), + ("Scratch.Dev.Temp", ua.Variant(20.0, ua.VariantType.Float), True), + ("Scratch.Dev.Locked", ua.Variant(5, ua.VariantType.Int32), False), + ("Press.PLC1.Speed", ua.Variant(10, ua.VariantType.Int32), True), + ("Fake.PLC2.Speed", ua.Variant(10, ua.VariantType.Int32), True), + ("Fake.PLC2._System._DeviceId", ua.Variant("192.168.1.10", ua.VariantType.String), False), + ]: + node = await objects.add_variable(ua.NodeId(name, 2), name, variant) + if writable: + await node.set_writable() + nodes[name] = node + return server, nodes + + +def _run(coro): + return asyncio.run(coro) + + +TYPED = { + "tags": {"lot": "ns=2;s=Scratch.Dev.Lot", "running": "ns=2;s=Scratch.Dev.Running", + "state": "ns=2;s=Scratch.Dev.State", "good": "ns=2;s=Scratch.Dev.Good", + "heads": "ns=2;s=Scratch.Dev.Heads", "temp": "ns=2;s=Scratch.Dev.Temp"}, + "drivers": {"Scratch": "Simulator"}, + "steps": [{"at": 0, "tag": "lot", "value": "LOT-7"}, + {"at": 0, "tag": "running", "value": True}, + {"at": 2, "tag": "state", "value": 3}, + {"at": 2, "tag": "heads", "value": [1, 2, 3, 4]}, + {"at": 4, "tag": "temp", "value": 61.5}], + "product": {"advance_s": 1, "start": 6, "stations": [{"station": "Dev", "count": "good", "by": 5}]}, +} + + +@pytest.mark.slow +def test_each_value_lands_at_its_time_as_the_tags_own_type(): + endpoint = f"opc.tcp://127.0.0.1:{_free_port()}/test" + plan = _parse(TYPED) + + async def go(): + server, nodes = await _server(endpoint) + async with server: + async with Client(endpoint) as client: + targets = await sc.inspect(client, plan) + assert {a: t.problem for a, t in targets.items() if t.problem} == {} + clock = [0.0] + stamped = [] + + async def sleep(seconds): + clock[0] += seconds + + def out(line): + stamped.append((clock[0], line.split(" set ")[1].split(" =")[0])) + + done = await sc.play(client, plan, targets, speed=2.0, out=out, + now=lambda: clock[0], sleep=sleep) + values = {name: await node.read_data_value() for name, node in nodes.items()} + return done, stamped, values + + done, stamped, values = _run(go()) + assert done == 6 + # Played at 2x: the writes due at 0, 2, 4 and 6 s land at 0, 1, 2 and 3 s. + assert stamped == [(0, "lot"), (0, "running"), (1, "state"), (1, "heads"), (2, "temp"), (3, "good")] + expect = {"Scratch.Dev.Lot": ("LOT-7", ua.VariantType.String), + "Scratch.Dev.Running": (True, ua.VariantType.Boolean), + "Scratch.Dev.State": (3, ua.VariantType.Int16), + "Scratch.Dev.Heads": ([1, 2, 3, 4], ua.VariantType.Int16), + "Scratch.Dev.Temp": (61.5, ua.VariantType.Float), + "Scratch.Dev.Good": (105, ua.VariantType.UInt32)} + for name, (value, vt) in expect.items(): + assert (values[name].Value.Value, values[name].Value.VariantType) == (value, vt), name + + +@pytest.mark.slow +def test_a_tag_on_a_plc_driver_is_refused_and_nothing_at_all_is_written(): + endpoint = f"opc.tcp://127.0.0.1:{_free_port()}/test" + plan = _parse({"tags": {"lot": "ns=2;s=Scratch.Dev.Lot", "speed": "ns=2;s=Press.PLC1.Speed"}, + "drivers": {"Scratch": "Simulator", "Press": "Siemens TCP/IP Ethernet"}, + "steps": [{"at": 0, "tag": "lot", "value": "LOT-1"}, + {"at": 1, "tag": "speed", "value": 99}]}) + + async def go(): + server, nodes = await _server(endpoint) + async with server: + async with Client(endpoint) as client: + targets = await sc.inspect(client, plan) + with pytest.raises(sc.ScenarioError) as caught: + await sc.play(client, plan, targets, sleep=lambda s: asyncio.sleep(0)) + return targets, caught.value, await nodes["Scratch.Dev.Lot"].read_value(), \ + await nodes["Press.PLC1.Speed"].read_value() + + targets, error, lot, speed = _run(go()) + assert targets["speed"].verdict == "refused" and "Siemens" in targets["speed"].problem + assert "speed: refused" in str(error) + assert (lot, speed) == ("", 10), "a refusal writes nothing, not even the allowed tags" + + +@pytest.mark.slow +def test_kepware_showing_a_network_device_overrules_a_simulator_declaration(): + endpoint = f"opc.tcp://127.0.0.1:{_free_port()}/test" + plan = _parse({"tags": {"speed": "ns=2;s=Fake.PLC2.Speed"}, "drivers": {"Fake": "Simulator"}, + "steps": [{"at": 0, "tag": "speed", "value": 1}]}) + + async def go(): + server, _ = await _server(endpoint, kepware=True) + async with server, Client(endpoint) as client: + return await sc.inspect(client, plan) + + target = _run(go())["speed"] + assert target.verdict == "refused" + assert "192.168.1.10" in target.problem + + +@pytest.mark.slow +def test_a_read_only_tag_is_reported_and_not_written(): + endpoint = f"opc.tcp://127.0.0.1:{_free_port()}/test" + plan = _parse({"tags": {"locked": "ns=2;s=Scratch.Dev.Locked", "missing": "ns=2;s=Scratch.Dev.Nope"}, + "drivers": {"Scratch": "Simulator"}, + "steps": [{"at": 0, "tag": "locked", "value": 9}, {"at": 0, "tag": "missing", "value": 1}]}) + + async def go(): + server, nodes = await _server(endpoint) + async with server: + async with Client(endpoint) as client: + targets = await sc.inspect(client, plan) + lines = [] + bad = sc.report(targets, lines.append) + return targets, bad, lines, await nodes["Scratch.Dev.Locked"].read_value() + + targets, bad, lines, locked = _run(go()) + assert targets["locked"].verdict == "read-only" + assert targets["missing"].verdict == "missing" + assert bad == 2 and lines[-1] == "2 tags: 0 can be written, 2 cannot." + assert locked == 5 + + +@pytest.mark.slow +def test_the_lab_product_walk_plays_into_its_own_rehearsal_server(): + endpoint = f"opc.tcp://127.0.0.1:{_free_port()}/rehearsal" + plan = sc.load(EXAMPLES / "product-through-the-line.json") + + async def go(): + server = await sc.rehearsal_server(plan, endpoint) + async with server, Client(endpoint) as client: + targets = await sc.inspect(client, plan) + await sc.play(client, plan, targets, speed=1000, out=lambda _: None) + return {alias: await client.get_node(tag.address).read_value() + for alias, tag in plan.tags.items()} + + values = _run(go()) + assert values["pal.good"] == 3 and values["ld.good"] == 3 + assert values["qi.serial"] == "CAN-0003" and values["qi.pass"] is True + assert values["ld.state"] == 0, "the loader goes back to 0 once the last can has left it" diff --git a/tests/test_shadow_mode.py b/tests/test_shadow_mode.py index c6eacb46..a0c707d2 100644 --- a/tests/test_shadow_mode.py +++ b/tests/test_shadow_mode.py @@ -58,6 +58,9 @@ async def write_value(self, value): self.writes.append(value) self.value = value + async def write_attribute(self, attribute, data_value): + await self.write_value(data_value.Value.Value) + async def read_value(self): return self.value @@ -136,6 +139,20 @@ def test_no_setpoint_reaches_the_plant_in_shadow_mode(shadow_on): assert "Shadow mode" in str(caught.value) +def test_a_scenario_writes_nothing_in_shadow_mode(shadow_on): + from fsmes.integrations.opc import scenario + + plan = scenario.parse('{"drivers": {"Sim": "Simulator"}, "tags": {"t": "ns=2;s=Sim.D.T"},' + ' "steps": [{"at": 0, "tag": "t", "value": 1}]}') + node = FakeNode(0) + target = scenario.Target(tag=plan.tags["t"], node=node, variant=scenario.ua.VariantType.Int16, + writable=True) + with pytest.raises(shadow.ShadowRefused) as caught: + asyncio.run(scenario.play(None, plan, {"t": target}, out=lambda line: None)) + assert node.writes == [] + assert "Shadow mode" in str(caught.value) + + def test_an_approved_recommendation_waits_instead_of_reaching_the_machine(session, scope, shadow_on): from fsmes.domain import AdjustmentStatus, RecommendedAdjustment @@ -404,6 +421,7 @@ def test_every_entry_in_the_register_is_a_distinct_place_with_a_verdict_and_a_re "opc.node_write": "test_no_setpoint_reaches_the_plant_in_shadow_mode", "opc.adjustment_write": "test_an_approved_recommendation_waits_instead_of_reaching_the_machine", "opc.order_code": "test_the_order_code_is_not_written_down_to_the_machines", + "opc.scenario_play": "test_a_scenario_writes_nothing_in_shadow_mode", "erp.adapter": "test_no_live_erp_adapter_can_be_built_in_shadow_mode", "erp.rest": "test_a_live_erp_and_shadow_mode_cannot_both_be_asked_for", "erp.erpnext": "test_a_live_erp_and_shadow_mode_cannot_both_be_asked_for", @@ -442,7 +460,8 @@ def test_every_path_shadow_mode_closes_is_proved_closed_in_this_file(): #: patterns matched against the source of each call. Named, because a #: failure should say what kind of reach it found. PRIMITIVES = { - "an OPC UA node write": r"\.write_value\s*\(|\.write_attribute_value\s*\(|\.set_writable\s*\(", + "an OPC UA node write": (r"\.write_value\s*\(|\.write_attribute_value\s*\(|\.write_attribute\s*\(|" + r"\.set_writable\s*\("), # `session` is deliberately absent from the receiver list: a SQLAlchemy # Session's .get() is a database read and appears in half the package. "an HTTP call": (r"\b(?:httpx|requests)\.\w*\s*\(|\burlopen\s*\(|"