"""End-to-end interop for secs_gemd: drive the daemon through BOTH faces. A gRPC tool client and a secsgem-py active host both connect to one running secs_gemd. This proves the daemon bridges gRPC <-> HSMS against a reference GEM host, not just that it compiles: - gRPC GetControlState agrees with the HSMS-driven control state - gRPC SetVariables(ChamberPressure=2.5) lands in the model, and - gRPC FireEvent(ProcessStarted) makes the host receive S6F11 CEID 300 carrying that 2.5 (so the value flowed gRPC -> engine -> HSMS -> host) - unknown variable / event names are rejected at the gRPC edge Exits 0 on success, non-zero on any failure. """ from __future__ import annotations import argparse import logging import os import sys import tempfile import threading import time import grpc import secsgem.common import secsgem.gem import secsgem.hsms import secsgem.secs LOG = logging.getLogger("daemon-interop") F = secsgem.secs.functions def gen_stubs(proto_dir: str): """Generate flat Python stubs from equipment.proto (proto package is 'secsgem.v1', which would collide with the secsgem-py package if we kept the nested path — so generate flat and import directly).""" from grpc_tools import protoc out = tempfile.mkdtemp(prefix="gemd_pb_") rc = protoc.main([ "protoc", f"-I{proto_dir}", f"--python_out={out}", f"--grpc_python_out={out}", os.path.join(proto_dir, "equipment.proto"), ]) if rc != 0: raise SystemExit("protoc stub generation failed") sys.path.insert(0, out) import equipment_pb2 as pb # noqa: E402 import equipment_pb2_grpc as pb_grpc # noqa: E402 return pb, pb_grpc def _scalars(body): """Flatten a decoded secsgem-py body (nested dict/list) to scalar leaves.""" if isinstance(body, dict): for v in body.values(): yield from _scalars(v) elif isinstance(body, (list, tuple)): for v in body: yield from _scalars(v) else: yield body def run(grpc_addr: str, hsms_host: str, hsms_port: int, session_id: int, proto_dir: str) -> int: pb, pb_grpc = gen_stubs(proto_dir) failures: list[str] = [] def check(label: str, ok: bool, detail: str = "") -> None: LOG.info("[%s] %s%s", "OK " if ok else "FAIL", label, f" — {detail}" if detail else "") if not ok: failures.append(f"{label}: {detail or 'failed'}") # ---- gRPC tool side ---- LOG.info("connecting gRPC tool to %s", grpc_addr) channel = grpc.insecure_channel(grpc_addr) try: grpc.channel_ready_future(channel).result(timeout=15) except grpc.FutureTimeoutError: check("gRPC channel ready", False, f"no daemon at {grpc_addr}") return 1 stub = pb_grpc.EquipmentStub(channel) # ---- secsgem-py host side ---- settings = secsgem.hsms.HsmsSettings( address=hsms_host, port=hsms_port, session_id=session_id, connect_mode=secsgem.hsms.HsmsConnectMode.ACTIVE, device_type=secsgem.common.DeviceType.HOST, ) client = secsgem.gem.GemHostHandler(settings) ceid300 = threading.Event() last_s6f11 = {} s5f1_seen = threading.Event() last_s5f1 = {} def on_s6f11(_handler, message): decoded = client.settings.streams_functions.decode(message) body = decoded.get() ceid = body.get("CEID") if isinstance(body, dict) else None LOG.info("[evt] S6F11 CEID=%s body=%r", ceid, body) if ceid == 300: last_s6f11["body"] = body ceid300.set() client.send_response(F.SecsS06F12(0), message.header.system) def on_s5f1(_handler, message): decoded = client.settings.streams_functions.decode(message) body = decoded.get() LOG.info("[alm] S5F1 body=%r", body) if isinstance(body, dict): last_s5f1.update(body) s5f1_seen.set() client.send_response(F.SecsS05F02(0), message.header.system) client.register_stream_function(6, 11, on_s6f11) client.register_stream_function(5, 1, on_s5f1) client.enable() try: if not client.waitfor_communicating(timeout=15): check("HSMS establish-communications", False) return 1 check("HSMS establish-communications", True) # Go online so control state -> OnlineRemote. client.send_and_waitfor_response(F.SecsS01F17()) time.sleep(0.3) # gRPC GetControlState should agree with the HSMS-driven state. cs = stub.GetControlState(pb.Empty()) check("gRPC GetControlState == ONLINE_REMOTE", cs.state == pb.ControlState.ONLINE_REMOTE, pb.ControlState.State.Name(cs.state)) # Configure an event report: RPTID 1 = [VID 101], CEID 300 -> RPTID 1. client.send_and_waitfor_response( F.SecsS02F33({"DATAID": 1, "DATA": [{"RPTID": 1, "VID": [101]}]})) client.send_and_waitfor_response( F.SecsS02F35({"DATAID": 1, "DATA": [{"CEID": 300, "RPTID": [1]}]})) client.send_and_waitfor_response( F.SecsS02F37({"CEED": True, "CEID": [300]})) check("host configured report CEID 300 -> RPTID 1 (VID 101)", True) # --- gRPC tool: negative paths (edge validation) --- ack = stub.SetVariables(pb.VariableUpdate( values={"definitely_not_a_var": pb.Value(real=1.0)})) check("gRPC SetVariables(unknown) -> PARAMETER_INVALID", ack.code == pb.Ack.PARAMETER_INVALID, pb.Ack.Code.Name(ack.code)) ack = stub.FireEvent(pb.Event(name="NoSuchEvent")) check("gRPC FireEvent(unknown) -> PARAMETER_INVALID", ack.code == pb.Ack.PARAMETER_INVALID, pb.Ack.Code.Name(ack.code)) # --- gRPC tool: set a value, then fire the event that reports it --- ack = stub.SetVariables(pb.VariableUpdate( values={"ChamberPressure": pb.Value(real=2.5)})) check("gRPC SetVariables(ChamberPressure=2.5) -> ACCEPT", ack.code == pb.Ack.ACCEPT, pb.Ack.Code.Name(ack.code)) # Read back through the API: SetVariables -> engine -> GetVariables. snap = stub.GetVariables(pb.VariableQuery(names=["ChamberPressure"])) got = snap.values["ChamberPressure"].real check("gRPC GetVariables round-trips ChamberPressure=2.5", abs(got - 2.5) < 0.01, f"got {got}") ack = stub.FireEvent(pb.Event(name="ProcessStarted")) check("gRPC FireEvent(ProcessStarted) -> ACCEPT", ack.code == pb.Ack.ACCEPT, pb.Ack.Code.Name(ack.code)) # The host should now receive S6F11 CEID 300 carrying the 2.5. got = ceid300.wait(timeout=10) check("host received S6F11 CEID 300 (gRPC FireEvent bridged to HSMS)", got) if got: vals = list(_scalars(last_s6f11.get("body"))) near = any(isinstance(v, (int, float)) and abs(float(v) - 2.5) < 0.01 for v in vals) check("S6F11 report carries ChamberPressure=2.5 (value flowed end-to-end)", near, f"scalars={vals}") # --- alarms: host enables ALID 1 (S5F3), gRPC raises it BY NAME, # host receives the unsolicited S5F1 with set-bit + ALID 1 --- client.send_and_waitfor_response( F.SecsS05F03({"ALED": 0x80, "ALID": 1})) ack = stub.SetAlarm(pb.Alarm(name="chiller_temp_high")) check("gRPC SetAlarm(chiller_temp_high) -> ACCEPT", ack.code == pb.Ack.ACCEPT, pb.Ack.Code.Name(ack.code)) got_alarm = s5f1_seen.wait(timeout=10) check("host received S5F1 (gRPC SetAlarm bridged to HSMS)", got_alarm) if got_alarm: check("S5F1 carries ALID 1 with the set bit", last_s5f1.get("ALID") == 1 and (int(last_s5f1.get("ALCD") or 0) & 0x80) != 0, f"body={last_s5f1}") stub.ClearAlarm(pb.Alarm(name="chiller_temp_high")) # --- the full conformant command loop (HCACK-4 contract) --- # Tool subscribes; host sends S2F41 START; daemon answers HCACK=4 and # forwards the command to the tool; tool completes it and fires the # event; host receives S6F11 — command -> tool -> outcome, end to end. received_cmd = {} cmd_seen = threading.Event() def consume_stream(): try: for hr in stub.Subscribe(pb.SubscribeRequest(client="interop-tool")): if hr.HasField("command"): received_cmd["cmd"] = hr.command cmd_seen.set() return except grpc.RpcError: pass # stream cancelled at teardown tool_thread = threading.Thread(target=consume_stream, daemon=True) tool_thread.start() time.sleep(0.5) # let the subscription register ceid300.clear() rsp = client.send_and_waitfor_response( F.SecsS02F41({"RCMD": "START", "PARAMS": []})) body = client.settings.streams_functions.decode(rsp).get() hcack = body.get("HCACK") if isinstance(body, dict) else None check("host got S2F42 HCACK=4 (accepted, will finish later)", int(hcack or -1) == 4, f"HCACK={hcack!r}") got_cmd = cmd_seen.wait(timeout=10) check("tool received START on the Subscribe stream", got_cmd) if got_cmd: cmd = received_cmd["cmd"] check("streamed command carries name + id", cmd.name == "START" and bool(cmd.id), f"name={cmd.name} id={cmd.id}") ack = stub.CompleteCommand(pb.CommandResult( id=cmd.id, ack=pb.Ack(code=pb.Ack.ACCEPT))) check("CompleteCommand(id) -> ACCEPT", ack.code == pb.Ack.ACCEPT, pb.Ack.Code.Name(ack.code)) # The tool reports the real outcome as an event; the host sees it. stub.FireEvent(pb.Event(name="ProcessStarted")) check("host received S6F11 after tool completed the command", ceid300.wait(timeout=10)) # --- operator-panel control state: take the tool offline via gRPC --- ack = stub.RequestControlState(pb.ControlStateRequest( desired=pb.ControlState.HOST_OFFLINE)) check("gRPC RequestControlState(HOST_OFFLINE) -> ACCEPT", ack.code == pb.Ack.ACCEPT, pb.Ack.Code.Name(ack.code)) cs = stub.GetControlState(pb.Empty()) check("control state is HOST_OFFLINE after operator request", cs.state == pb.ControlState.HOST_OFFLINE, pb.ControlState.State.Name(cs.state)) finally: client.disable() if failures: LOG.error("FAILURES (%d):", len(failures)) for f in failures: LOG.error(" - %s", f) return 1 LOG.info("all daemon interop checks passed") return 0 def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--grpc", default="gemd:50051") ap.add_argument("--hsms-host", default="gemd") ap.add_argument("--hsms-port", type=int, default=5000) ap.add_argument("--session-id", type=int, default=0) ap.add_argument("--proto-dir", default="/app/proto/secsgem/v1") args = ap.parse_args() logging.basicConfig(level=logging.INFO, format="%(message)s") return run(args.grpc, args.hsms_host, args.hsms_port, args.session_id, args.proto_dir) if __name__ == "__main__": raise SystemExit(main())