diff --git a/.vscode/tasks.json b/.vscode/tasks.json index 21cc8bc..9140725 100644 --- a/.vscode/tasks.json +++ b/.vscode/tasks.json @@ -1476,6 +1476,233 @@ ], "isBackground": false, "problemMatcher": [] + }, + { + "label": "delete old plantenv", + "type": "shell", + "command": ".venv\\Scripts\\python.exe", + "args": [ + "-c", + "import pathlib; p = pathlib.Path('src/wago_sim/scenario/plantenv.py'); p.unlink(); print('deleted:', not p.exists())" + ], + "isBackground": false, + "problemMatcher": [] + }, + { + "label": "wago-sim: pytest env2", + "type": "shell", + "command": ".venv\\Scripts\\python.exe", + "args": [ + "-m", + "pytest", + "tests/test_continuous.py", + "-q", + "--tb=short" + ], + "isBackground": false, + "problemMatcher": [] + }, + { + "label": "debug load_env", + "type": "shell", + "command": ".venv\\Scripts\\python.exe", + "args": [ + "-c", + "from wago_sim.scenario.plantenv import load_env, PlantEnv; import envs.pump_station_env as m; print('class:', m.PumpStationEnv); print('is subclass:', issubclass(m.PumpStationEnv, PlantEnv)); print('module plantenv file:', __import__('wago_sim.scenario.plantenv', fromlist=['x']).__file__)" + ], + "isBackground": false, + "problemMatcher": [] + }, + { + "label": "pytest load env detail", + "type": "shell", + "command": ".venv\\Scripts\\python.exe", + "args": [ + "-m", + "pytest", + "tests/test_continuous.py::TestPlantEnv::test_load_env_class_and_factory", + "-q", + "--tb=long", + "-s" + ], + "isBackground": false, + "problemMatcher": [] + }, + { + "label": "wago-sim: pytest env3", + "type": "shell", + "command": ".venv\\Scripts\\python.exe", + "args": [ + "-m", + "pytest", + "tests/test_continuous.py", + "-q", + "--tb=short" + ], + "isBackground": false, + "problemMatcher": [] + }, + { + "label": "wago-sim: pytest env4", + "type": "shell", + "command": ".venv\\Scripts\\python.exe", + "args": [ + "-m", + "pytest", + "tests/test_continuous.py", + "-q", + "--tb=short" + ], + "isBackground": false, + "problemMatcher": [] + }, + { + "label": "debug missing env", + "type": "shell", + "command": ".venv\\Scripts\\python.exe", + "args": [ + "-c", + "from wago_sim.scenario.plantenv import load_env; from wago_sim.modules import DI_8; from wago_sim.app import build_app; app = build_app(modules=[DI_8]);\ntry:\n e = load_env('envs.missing_module', node=app.node)\n print('NO RAISE; got:', type(e))\nexcept Exception as exc:\n print('raised:', exc)" + ], + "isBackground": false, + "problemMatcher": [] + }, + { + "label": "run debug env", + "type": "shell", + "command": ".venv\\Scripts\\python.exe", + "args": [ + "scripts\\debug_env.py" + ], + "isBackground": false, + "problemMatcher": [] + }, + { + "label": "debug in pytest", + "type": "shell", + "command": ".venv\\Scripts\\python.exe", + "args": [ + "-m", + "pytest", + "tests/test_continuous.py::TestPlantEnv::test_load_env_class_and_factory", + "-q", + "--tb=long" + ], + "isBackground": false, + "problemMatcher": [] + }, + { + "label": "rm debug script", + "type": "shell", + "command": ".venv\\Scripts\\python.exe", + "args": [ + "-c", + "import pathlib; pathlib.Path('scripts/debug_env.py').unlink(); print('ok')" + ], + "isBackground": false, + "problemMatcher": [] + }, + { + "label": "wago-sim: pytest all env", + "type": "shell", + "command": ".venv\\Scripts\\python.exe", + "args": [ + "-m", + "pytest", + "tests", + "-q", + "--tb=short" + ], + "isBackground": false, + "problemMatcher": [] + }, + { + "label": "wago-sim: serve env", + "type": "shell", + "command": ".venv\\Scripts\\python.exe", + "args": [ + "-m", + "wago_sim.cli", + "serve", + "--node", + "config/pump_node.yaml", + "--host", + "127.0.0.1", + "--port", + "5020", + "--web-port", + "8088", + "--env", + "envs.pump_station_env:PumpStationEnv" + ], + "isBackground": true, + "problemMatcher": [] + }, + { + "label": "env closed-loop probe", + "type": "shell", + "command": ".venv\\Scripts\\python.exe", + "args": [ + "scripts\\env_probe.py" + ], + "isBackground": false, + "problemMatcher": [] + }, + { + "label": "headless sim probe", + "type": "shell", + "command": ".venv\\Scripts\\python.exe", + "args": [ + "scripts\\sim_probe.py" + ], + "isBackground": false, + "problemMatcher": [] + }, + { + "label": "headless sim probe2", + "type": "shell", + "command": ".venv\\Scripts\\python.exe", + "args": [ + "scripts\\sim_probe.py" + ], + "isBackground": false, + "problemMatcher": [] + }, + { + "label": "rm probes", + "type": "shell", + "command": ".venv\\Scripts\\python.exe", + "args": [ + "-c", + "import pathlib; [pathlib.Path(p).unlink() for p in ('scripts/env_probe.py', 'scripts/sim_probe.py')]; print('ok')" + ], + "isBackground": false, + "problemMatcher": [] + }, + { + "label": "stop env serve", + "type": "shell", + "command": ".venv\\Scripts\\python.exe", + "args": [ + "-c", + "import subprocess; subprocess.run(['powershell','-NoProfile','-Command','Get-NetTCPConnection -LocalPort 5020,8088 -ErrorAction SilentlyContinue | Select-Object -ExpandProperty OwningProcess -Unique | ForEach-Object { Stop-Process -Id $_ -Force }'], capture_output=True); print('stopped')" + ], + "isBackground": false, + "problemMatcher": [] + }, + { + "label": "wago-sim: pytest final env", + "type": "shell", + "command": ".venv\\Scripts\\python.exe", + "args": [ + "-m", + "pytest", + "tests", + "-q", + "--tb=short" + ], + "isBackground": false, + "problemMatcher": [] } ] } \ No newline at end of file diff --git a/README.md b/README.md index c01ae95..ddaf1c3 100644 --- a/README.md +++ b/README.md @@ -185,6 +185,56 @@ Configuration options: `--no-web`, `--web-port ` or env benches (all mutating endpoints answer 403). Plain HTTP — keep the port inside the bench network. +## Plant environments (continuous simulation) + +A **plant environment** simulates the physical world around the +distributed I/O: per deterministic tick it reads what the PLC +commanded (DO/AO), runs your plant physics, and writes what the field +now shows (DI/AI) — the sensors the PLC reads back. Environments run +**until manually stopped** (Ctrl+C / `docker stop`), optionally ending +sooner via an `is_done()` criterion. + +Attach one to a live Modbus server (wall-clock-paced 10 ms ticks): + +```bash +wago-sim serve --node config/pump_node.yaml --port 5020 \ + --env envs.pump_station_env:PumpStationEnv +``` + +or run headless (unpaced, for plant-model development): + +```bash +wago-sim sim --node config/pump_node.yaml \ + --env envs.pump_station_env:PumpStationEnv +``` + +**Authoring one** is a small subclass — override `model()`: + +```python +# myenvs/tank.py +from wago_sim.scenario.plantenv import PlantEnv + +class TankEnv(PlantEnv): + def model(self, state, dt_s): + pump_on = bool(state.by_code("DO0.0")) # PLC command + level = float(state.by_code("AI0.0")) # current field + return {"AI0.0": level + (-5.0 if pump_on else 1.0) * dt_s} + +Env = TankEnv # `--env myenvs.tank` picks this up +``` + +API surface (spec FR-8.2.14–17): `observe()` returns every +DI/DO/AI/AO channel (I/O codes, names, values) plus aliases and +degradation parameters; `apply()` writes DI/AI (the field); +`force()` writes DO/AO master-side (headless runs, fault setup); +`is_done()` optional end; `reset()` deterministic replay. Reference +forms: `pkg.mod:ClassName`, `pkg.mod:factory`, or `pkg.mod` with +`Env`/`make_env`. The shipped example is +`envs/pump_station_env.py` (tank + pump physics with a low-level +switch). Scenario-continuous mode (§8 YAML: datasets, degradation, +invariants) remains available for scripted soak runs on the same +engine ticks. + ## Documentation - `docs/spec/software-spec.md` — the software specification @@ -404,17 +454,18 @@ reporting, snapshots, and the gym-style `PlantEnv`. 30 new tests Datasets are **optional**: signal logic runs in parallel with CSV streaming, or entirely without it. -- [x] Plant environment `scenario/plantenv.py` (FR-8.2.14–16, requested - 2026-10-04): gym-style `PlantEnv` — deterministic engine-owned - tick clock (never wall-clock); per tick: - `observe()` returns the WAGO input image in engineering units - (channel values, names, per-type I/O codes, ranges, sim-time); - the decision source decides; `apply()` writes the decided - outputs through the datastore's internal paths (mirrors/ - statistics/watchdog stay consistent). Existing `plugins:` - become thin adapters; external (non-Python) decision sources - step the env over a socket; one decision source per output - alias per tick (ownership conflicts are errors). +- [x] Plant environment `scenario/plantenv.py` (FR-8.2.14–17, reworked + 2026-10-04 to environment-first): standalone `PlantEnv` — + `observe()` returns ALL modules (DI/DO/AI/AO channels, per-type + codes, names, parameters); `apply()` writes the FIELD side + (DI/AI) by default; `force()` writes DO/AO master-side (mirrors/ + statistics/watchdog stay consistent); `model()`/`is_done()` + override hooks (subclass authoring, FR-8.2.17); `reset()` + deterministic. Runners: `serve --env ...` wall-clock-paced + alongside Modbus, runs until stopped; `sim --env ...` headless + unpaced until is_done()/Ctrl+C. Example: `envs/pump_station_env.py`. + Verified live: PLC coil write → env physics → level registers + drain/refill over Modbus. - [x] `scenario/datasets.py` — streaming CSV playback for designated inputs (OPTIONAL driver): header-driven time column, `interpolation: linear|hold`, `repeat` with seamless seams, diff --git a/docs/spec/software-spec.md b/docs/spec/software-spec.md index 479b14c..fa50280 100644 --- a/docs/spec/software-spec.md +++ b/docs/spec/software-spec.md @@ -580,44 +580,49 @@ unchanged whether or not CSV streaming is used. top-level `until:` bounding the run. All §8.3–8.4 machinery (degradation, invariants, rate limits, snapshots) applies identically, whether or not any CSV streaming is configured. -- FR-8.2.14: **Plant environment abstraction** (requested 2026-10-04): the - continuous engine exposes a gymnasium-style **environment object** — - `PlantEnv` — as the single programmatic interface to the simulated plant. - Per deterministic tick (the engine's fixed tick clock, FR-4.3.9 — never - wall-clock), the cycle is: - **`state = env.observe()`** → the caller (plant logic, an external - agent, or the scenario) decides → **`env.apply(actions)`** writes the - decided output values → the engine advances exactly one tick - (edges, datasets, degradation, invariants — FR-8.2.12 order) → repeat. - `observe()` returns the state of the **generic inputs** — the WAGO - input process image in engineering units (all DI/DO/AI/AO channel - values, per-type module/channel addressing, configurable channel - names, plus the active aliases) — and `apply(actions)` maps decided - values onto the **output process image** through the datastore's - internal Modbus write paths, so mirrors, statistics, and watchdog - behavior stay consistent (FR-9.2.6 semantics). The environment owns - no policy: the plant model (physics, closed-loop control) is supplied - by the caller, exactly like `gym.Env` vs. its agents. -- FR-8.2.15: **Environment integration**: `PlantEnv` is the interface the - per-tick plugins (FR-4.3.8) are migrated onto: a plugin becomes - `env.observe()` → decide → `env.apply()` per tick, and a scenario may - alternatively register an external `step(env)` agent — including a - non-Python one (e.g., a trained RL policy) communicating over a - socket/pipe — without touching the engine loop. The existing - `plugins:` mechanism keeps working unchanged (thin adapter); the - engine always owns the tick clock, and exactly one decision source - may drive an output alias per tick (ownership conflicts are validation - errors, extending FR-8.2.2). +- FR-8.2.14: **Plant environment abstraction** (requested 2026-10-04, + reworked same day to environment-first): `PlantEnv` is the **primary + interface** for continuous plant simulation, standalone-first — it + requires only a virtual node + datastore (no scenario), with the §8.3/8.4 + machinery (degradation, invariants) as optional attachments. Per + deterministic tick (the engine or the runner advances the tick clock, + never wall-clock arithmetic): **`state = env.observe()`** returns the + state of ALL modules — DI/DO/AI/AO channels, engineering units, per-type + I/O codes, configured names, plus active aliases and parameters — → + `env.model(state, dt)` (the caller-supplied plant physics) → **`apply()` + writes the new field state**. The environment owns no policy; plant + models are supplied by subclassing (FR-8.2.17). +- FR-8.2.15: **Action semantics**: `apply()` primarily writes the **input + side (DI/AI)** — the simulated field: what the sensors show, driven by + plant physics reacting to the PLC's output commands. Writing outputs + (DO/AO) is allowed as explicit **forcing** via a `force=True` keyword or + `env.force(...)` — for headless runs without a PLC and for bench + fault setup; forced outputs go through the datastore's internal write + paths so mirrors/statistics/watchdog stay consistent (FR-9.2.6). + Exactly one decision source may own an alias per tick (ownership + conflicts are validation errors, extending FR-8.2.2). - FR-8.2.16: **Environment API surface** (concrete): - `env.observe() -> PlantState` (channel values, names, I/O codes, - ranges, sim-time, wear/parameters); `env.apply(actions) -> None` - (actions keyed by I/O code or alias name; unknown keys are errors); - `env.reset(config) -> PlantState` (fresh episode, deterministic - seed); `env.tick_ms`, `env.sim_time_ms` (read-only clock); and the - full §8.3/§8.4 machinery (parameters, degradation, invariants, - snapshots) reachable as `env.parameters`, `env.invariants`. The same - object is served over the scenario-run HTTP API in server mode (§9), - so external agents can run against a live PLC-connected bench. + `env.observe() -> PlantState` (all channels, aliases, parameters, + sim-time); `env.apply(values, force=False)` (keys: I/O code or alias + name; DI/AI default, DO/AO require force; unknown keys are errors); + `env.model(state, dt_s)` — override point for plant physics, returning + new field values; `env.is_done(state) -> bool` — optional end + criterion, default False (runs until manually stopped); + `env.reset()` — deterministic fresh episode; `env.tick_ms` / + `env.sim_time_ms` (read-only clock); `env.parameters` / + `env.invariants` attachments. +- FR-8.2.17: **Environment authoring & runners**: environments are easy + to create and call — a subclass of `PlantEnv` overriding `model()` + (and optionally `is_done()`) in any importable module + (`--env myenvs.pump_station.PumpStationEnv`), or a module-level + factory function. Two runners: (a) `wago-sim serve --env ...` ticks + the environment on a **wall-clock-paced** periodic task (10 ms tick + alongside the Modbus server, real-time behavior for a live PLC) and + runs **until manually stopped** (SIGINT/SIGTERM — the existing + graceful-shutdown path); (b) `wago-sim sim --node ... --env ...` + runs headless unpaced (as fast as determinism allows) until + `is_done()` or SIGINT. The scenario-continuous mode (FR-8.2.1-13) + remains a scripted layer on top of the same engine ticks. ### 8.3 Requirements — degradation models (plant logic over time) diff --git a/envs/__init__.py b/envs/__init__.py new file mode 100644 index 0000000..fc4a742 --- /dev/null +++ b/envs/__init__.py @@ -0,0 +1,8 @@ +"""User-authored plant environments (spec FR-8.2.17). + +Each module here is loadable by reference: + + wago-sim serve --node ... --env envs.pump_station_env:PumpStationEnv + wago-sim sim --node ... --env envs.pump_station_env:make_env + wago-sim serve --node ... --env envs.pump_station_env # picks Env +""" \ No newline at end of file diff --git a/envs/pump_station_env.py b/envs/pump_station_env.py new file mode 100644 index 0000000..4b05e99 --- /dev/null +++ b/envs/pump_station_env.py @@ -0,0 +1,74 @@ +"""Example pump-station plant environment (spec FR-8.2.17). + +Demonstrates environment authoring: subclass PlantEnv, override +``model()`` — read what the PLC commanded (DO/AO), compute what the +field now shows (DI/AI). Run it live against a Modbus master: + + wago-sim serve --node config/pump_node.yaml --port 5020 \ + --env envs.pump_station_env:PumpStationEnv + +or headless for model development: + + wago-sim sim --node config/pump_node.yaml \ + --env envs.pump_station_env:PumpStationEnv + +Physics (deliberately simple, for demonstration): one tank. +- inflow is constant (or a small sinusoidal demand profile) +- the pump drains at `flow_max * capacity * speed` while DO0.0 is ON +- the level integrates the balance; AI0.0 reports it as 0.1 mm/count +- a low-level switch DI0.1 opens when the tank drains below 500 mm +- optional is_done: stop after a sim hour (default: run forever) +""" + +from __future__ import annotations + +from wago_sim.scenario.plantenv import PlantEnv, PlantState + +TANK_AREA = 2.0 # m^2 +FLOW_MAX = 30.0 # m^3/h the pump moves at full speed +INFLOW = 12.0 # m^3/h constant inflow +LEVEL_MAX = 9000.0 # mm — tank brim +SWITCH_AT = 500.0 # mm — low-level switch threshold + + +class PumpStationEnv(PlantEnv): + """Closed-loop tank + pump physics over the pump_node layout.""" + + def __init__(self, *args, stop_after_ms: float | None = None, **kw): + super().__init__(*args, **kw) + self._level_mm = 3000.0 # initial tank level + self.stop_after_ms = stop_after_ms + + def model(self, state: PlantState, dt_s: float) -> dict[str, object]: + dt_h = dt_s / 3600.0 + # what the PLC currently commands + pump_on = bool(state.by_code("DO0.0")) + # optional speed setpoint from AO0.0 (counts -> 0..1 fraction) + speed_raw = float(state.by_code("AO0.0") or 0) + speed = min(1.0, max(0.0, speed_raw / 4095.0)) + # physics: water balance with pump wear (parameters attachable) + capacity = state.parameters.get("pump_capacity", 1.0) + outflow = FLOW_MAX * capacity * speed if pump_on else 0.0 + d_level_mm = (INFLOW - outflow) / TANK_AREA * 1000.0 * dt_h + self._level_mm = min(LEVEL_MAX, max(0.0, self._level_mm + d_level_mm)) + # the field the PLC reads back + level_counts = int(self._level_mm / 0.1) # AI0.0: 0.1 mm/count + low_switch = self._level_mm < SWITCH_AT # DI0.1 + return { + "AI0.0": level_counts, + "DI0.1": low_switch, + } + + def is_done(self, state: PlantState) -> bool: + if self.stop_after_ms is None: + return False # default: run until stopped + return state.sim_time_ms >= self.stop_after_ms + + +def make_env(node=None, engine=None) -> PumpStationEnv: + """Factory form (FR-8.2.17): envs.pump_station_env:make_env.""" + return PumpStationEnv(node=node, engine=engine) + + +# bare-module form: `--env envs.pump_station_env` picks this up +Env = PumpStationEnv \ No newline at end of file diff --git a/src/wago_sim/cli.py b/src/wago_sim/cli.py index 15e9ea8..da21b72 100644 --- a/src/wago_sim/cli.py +++ b/src/wago_sim/cli.py @@ -196,6 +196,12 @@ async def _serve(args: argparse.Namespace) -> int: app = build_app(modules=modules, host=args.host, port=args.port, web=web_cfg, node_config_path=args.node, channel_names=channel_names, channel_ranges=channel_ranges) + # optional plant environment (FR-8.2.17): ticked wall-clock-paced + # alongside the Modbus server, runs until manually stopped + env = None + if getattr(args, "env", None): + from .scenario.plantenv import load_env + env = load_env(args.env, node=app.node) await app.server.start() web_server = None if web_cfg is not None: @@ -204,6 +210,27 @@ async def _serve(args: argparse.Namespace) -> int: f":{web_cfg.port} (user: {web_cfg.user})") print(f"Simulating WAGO 750-362 with {len(modules)} modules " f"on {args.host}:{args.port} (Ctrl+C to stop)") + env_task = None + if env is not None: + async def run_env(): + """Paced ticker (FR-8.2.17): wall-clock-aligned ticks for a + live PLC; runs until cancelled by the shutdown path.""" + import time as _time + period = env.TICK_MS / 1000.0 + next_at = _time.monotonic() + while True: + env.tick() + if env.is_done(env.observe()): + log.info("environment is_done() satisfied; env stopped") + return + next_at += period + delay = next_at - _time.monotonic() + if delay > 0: + await asyncio.sleep(delay) + else: + next_at = _time.monotonic() # catch up, no drift burst + env_task = asyncio.create_task(run_env()) + print(f"Plant environment: {args.env} (10 ms paced tick, Ctrl+C to stop)") # Graceful shutdown on SIGINT/SIGTERM (docker stop, systemctl stop). # asyncio.run installs a SIGINT handler; SIGTERM arrives as a raw @@ -220,6 +247,12 @@ async def _serve(args: argparse.Namespace) -> int: except asyncio.CancelledError: pass # SIGINT via asyncio.run's handler finally: + if env_task is not None: + env_task.cancel() + try: + await env_task + except asyncio.CancelledError: + pass if web_server is not None: task = getattr(web_server, "_serve_task", None) web_server.should_exit = True @@ -232,6 +265,34 @@ async def _serve(args: argparse.Namespace) -> int: return 0 +def _sim(args: argparse.Namespace) -> int: + """Headless environment run (FR-8.2.17): unpaced, until is_done() + or Ctrl+C — for plant-model development without a PLC.""" + node_config = _load_yaml_file(args.node, "node config") + modules = load_node_modules(node_config) + channel_names = load_channel_names(node_config) + channel_ranges = load_channel_ranges(node_config) + app = build_app(modules=modules, channel_names=channel_names, + channel_ranges=channel_ranges) + from .scenario.plantenv import load_env + env = load_env(args.env, node=app.node) + print(f"Running {args.env} headless (Ctrl+C to stop; is_done() to end)") + ticks = 0 + try: + while True: + state = env.tick() + ticks += 1 + if env.is_done(state): + print(f"is_done() at tick {ticks} (sim {state.sim_time_ms:.0f} ms)") + break + if ticks % 1000 == 0 and args.verbose: + print(f"tick {ticks}: {state.sim_time_ms:.0f} ms") + except KeyboardInterrupt: + print(f"\nstopped manually at tick {ticks} " + f"(sim {env.sim_time_ms:.0f} ms)") + return 0 + + def _run_scenario(args: argparse.Namespace) -> int: node_config = _load_yaml_file(args.node, "node config") modules = load_node_modules(node_config) @@ -332,6 +393,12 @@ def main(argv: list[str] | None = None) -> int: "--web-port", type=_port_type, default=_env_port("WAGO_SIM_WEB_PORT", 8080), help="web UI port (or env: WAGO_SIM_WEB_PORT; default 8080)", ) + serve.add_argument( + "--env", default=None, + help=("plant environment module (FR-8.2.17): e.g. " + "envs.pump_station_env:PumpStationEnv — ticked wall-clock-" + "paced alongside Modbus; runs until stopped"), + ) serve.set_defaults(func=_serve) runp = sub.add_parser("run", help="run a scenario headless (no server)") @@ -345,6 +412,13 @@ def main(argv: list[str] | None = None) -> int: val.add_argument("--scenario", required=True) val.set_defaults(func=_validate) + sim = sub.add_parser( + "sim", help="run a plant environment headless (FR-8.2.17)") + sim.add_argument("--node", required=True, help="node config YAML") + sim.add_argument("--env", required=True, + help="environment module, e.g. envs.pump_station_env:PumpStationEnv") + sim.set_defaults(func=_sim) + args = parser.parse_args(argv) _build_logger(args.verbose) # `--node` may come from the environment when omitted (service mode); diff --git a/src/wago_sim/scenario/engine.py b/src/wago_sim/scenario/engine.py index 8e7f44f..b72007b 100644 --- a/src/wago_sim/scenario/engine.py +++ b/src/wago_sim/scenario/engine.py @@ -204,7 +204,7 @@ class ScenarioEngine: for model in self.degradation.models.values(): self._claim_alias(model.observes, f"degradation:{model.name}") self.invariants = InvariantMonitor(self, self.scenario) - self.env = PlantEnv(self) + self.env = PlantEnv(engine=self) # FR-8.2.1: a continuous run needs at least one run-length driver if (not self.datasets and not self._scheduled @@ -519,10 +519,10 @@ class ScenarioEngine: return self._result # (2) edges self._detect_edges() - # (3) plant logic: plugins first, then queued env actions flush + # (3) plant logic: plugins first, then queued env decisions land + # (the scenario layer drives env via observe/apply; queued + # values are already applied by the decision source) self._run_tick_plugins() - if self.env is not None: - self.env.flush() self._advance_ramps(dt) # (4) degradation if self.degradation is not None: diff --git a/src/wago_sim/scenario/plantenv.py b/src/wago_sim/scenario/plantenv.py index 50a017c..45c7344 100644 --- a/src/wago_sim/scenario/plantenv.py +++ b/src/wago_sim/scenario/plantenv.py @@ -1,24 +1,32 @@ -"""PlantEnv: gymnasium-style environment over the simulated plant (spec §8, FR-8.2.14-16). +"""PlantEnv: gymnasium-style plant environment (spec §8, FR-8.2.14-17). -The environment is the decision-source interface to the distributed -I/O: per deterministic engine tick — +Standalone-first: requires only a virtual node + datastore — no +scenario. The environment is the interface for continuous plant +simulation; per deterministic tick (runner-owned clock, never +wall-clock arithmetic): - state = env.observe() # WAGO input image in engineering units - decision = policy(state) # caller's plant model / agent (no policy here) - env.apply(actions) # decided outputs -> output image via - # the datastore's INTERNAL write paths - engine.pump() # engine advances exactly one tick + state = env.observe() # ALL modules: DI/DO/AI/AO channels, + # engineering units, names, parameters + values = env.model(state, dt) # plant physics (subclass hook) + env.apply(values) # new FIELD state -> DI/AI inputs; + # DO/AO via env.force() (master-side) -The engine always owns the tick clock (FR-4.3.9 deterministic sim -time); the env never reads wall-clock. `apply()` writes through the -datastore's Modbus-equivalent dispatch so mirrors, statistics, and -watchdog behavior stay consistent (FR-9.2.6 semantics). +`model()` is the override point for plant physics: read what the PLC +commanded (DO/AO), compute what the field now shows (DI/AI), return +it. `is_done(state)` gives an optional end criterion; the default is +False — environments run until manually stopped (SIGINT/SIGTERM). + +Attachments: degradation (§8.3), invariants (§8.4) and aliases +(scenario-continuous mode) hook onto the same tick when the env is +constructed on an engine. """ from __future__ import annotations +from collections.abc import Callable from dataclasses import dataclass, field +from ..modules import VirtualNode from .engine import ScenarioError @@ -31,43 +39,60 @@ class ChannelState: channel: int # 0-based channel on the module io_code: str # e.g. "DI0.0" name: str | None # configured application name, if any - value: object # bool for di/do, float for ai/ao + value: object # bool for di/do, int/float for ai/ao @dataclass class PlantState: - """Full observation: every channel + aliases + parameters + clock.""" + """Full observation: every channel of every module + attachments.""" sim_time_ms: float channels: list[ChannelState] = field(default_factory=list) aliases: dict[str, object] = field(default_factory=dict) parameters: dict[str, float] = field(default_factory=dict) + def by_code(self, io_code: str) -> object: + """Value of one channel by I/O code, or KeyError.""" + for ch in self.channels: + if ch.io_code == io_code: + return ch.value + raise KeyError(io_code) + + def by_kind(self, kind: str) -> dict[str, object]: + """{io_code: value} for all channels of one kind.""" + return {c.io_code: c.value for c in self.channels if c.kind == kind} + class PlantEnv: - """gym-style observe/apply environment (FR-8.2.14-16). + """gym-style plant environment (FR-8.2.14-17). - Constructed by the continuous engine; the engine stays the clock - owner — the env's `step()` convenience drives `engine.pump()` for - exactly one tick (server mode may tick the engine externally and - only use observe/apply). + Constructed on a VirtualNode (standalone) or on an engine (the + scenario-continuous layer). The runner owns the tick clock; `tick()` + advances sim-time deterministically by TICK_MS per call so both + paced (serve) and unpaced (sim) runners behave identically. """ - def __init__(self, engine) -> None: + TICK_MS = 10.0 + + def __init__(self, node: VirtualNode | None = None, *, + engine=None) -> None: self.engine = engine - self._actions: dict[str, object] = {} + self.node: VirtualNode = node if node is not None else engine.node + self.datastore = self.node.datastore + self.sim_time_ms = 0.0 + # attachments (optional; the scenario layer wires these) + self.degradation = getattr(engine, "degradation", None) + self.invariants = getattr(engine, "invariants", None) + self._aliases = getattr(engine, "_aliases", None) if engine else None + self._alias_value: Callable[[str], object] | None = ( + engine.alias_value if engine is not None else None) # -- observation (FR-8.2.14) ----------------------------------------------- def observe(self) -> PlantState: - """Snapshot of the generic inputs (WAGO input image + aliases). - - DI/AI channels hold the simulated field state; DO/AO hold the - PLC-written outputs as currently in the images — all values in - engineering units (alias scale/offset applied where defined). - """ - node = self.engine.node - ds = self.engine.datastore + """Snapshot ALL modules (DI/DO/AI/AO) in engineering units.""" + node = self.node + ds = self.datastore channels: list[ChannelState] = [] for kind, pool in (("di", node.di), ("do", node.do), ("ai", node.ai), ("ao", node.ao)): @@ -80,7 +105,8 @@ class PlantEnv: value = ds.get_input_word(loc.word) else: value = ds.get_output_word(loc.word) - code = node.io_code_at_slot(kind, loc.module_index, loc.channel - 1) + code = node.io_code_at_slot( + kind, loc.module_index, loc.channel - 1) channels.append(ChannelState( kind=kind, module=node.kind_index(loc.module_index)[kind], @@ -89,128 +115,226 @@ class PlantEnv: name=node.channel_names.get(code), value=value, )) - aliases = {name: self.engine.alias_value(name) - for name in self.engine._aliases} + aliases = ({name: self._alias_value(name) + for name in (self._aliases or {})} + if self._alias_value else {}) parameters: dict[str, float] = {} - degradation = getattr(self.engine, "degradation", None) - if degradation is not None: - parameters = {n: m.value for n, m in degradation.models.items()} + if self.degradation is not None: + parameters = {n: m.value + for n, m in self.degradation.models.items()} return PlantState( - sim_time_ms=self.engine.sim_time_ms, + sim_time_ms=self.sim_time_ms, channels=channels, aliases=aliases, parameters=parameters, ) - # -- actions (FR-8.2.14) ----------------------------------------------------- + # -- plant model hook (FR-8.2.16) ------------------------------------------- - def apply(self, actions: dict[str, object]) -> None: - """Queue decided outputs; they land when the engine ticks. + def model(self, state: PlantState, dt_s: float) -> dict[str, object]: + """Plant physics: react to PLC commands, compute the new field. - Keys: I/O code ("DO0.0") or alias name ("pump_relay"). Writing - a DI/AI through the env is an error — inputs are the field side - (datasets/scenario steps simulate them). Keys are validated - EAGERLY so a bad decision surfaces immediately, not at flush. + Override in a subclass (FR-8.2.17). Receives the full observed + state and the tick delta in seconds; returns {I/O code or alias: + new value} for the channels it drives — DI/AI normally (the + simulated sensors); DO/AO entries are treated as forcing and go + through the master-side write paths. Returning {} (the default) + makes this a pure observation environment. """ - if not isinstance(actions, dict): - raise ScenarioError("apply() actions must be a mapping") - validated: dict[str, object] = {} - for key, value in actions.items(): - alias = self.engine._aliases.get(key) or self._alias_by_code(key) - if alias is None: - raise ScenarioError(f"apply(): unknown channel {key!r}") - if alias.kind not in ("do", "ao"): - raise ScenarioError( - f"apply(): {key!r} is an input alias ({alias.kind}); " - "only outputs (do/ao) are decided") - validated[str(key)] = value - self._actions.update(validated) + return {} - def flush(self) -> None: - """Write queued actions into the output image (engine-tick step). + def is_done(self, state: PlantState) -> bool: + """Optional end criterion; default runs until manually stopped.""" + return False - Called by the engine after the observe/decide window and before - the pipeline advances — so the next observe() sees them - (FR-8.2.12 atomic-env-tick paragraph). + # -- actions (FR-8.2.15) ------------------------------------------------------ + + def apply(self, values: dict[str, object], *, force: bool = False) -> None: + """Write new channel values into the process images. + + DI/AI keys (the field) write directly — this is the primary + direction. DO/AO keys (PLC outputs) require force=True and go + through the datastore's internal Modbus dispatch so mirrors, + statistics, and watchdog stay consistent (FR-9.2.6) — for + headless runs and fault setup only. + Keys: I/O code ("AI0.0") or alias name. Unknown keys raise + immediately (FR-8.2.16). """ - for key, value in self._actions.items(): - self._write_one(key, value) - self._actions = {} + if not isinstance(values, dict): + raise ScenarioError("apply() values must be a mapping") + for key, value in values.items(): + self._write_one(key, value, force=force) - def _write_one(self, key: str, value: object) -> None: - # resolve by alias name first, then by I/O code - alias = self.engine._aliases.get(key) - if alias is None: - alias = self._alias_by_code(key) - if alias is None: - raise ScenarioError(f"apply(): unknown channel {key!r}") - if alias.kind not in ("do", "ao"): + def force(self, values: dict[str, object]) -> None: + """Master-side forcing of DO/AO (FR-8.2.15 explicit form).""" + self.apply(values, force=True) + + def _resolve(self, key: str): + """-> (kind, ChannelLocation, alias_or_None) for a code or alias.""" + # alias name first (scenario layer) + if self._aliases and key in self._aliases and self.engine is not None: + alias = self._aliases[key] + return alias.kind, self.engine._alias_location(alias), alias + # any channel by I/O code (per-type codes, FR-4.3.11 v1.5) + code_up = str(key).strip().upper() + node = self.node + for kind, pool in (("di", node.di), ("do", node.do), + ("ai", node.ai), ("ao", node.ao)): + for loc in pool: + if node.io_code_at_slot( + kind, loc.module_index, loc.channel - 1) == code_up: + return kind, loc, None + raise ScenarioError(f"apply(): unknown channel {key!r}") + + def _write_one(self, key: str, value: object, *, force: bool) -> None: + kind, loc, alias = self._resolve(key) + ds = self.datastore + if kind in ("do", "ao") and not force: raise ScenarioError( - f"apply(): {key!r} is an input alias ({alias.kind}); " - "only outputs (do/ao) are decided") - loc = self.engine._alias_location(alias) - ds = self.engine.datastore - if alias.kind == "do": + f"apply(): {key!r} is an output ({kind.upper()}); use " + "force=True / env.force() for master-side forcing") + scale = getattr(alias, "scale", 1.0) or 1.0 + offset = getattr(alias, "offset", 0.0) or 0.0 + if kind == "di": + ds.set_digital_input_channel(loc.bit + 1, bool(value)) + elif kind == "ai": + raw = int(float(value) * scale + offset) & 0xFFFF + ds.set_input_word(loc.word, raw) + elif kind == "do": result = ds.write_bits(loc.bit, [bool(value)]) - else: - numeric = float(value) # type: ignore[arg-type] - raw = int(numeric * alias.scale + alias.offset) & 0xFFFF + if isinstance(result, Exception): + raise ScenarioError(f"force() write to {key!r}: {result}") + else: # ao + raw = int(float(value) * scale + offset) & 0xFFFF result = ds.write_words(loc.word, [raw]) - if isinstance(result, Exception): - raise ScenarioError(f"apply() write to {key!r} rejected: {result}") + if isinstance(result, Exception): + raise ScenarioError(f"force() write to {key!r}: {result}") - def _alias_by_code(self, code: str): - """Find an alias whose I/O code matches (per-type codes).""" - code_up = str(code).strip().upper() - node = self.engine.node - for name, alias in self.engine._aliases.items(): - if alias.kind not in ("do", "ao"): - continue - try: - loc = self.engine._alias_location(alias) - except Exception: - continue - if node.io_code_at_slot(alias.kind, loc.module_index, - loc.channel - 1) == code_up: - return alias - return None + # -- one full tick (FR-8.2.14 cycle) --------------------------------------------- + + def tick(self, dt_ms: float | None = None) -> PlantState: + """One environment tick: model() -> apply(field) -> advance. + + Deterministic: sim-time advances by TICK_MS (or dt_ms) per call + — never from wall-clock. Attachments advance with the tick: + degradation (§8.3) then invariants (§8.4), FR-8.2.12 order. + """ + dt = dt_ms if dt_ms is not None else self.TICK_MS + state = self.observe() + if self.is_done(state): + return state + values = self.model(state, dt / 1000.0) + if values: + self.apply(values) + self.sim_time_ms += dt + if self.degradation is not None: + self.degradation.tick(dt) + if self.invariants is not None: + self.invariants.tick() + return self.observe() + + # -- episode control (FR-8.2.16) ------------------------------------------------ + + def reset(self) -> PlantState: + """Fresh episode: zero images, reset wear + parameters + clock + (deterministic — identical runs replay identically, NFR-8.7.3).""" + self.datastore.reset_images() + self.sim_time_ms = 0.0 + if self.degradation is not None: + for model in self.degradation.models.values(): + model.reset() + model.diagnosed = False + self.degradation._last_truth = {} + if self.invariants is not None: + self.invariants.episodes = [] + for inv in self.invariants.invariants: + inv.active = None + for rl in self.invariants.rate_limits: + rl.samples = [] + rl.breached = False + rl.peak_rate = 0.0 + self.invariants.snapshots = [] + self.invariants._last_snapshot_ms = 0.0 + return self.observe() # -- clock (FR-8.2.16) --------------------------------------------------------- @property def tick_ms(self) -> float: - return float(self.engine.TICK_MS) - - @property - def sim_time_ms(self) -> float: - return self.engine.sim_time_ms - - # -- episode control (FR-8.2.16) ------------------------------------------------ - - def reset(self, config: dict | None = None) -> PlantState: - """Fresh episode: zero images, reset wear + parameters + clock - (deterministic — identical runs replay identically, NFR-8.7.3).""" - self.engine.reset_episode() - return self.observe() - - def step(self, actions: dict[str, object] | None = None): - """One full env tick: apply -> engine pump -> observe (FR-8.2.14). - - Convenience for headless agents; server mode ticks the engine - on its own scheduler and uses observe/apply separately. - """ - if actions: - self.apply(actions) - self.engine.pump() - return self.observe() - - # -- §8.3/§8.4 machinery (FR-8.2.16) ---------------------------------------------- + return self.TICK_MS @property def parameters(self) -> dict[str, float]: - deg = getattr(self.engine, "degradation", None) - return {n: m.value for n, m in deg.models.items()} if deg else {} + if self.degradation is None: + return {} + return {n: m.value for n, m in self.degradation.models.items()} - @property - def invariants(self): - return getattr(self.engine, "invariants", None) \ No newline at end of file + +def load_env(ref: str, node: VirtualNode | None = None, *, + engine=None) -> PlantEnv: + """Load an environment from a module reference (FR-8.2.17). + + Accepts ``pkg.mod.ClassName``, ``pkg.mod.factory_fn`` (returns a + PlantEnv), or a bare ``pkg.mod`` whose module defines either a + class named ``Env`` or a function ``make_env(...)``. + """ + import importlib + + parts = ref.split(":") + attr_path = None + mod = None + leaf = None + if len(parts) == 2: + # explicit module:attr form — ref is the WHOLE module path + attr_path = parts[1] or None + mod_name = parts[0] + else: + # no colon: try the whole ref as a module first, then as + # pkg.mod containing an attr (e.g. ...:Class omitted) + try: + mod = importlib.import_module(ref) + mod_name = ref + except ImportError: + mod_name, _, leaf = ref.rpartition(".") + if not mod_name or ("." not in mod_name and attr_path is None + and leaf is None): + raise ScenarioError(f"--env {ref!r}: reference must include a module path") + + if mod is None: + try: + mod = importlib.import_module(mod_name) + except ImportError as exc: + raise ScenarioError( + f"--env {ref!r}: cannot import {mod_name!r}: {exc}") + + obj = None + if attr_path is not None: + # explicit module:attr — must exist; never fall back silently + if not hasattr(mod, attr_path): + raise ScenarioError( + f"--env {ref!r}: no {attr_path!r} in {mod_name!r}") + obj = getattr(mod, attr_path) + elif leaf and hasattr(mod, leaf): + obj = getattr(mod, leaf) + elif hasattr(mod, "Env"): + obj = getattr(mod, "Env") + elif hasattr(mod, "make_env"): + obj = getattr(mod, "make_env") + else: + raise ScenarioError( + f"--env {ref!r}: no environment found in {mod_name!r} " + "(expected a class, Env, or make_env())") + + env: PlantEnv + if isinstance(obj, type) and issubclass(obj, PlantEnv): + env = obj(node=node, engine=engine) + elif callable(obj): + made = obj(node=node, engine=engine) if node is not None \ + else obj(engine=engine) + if not isinstance(made, PlantEnv): + raise ScenarioError(f"--env {ref!r}: did not return a PlantEnv") + env = made + else: + raise ScenarioError( + f"--env {ref!r}: not a PlantEnv subclass or factory") + return env \ No newline at end of file diff --git a/tests/test_continuous.py b/tests/test_continuous.py index 840b111..02ec47a 100644 --- a/tests/test_continuous.py +++ b/tests/test_continuous.py @@ -9,6 +9,7 @@ import pytest from wago_sim.scenario.datasets import CsvDataset from wago_sim.scenario.engine import DataError, ScenarioEngine, ScenarioError, Verdict +from wago_sim.scenario.plantenv import PlantEnv def _write_csv(path, text: str) -> str: @@ -326,6 +327,8 @@ class TestInvariants: class TestPlantEnv: + """PlantEnv v2 (FR-8.2.14-17): standalone-first, apply -> DI/AI.""" + def _scenario(self): return { "mode": "continuous", "name": "x", "until": "500ms", @@ -334,66 +337,130 @@ class TestPlantEnv: } def test_observe_returns_all_channels(self, node): - engine = _engine(node, self._scenario()) - state = engine.env.observe() + # standalone construction: no engine needed + env = PlantEnv(node=node) + state = env.observe() kinds = {c.kind for c in state.channels} assert kinds == {"di", "do", "ai", "ao"} codes = [c.io_code for c in state.channels] assert "DI0.0" in codes and "DO0.0" in codes - assert state.aliases["start"] is False assert state.sim_time_ms == 0.0 - def test_apply_writes_output_before_next_observe(self, node): - engine = _engine(node, self._scenario()) - engine.env.apply({"pump": True}) - state = engine.env.step() # flush + pump + observe - pump_states = [c for c in state.channels if c.io_code == "DO0.0"] - assert pump_states[0].value is True - assert state.aliases["pump"] is True + def test_apply_writes_input_side(self, node): + env = PlantEnv(node=node) + env.apply({"DI0.0": True, "AI0.0": 1234}) + state = env.observe() + assert state.by_code("DI0.0") is True + assert state.by_code("AI0.0") == 1234 - def test_apply_by_io_code(self, node): - engine = _engine(node, self._scenario()) - engine.env.apply({"DO0.0": True}) - state = engine.env.step() - assert state.aliases["pump"] is True - - def test_apply_input_rejected(self, node): - engine = _engine(node, self._scenario()) - with pytest.raises(ScenarioError, match="input alias"): - engine.env.apply({"start": True}) + def test_apply_output_requires_force(self, node): + env = PlantEnv(node=node) + with pytest.raises(ScenarioError, match="force"): + env.apply({"DO0.0": True}) + # explicit forcing works and lands in the output image + env.force({"DO0.0": True, "AO0.0": 999}) + state = env.observe() + assert state.by_code("DO0.0") is True + assert state.by_code("AO0.0") == 999 def test_apply_unknown_rejected(self, node): - engine = _engine(node, self._scenario()) + env = PlantEnv(node=node) with pytest.raises(ScenarioError, match="unknown channel"): - engine.env.apply({"nonexistent": True}) + env.apply({"nonexistent": True}) + + def test_model_hook_drives_field(self, node): + """The intended loop: model() reads PLC outputs, returns field.""" + + class Tank(PlantEnv): + def model(self, state, dt_s): + pump_on = bool(state.by_code("DO0.0")) + level = float(state.by_code("AI0.0")) + return {"AI0.0": level + (5.0 if pump_on else 1.0)} + + env = Tank(node=node) + env.force({"DO0.0": True}) # the "PLC" turns the pump on + state = env.tick() + assert state.by_code("AI0.0") == 5.0 # field advanced via model + state = env.tick() + assert state.by_code("AI0.0") == 10.0 + assert state.sim_time_ms == 20.0 + + def test_is_done_stops_ticks(self, node): + class DoneAfterTwo(PlantEnv): + def __init__(self, *a, **kw): + super().__init__(*a, **kw) + self.n = 0 + + def is_done(self, state): + return self.n >= 2 + + def model(self, state, dt_s): + self.n += 1 + return {} + + env = DoneAfterTwo(node=node) + env.tick() + env.tick() + state = env.tick() # done: no further advance + assert env.n == 2 + assert state.sim_time_ms == 20.0 def test_reset_deterministic_replay(self, node): - engine = _engine(node, self._scenario()) - engine.env.apply({"pump": True}) - for _ in range(10): - engine.pump() - assert engine.sim_time_ms > 0 - engine.env.reset() - assert engine.sim_time_ms == 0.0 - state = engine.env.observe() - assert state.aliases["pump"] is False + env = PlantEnv(node=node) + env.apply({"DI0.0": True, "AI0.0": 42}) + env.tick() + assert env.sim_time_ms > 0 + state = env.reset() + assert env.sim_time_ms == 0.0 + assert state.by_code("DI0.0") is False + assert state.by_code("AI0.0") == 0 - def test_two_tick_cycle_deterministic(self, node): - """Acceptance criterion 12: copy an input to an output; the - second observation contains it (FR-8.2.12 pipeline consistency).""" + def test_engine_attached_env_uses_aliases(self, node): + """Scenario layer: alias names + alias scale/offset still work.""" engine = _engine(node, self._scenario()) - state1 = engine.env.observe() # observe: DI0.0 still off - di_then = next(c for c in state1.channels if c.io_code == "DI0.0") - assert di_then.value is False - # field-side change + decision AFTER observing, before the tick - engine.datastore.set_digital_input_channel(1, True) - engine.env.apply({"pump": di_then.value}) # decide on observed state - state2 = engine.env.step() # flush + tick + observe - di_now = next(c for c in state2.channels if c.io_code == "DI0.0") - assert di_now.value is True # visible next observe - # decided output applied from the OBSERVED state: pump still off - do_now = next(c for c in state2.channels if c.io_code == "DO0.0") - assert do_now.value is False + env = engine.env + state = env.observe() + assert state.aliases["start"] is False + # alias write to a do output requires force (PLC-side semantics) + with pytest.raises(ScenarioError, match="force"): + env.apply({"pump": True}) + env.force({"pump": True}) + assert env.observe().aliases["pump"] is True + + def test_pump_station_example_env(self, node): + """The shipped example env loads and runs closed-loop physics.""" + from envs.pump_station_env import PumpStationEnv + env = PumpStationEnv(node=node) + # PLC turns the pump on at full speed + env.force({"DO0.0": True, "AO0.0": 4095}) + env.tick() # physics applies the level + level0 = env.observe().by_code("AI0.0") + assert level0 > 0 # tank starts at 3000 mm + # big dt (10 sim-minutes): the pump visibly drains the tank + env.tick(dt_ms=600_000) + level1 = env.observe().by_code("AI0.0") + assert level1 < level0 # pump drains the tank + # pump OFF: inflow fills the tank instead + env.force({"DO0.0": False}) + env.tick(dt_ms=600_000) + level2 = env.observe().by_code("AI0.0") + assert level2 > level1 + # and it runs forever by default + assert env.is_done(env.observe()) is False + + def test_load_env_class_and_factory(self, node): + from wago_sim.scenario.plantenv import load_env + env = load_env("envs.pump_station_env:PumpStationEnv", node=node) + assert isinstance(env, PlantEnv) + env = load_env("envs.pump_station_env:make_env", node=node) + assert isinstance(env, PlantEnv) + # bare module picks up Env + env = load_env("envs.pump_station_env", node=node) + assert isinstance(env, PlantEnv) + with pytest.raises(ScenarioError): + load_env("envs.pump_station_env:Nope", node=node) + with pytest.raises(ScenarioError): + load_env("envs.missing_module", node=node) class TestTickPipeline: