secs-gem/interop/virtual_fab.py

"""A randomized virtual fab.

N secs_gemd equipment instances (the `fab` compose service) each get TWO
independent actors attached:

  - a secsgem-py GemHostHandler playing the fab host over HSMS, and
  - a secsgem_client Equipment playing the tool software over gRPC,

then a seeded random scenario drives all of them concurrently: variable
writes with read-back verification, event fires that must arrive at the
host as S6F11, alarm set/clear that must arrive as S5F1, host S1F3 polls,
and host S2F41 commands that must come back HCACK=4 and reach the tool's
handler. Every violated invariant is recorded; the seed is printed so any
failure is reproducible.

    python3 virtual_fab.py --host fab --n 3 --seconds 20 [--seed 1234]

Exit 0 = every invariant held on every equipment.
"""

from __future__ import annotations

import argparse
import logging
import random
import sys
import threading
import time

sys.path.insert(0, "/app/clients/python")

import secsgem.common
import secsgem.gem
import secsgem.hsms
import secsgem.secs

from secsgem_client import Equipment, SecsGemError

LOG = logging.getLogger("fab")
F = secsgem.secs.functions


class EquipmentActor:
    """One equipment: its host, its tool, its tallies, its verdicts."""

    def __init__(self, idx: int, host: str, hsms_port: int, grpc_port: int,
                 rng: random.Random):
        self.idx = idx
        self.rng = rng
        self.violations: list[str] = []
        self.s6f11_seen = 0
        self.s5f1_seen = 0
        self.events_fired = 0
        self.alarm_ops = 0
        self.commands_sent = 0
        self.commands_received = 0
        self.ops = 0

        self.tool = Equipment(f"{host}:{grpc_port}")

        @self.tool.on("*")
        def _any(cmd):  # noqa: ANN001
            self.commands_received += 1
            self.tool.fire("ProcessStarted")  # completion signal

        self.tool.listen(background=True)

        settings = secsgem.hsms.HsmsSettings(
            address=host, port=hsms_port, session_id=0,
            connect_mode=secsgem.hsms.HsmsConnectMode.ACTIVE,
            device_type=secsgem.common.DeviceType.HOST)
        self.host = secsgem.gem.GemHostHandler(settings)
        self.host.register_stream_function(6, 11, self._on_s6f11)
        self.host.register_stream_function(5, 1, self._on_s5f1)
        self.host.enable()

    def _on_s6f11(self, _h, message):
        self.s6f11_seen += 1
        self.host.send_response(F.SecsS06F12(0), message.header.system)

    def _on_s5f1(self, _h, message):
        self.s5f1_seen += 1
        self.host.send_response(F.SecsS05F02(0), message.header.system)

    def violate(self, what: str) -> None:
        self.violations.append(what)
        LOG.error("[eq%d] VIOLATION: %s", self.idx, what)

    def setup(self) -> bool:
        if not self.host.waitfor_communicating(timeout=20):
            self.violate("HSMS establish-communications failed")
            return False
        self.host.send_and_waitfor_response(F.SecsS01F17())
        self.host.send_and_waitfor_response(
            F.SecsS02F33({"DATAID": 1, "DATA": [{"RPTID": 1, "VID": [101]}]}))
        self.host.send_and_waitfor_response(
            F.SecsS02F35({"DATAID": 1, "DATA": [{"CEID": 300, "RPTID": [1]}]}))
        self.host.send_and_waitfor_response(
            F.SecsS02F37({"CEED": True, "CEID": [300]}))
        self.host.send_and_waitfor_response(
            F.SecsS05F03({"ALED": 0x80, "ALID": 1}))
        return True

    def step(self) -> None:
        """One random action with its invariant."""
        self.ops += 1
        roll = self.rng.random()
        try:
            if roll < 0.30:
                v = round(self.rng.uniform(0.1, 9.9), 3)
                self.tool.set(ChamberPressure=v)
                got = self.tool.get("ChamberPressure")["ChamberPressure"]
                if abs(got - v) > 1e-3:
                    self.violate(f"round-trip: set {v} got {got}")
            elif roll < 0.45:
                self.tool.fire("ProcessStarted")
                self.events_fired += 1
            elif roll < 0.60:
                if self.rng.random() < 0.5:
                    self.tool.alarm("chiller_temp_high")
                else:
                    self.tool.clear("chiller_temp_high")
                self.alarm_ops += 1
            elif roll < 0.80:
                rsp = self.host.send_and_waitfor_response(F.SecsS01F03([]))
                body = self.host.settings.streams_functions.decode(rsp).get()
                if not isinstance(body, list) or len(body) < 1:
                    self.violate(f"S1F3 poll returned {body!r}")
            else:
                rsp = self.host.send_and_waitfor_response(
                    F.SecsS02F41({"RCMD": "START", "PARAMS": []}))
                body = self.host.settings.streams_functions.decode(rsp).get()
                hcack = int(body.get("HCACK", -1)) if isinstance(body, dict) else -1
                self.commands_sent += 1
                # A tool is subscribed for the whole run: must be HCACK 4.
                if hcack != 4:
                    self.violate(f"S2F41 with subscriber -> HCACK {hcack} (want 4)")
        except SecsGemError as e:
            self.violate(f"client raised: {e}")
        except Exception as e:  # noqa: BLE001
            self.violate(f"actor raised: {type(e).__name__}: {e}")

    def finish(self) -> None:
        time.sleep(2.0)  # let in-flight S6F11/S5F1 land
        if self.events_fired + self.commands_received > 0 and self.s6f11_seen == 0:
            self.violate("events fired but host never received S6F11")
        if self.alarm_ops > 0 and self.s5f1_seen == 0:
            self.violate("alarms toggled but host never received S5F1")
        if self.commands_sent > 0 and self.commands_received == 0:
            self.violate("commands sent but tool handler never ran")
        try:
            self.host.disable()
            self.tool.close()
        except Exception:  # noqa: BLE001
            pass


def run(host: str, n: int, seconds: float, seed: int) -> int:
    LOG.info("virtual fab: %d equipment, %.0fs, seed=%d", n, seconds, seed)
    actors = [EquipmentActor(i, host, 5100 + i, 51000 + i,
                             random.Random(seed + i)) for i in range(n)]
    if not all(a.setup() for a in actors):
        return 1

    deadline = time.monotonic() + seconds

    def drive(a: EquipmentActor) -> None:
        while time.monotonic() < deadline and len(a.violations) < 5:
            a.step()
            time.sleep(a.rng.uniform(0.01, 0.10))

    threads = [threading.Thread(target=drive, args=(a,)) for a in actors]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    for a in actors:
        a.finish()

    bad = 0
    for a in actors:
        LOG.info("[eq%d] ops=%d cmds=%d/%d s6f11=%d s5f1=%d violations=%d",
                 a.idx, a.ops, a.commands_received, a.commands_sent,
                 a.s6f11_seen, a.s5f1_seen, len(a.violations))
        bad += len(a.violations)
    if bad:
        LOG.error("FAB FAILED: %d violations (seed=%d to reproduce)", bad, seed)
        return 1
    LOG.info("virtual fab: all invariants held on all %d equipment (seed=%d)",
             n, seed)
    return 0


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--host", default="fab")
    ap.add_argument("--n", type=int, default=3)
    ap.add_argument("--seconds", type=float, default=20.0)
    ap.add_argument("--seed", type=int, default=None)
    args = ap.parse_args()
    logging.basicConfig(level=logging.INFO, format="%(message)s")
    for noisy in ("communication", "hsms_connection", "secsgem"):
        logging.getLogger(noisy).setLevel(logging.WARNING)
    seed = args.seed if args.seed is not None else random.SystemRandom().randrange(1 << 31)
    return run(args.host, args.n, args.seconds, seed)


if __name__ == "__main__":
    raise SystemExit(main())