secs-gem/interop/daemon_interop.py

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