From c8a990a3f60316ef4155d5fd993c133d040d33ec Mon Sep 17 00:00:00 2001 From: scottkalw Date: Tue, 3 Nov 2026 18:11:40 -0600 Subject: [PATCH 1/4] fsmes scenario: play and check timed tag values against any OPC UA server A scenario file (JSON) names tags once, then says when each takes which value, with a ramp and a product-through-stations shorthand. play writes each value as the node's own type at its time; check names the line of each mistake and, connected, reports type, access and the driver rule. Only Simulator or Memory Based channels, declared in the file, are written; Kepware showing a serial or Ethernet device behind a channel refuses it regardless, and a read-only tag is reported. Nothing is written unless every target passes. serve holds a scenario's tags as writable memory tags for rehearsal. Co-Authored-By: Claude Opus 5.5 Signed-off-by: scottkalw --- src/fsmes/cli.py | 197 +++++++ src/fsmes/integrations/opc/scenario.py | 773 +++++++++++++++++++++++++ tests/test_scenario_player.py | 325 +++++++++++ 3 files changed, 1295 insertions(+) create mode 100644 src/fsmes/integrations/opc/scenario.py create mode 100644 tests/test_scenario_player.py diff --git a/src/fsmes/cli.py b/src/fsmes/cli.py index c846c7ad..02287b3c 100644 --- a/src/fsmes/cli.py +++ b/src/fsmes/cli.py @@ -2309,6 +2309,203 @@ 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 + 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..6966af74 --- /dev/null +++ b/src/fsmes/integrations/opc/scenario.py @@ -0,0 +1,773 @@ +"""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 + +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.""" + 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/tests/test_scenario_player.py b/tests/test_scenario_player.py new file mode 100644 index 00000000..ce13d564 --- /dev/null +++ b/tests/test_scenario_player.py @@ -0,0 +1,325 @@ +"""`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 +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 + + +# --------------------------------------------------------------- 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" From d15d2b140978f1a94e3916809414e13db4f7b911 Mon Sep 17 00:00:00 2001 From: scottkalw Date: Tue, 3 Nov 2026 18:11:40 -0600 Subject: [PATCH 2/4] Scenarios: the docs page, Kepware notes, and two KepSim examples A product through the six KepSim stations on a scratch Simulator channel, and a set-and-ramp drill on the replay server's writable setpoints. Co-Authored-By: Claude Opus 5.5 Signed-off-by: scottkalw --- docs/operate/scenarios.md | 179 ++++++++++++++++++ labs/kepsim/README.md | 1 + .../scenarios/product-through-the-line.json | 30 +++ labs/kepsim/scenarios/setpoint-drill.json | 17 ++ mkdocs.yml | 1 + 5 files changed, 228 insertions(+) create mode 100644 docs/operate/scenarios.md create mode 100644 labs/kepsim/scenarios/product-through-the-line.json create mode 100644 labs/kepsim/scenarios/setpoint-drill.json diff --git a/docs/operate/scenarios.md b/docs/operate/scenarios.md new file mode 100644 index 00000000..cb5d88da --- /dev/null +++ b/docs/operate/scenarios.md @@ -0,0 +1,179 @@ +# 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. + +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 From 13c82cf0fe1ae99eec5c713f9e7edd581367c1e2 Mon Sep 17 00:00:00 2001 From: scottkalw Date: Tue, 3 Nov 2026 18:27:08 -0600 Subject: [PATCH 3/4] Shadow mode refuses fsmes scenario play; both scenario paths in the register The player changes tags on another server, so in shadow mode it is refused before it connects (the CLI) and again at the call site (play). The rehearsal server is registered as allowed: its own loopback server. The register's scan now matches .write_attribute( as an OPC UA node write; without it the player's write was invisible to the ratchet. Co-Authored-By: Claude Opus 5.5 Signed-off-by: scottkalw --- docs/design/agentic-harness.md | 2 +- docs/operate/scenarios.md | 4 ++++ src/fsmes/cli.py | 6 ++++++ src/fsmes/integrations/opc/scenario.py | 6 +++++- src/fsmes/shadow.py | 20 +++++++++++++++++++- tests/test_scenario_player.py | 18 ++++++++++++++++++ tests/test_shadow_mode.py | 21 ++++++++++++++++++++- 7 files changed, 73 insertions(+), 4 deletions(-) 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 index cb5d88da..5e3abfa5 100644 --- a/docs/operate/scenarios.md +++ b/docs/operate/scenarios.md @@ -41,6 +41,10 @@ them fails: `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)). diff --git a/src/fsmes/cli.py b/src/fsmes/cli.py index 02287b3c..30ffdfe3 100644 --- a/src/fsmes/cli.py +++ b/src/fsmes/cli.py @@ -2441,6 +2441,12 @@ def scenario_play( typer.echo(f"--from {exc}") raise typer.Exit(2) from None settings = _scenario_settings(endpoint, security, user, password) + from fsmes import shadow + + if shadow.enabled(settings): + 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) async def go() -> int: client = await _scenario_connect(settings) diff --git a/src/fsmes/integrations/opc/scenario.py b/src/fsmes/integrations/opc/scenario.py index 6966af74..7468e0fa 100644 --- a/src/fsmes/integrations/opc/scenario.py +++ b/src/fsmes/integrations/opc/scenario.py @@ -38,6 +38,8 @@ 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]=") @@ -668,7 +670,9 @@ async def play(client: Client, scenario: Scenario, targets: dict[str, Target], * """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.""" + 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]) 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 index ce13d564..83b0921c 100644 --- a/tests/test_scenario_player.py +++ b/tests/test_scenario_player.py @@ -105,6 +105,24 @@ def test_check_without_a_server_says_what_is_wrong_and_exits_2(tmp_path): 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() + 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() + assert result.exit_code == 1 + assert "Nothing written" in result.output and "Shadow mode" in result.output + + # --------------------------------------------------------------- the safety 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*\(|" From faaa95d27272b25533f084f6e621605020e4104e Mon Sep 17 00:00:00 2001 From: scottkalw Date: Tue, 3 Nov 2026 19:01:52 -0600 Subject: [PATCH 4/4] scenario play: check shadow mode before setting up logging The shadow-mode CLI test reached _scenario_settings, whose setup_logging("ERROR") stayed configured for the rest of the test process; three floor tests that capture structlog events then saw none on CI. The refusal now comes first, and the test restores structlog's config. Co-Authored-By: Claude Opus 5.5 Signed-off-by: scottkalw --- src/fsmes/cli.py | 4 ++-- tests/test_scenario_player.py | 3 +++ 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/src/fsmes/cli.py b/src/fsmes/cli.py index 30ffdfe3..f9a099ff 100644 --- a/src/fsmes/cli.py +++ b/src/fsmes/cli.py @@ -2440,13 +2440,13 @@ def scenario_play( except ValueError as exc: typer.echo(f"--from {exc}") raise typer.Exit(2) from None - settings = _scenario_settings(endpoint, security, user, password) from fsmes import shadow - if shadow.enabled(settings): + 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) diff --git a/tests/test_scenario_player.py b/tests/test_scenario_player.py index 83b0921c..67e6bfff 100644 --- a/tests/test_scenario_player.py +++ b/tests/test_scenario_player.py @@ -13,6 +13,7 @@ from pathlib import Path import pytest +import structlog from asyncua import Client, Server, ua from typer.testing import CliRunner @@ -112,6 +113,7 @@ def test_play_in_shadow_mode_writes_nothing_and_does_not_connect(monkeypatch): 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. @@ -119,6 +121,7 @@ def test_play_in_shadow_mode_writes_nothing_and_does_not_connect(monkeypatch): "--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