"""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())