raphael
7c28e2589c
Adds round-trip checks for the SECS-II messages added in the AA
catalog-growth commit but never cross-validated against secsgem-py:
* S2F21/F22 — legacy remote command (no params). secsgem-py's
stock S2F21 sends with W=0; we register a W=1 override so the
transaction awaits our S2F22 reply. Also widens CMDA's allowed
types to include Binary (secsgem-py 0.3.0 declares CMDA as
Dynamic[U1, I1] only; SEMI E5 §10.18 says Binary, and our server
emits it that way).
* S6F15/F16 — event-report request by CEID.
* S6F19/F20 — individual report request by RPTID.
* S6F21/F22 — annotated individual report request.
* S7F1/F2 — PP load inquire.
* S7F17/F18 — PP delete.
Suite is now 32 named host-vs-server checks — all green in three
consecutive runs.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
361 lines
14 KiB
Python
361 lines
14 KiB
Python
"""Drive the C++ passive secs_server with secsgem-py as the active host.
|
|
|
|
Exercises a broad slice of streams that both implementations support:
|
|
HSMS select/linktest/separate, S1/S2/S5/S6/S7/S10 message exchanges,
|
|
event reports, alarms, recipes, host commands, variable limits, and the
|
|
spool path. Exits 0 on success, non-zero on any failure.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import logging
|
|
import sys
|
|
import threading
|
|
import time
|
|
|
|
import secsgem.common
|
|
import secsgem.gem
|
|
import secsgem.hsms
|
|
import secsgem.secs
|
|
from secsgem.secs.functions.base import SecsStreamFunction
|
|
from secsgem.secs.data_items.base import DataItemBase
|
|
import secsgem.secs.data_items as _di
|
|
import secsgem.secs.variables as _v
|
|
|
|
|
|
LOG = logging.getLogger("interop")
|
|
F = secsgem.secs.functions
|
|
|
|
|
|
# secsgem-py 0.3.0 declares CMDA as Dynamic[U1, I1] only, but SEMI E5
|
|
# spells it as Binary in §10.18 and our C++ server emits it that way.
|
|
# Widen the allowed type set so S2F22 decode succeeds.
|
|
class _CMDA(DataItemBase):
|
|
__type__ = _v.Dynamic
|
|
__allowedtypes__ = [_v.Binary, _v.U1, _v.I1]
|
|
_di.CMDA = _CMDA
|
|
|
|
|
|
# secsgem-py's stock S2F21 ships with `_is_reply_required = False` and
|
|
# the wire goes out with W=0; our server replies anyway but secsgem-py
|
|
# never correlates the response so `send_and_waitfor_response` times
|
|
# out. Override with the spec-correct W=1.
|
|
class S02F21W(SecsStreamFunction):
|
|
_stream = 2
|
|
_function = 21
|
|
_to_host = False
|
|
_to_equipment = True
|
|
_has_reply = True
|
|
_is_reply_required = True
|
|
_is_multi_block = False
|
|
_data_format = "< RCMD >"
|
|
|
|
|
|
def run(host: str, port: int, session_id: int) -> int:
|
|
settings = secsgem.hsms.HsmsSettings(
|
|
address=host,
|
|
port=port,
|
|
session_id=session_id,
|
|
connect_mode=secsgem.hsms.HsmsConnectMode.ACTIVE,
|
|
device_type=secsgem.common.DeviceType.HOST,
|
|
)
|
|
|
|
client = secsgem.gem.GemHostHandler(settings)
|
|
|
|
failures: list[str] = []
|
|
events_received: list[int] = []
|
|
alarms_received: list[tuple[int, int]] = []
|
|
|
|
# Per-CEID arrival events so we can wait for a specific event ID, not just
|
|
# "any S6F11." The C++ server emits CEID 100 (control-state change) twice
|
|
# during S1F17 (HostOffline -> AttemptOnline -> OnlineRemote), so a generic
|
|
# "first message wins" gate races against the CEID 300 from RCMD=START.
|
|
events_lock = threading.Lock()
|
|
events_seen: dict[int, threading.Event] = {}
|
|
|
|
def event_arrival(ceid: int) -> threading.Event:
|
|
with events_lock:
|
|
ev = events_seen.get(ceid)
|
|
if ev is None:
|
|
ev = threading.Event()
|
|
events_seen[ceid] = ev
|
|
return ev
|
|
|
|
s5f1_event = threading.Event()
|
|
|
|
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
|
|
if ceid is not None:
|
|
ceid_i = int(ceid)
|
|
events_received.append(ceid_i)
|
|
LOG.info("[evt] received S6F11 CEID=%s", ceid)
|
|
event_arrival(ceid_i).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()
|
|
alcd = body.get("ALCD") if isinstance(body, dict) else None
|
|
alid = body.get("ALID") if isinstance(body, dict) else None
|
|
if alid is not None:
|
|
alarms_received.append((int(alcd or 0), int(alid)))
|
|
LOG.info("[alm] received S5F1 ALID=%s ALCD=0x%02x", alid, alcd or 0)
|
|
s5f1_event.set()
|
|
client.send_response(F.SecsS05F02(0), message.header.system)
|
|
|
|
# Register the primary-message handlers up-front.
|
|
client.register_stream_function(6, 11, on_s6f11)
|
|
client.register_stream_function(5, 1, on_s5f1)
|
|
# Override secsgem-py's W=0 S2F21 so its host transactions wait for
|
|
# the server's S2F22 reply.
|
|
client.settings.streams_functions.update(S02F21W)
|
|
|
|
def check(label: str, ok: bool, detail: str = "") -> None:
|
|
status = "OK " if ok else "FAIL"
|
|
LOG.info("[%s] %s%s", status, label, f" — {detail}" if detail else "")
|
|
if not ok:
|
|
failures.append(f"{label}: {detail or 'failed'}")
|
|
|
|
def round_trip(label: str, msg, expect=lambda body: True, detail_fn=str):
|
|
rsp = client.send_and_waitfor_response(msg)
|
|
decoded = client.settings.streams_functions.decode(rsp)
|
|
body = decoded.get()
|
|
try:
|
|
ok = expect(body)
|
|
except Exception as e: # noqa: BLE001
|
|
ok = False
|
|
body = f"{body!r} (predicate raised {e!r})"
|
|
check(label, ok, detail_fn(body))
|
|
return body
|
|
|
|
LOG.info("enabling host handler (active connect to %s:%d, session=%d)",
|
|
host, port, session_id)
|
|
client.enable()
|
|
try:
|
|
ok = client.waitfor_communicating(timeout=15)
|
|
check("HSMS select + S1F13/F14 establish-communications", ok)
|
|
if not ok:
|
|
return 1
|
|
|
|
# ---- S1 ----
|
|
round_trip("S1F1 → S1F2 (Are You There)", F.SecsS01F01(),
|
|
lambda b: isinstance(b, list) and len(b) == 2)
|
|
round_trip("S1F3 → S1F4 (all SVIDs)", F.SecsS01F03([]),
|
|
lambda b: isinstance(b, list) and len(b) >= 1,
|
|
lambda b: f"{len(b)} values")
|
|
round_trip("S1F11 → S1F12 (status namelist)", F.SecsS01F11([]),
|
|
lambda b: isinstance(b, list) and len(b) >= 1,
|
|
lambda b: f"{len(b)} rows")
|
|
round_trip("S1F17 → S1F18 (online req)", F.SecsS01F17(),
|
|
lambda b: b in (0, 1, 2),
|
|
lambda b: f"ONLACK={b!r}")
|
|
round_trip("S1F23 → S1F24 (collection event namelist)", F.SecsS01F23([]),
|
|
lambda b: isinstance(b, list),
|
|
lambda b: f"{len(b) if isinstance(b, list) else 0} CEIDs")
|
|
|
|
# ---- S2 (constants, time, events) ----
|
|
round_trip("S2F13 → S2F14 (EC values, all)", F.SecsS02F13([]),
|
|
lambda b: isinstance(b, list),
|
|
lambda b: f"{len(b)} values")
|
|
round_trip("S2F17 → S2F18 (clock)", F.SecsS02F17(),
|
|
lambda b: isinstance(b, str) and len(b) in (12, 16),
|
|
lambda b: f"TIME={b!r}")
|
|
round_trip("S2F29 → S2F30 (EC namelist)", F.SecsS02F29([]),
|
|
lambda b: isinstance(b, list) and len(b) >= 1,
|
|
lambda b: f"{len(b)} ECs")
|
|
|
|
# Define + link + enable reports. Equipment.yaml maps RCMD=START
|
|
# to CEID=300, so we link RPTID 1 to CEIDs 100 and 300 and enable
|
|
# both — that way we exercise both the CEID-already-defined and
|
|
# the freshly-defined linkage paths.
|
|
round_trip(
|
|
"S2F33 → S2F34 (define RPTID 1 → SVID 2)",
|
|
F.SecsS02F33({"DATAID": 1, "DATA": [{"RPTID": 1, "VID": [2]}]}),
|
|
lambda b: b == 0,
|
|
lambda b: f"DRACK={b!r}",
|
|
)
|
|
round_trip(
|
|
"S2F35 → S2F36 (link CEIDs 100,300 → RPTID 1)",
|
|
F.SecsS02F35({"DATAID": 1, "DATA": [
|
|
{"CEID": 100, "RPTID": [1]},
|
|
{"CEID": 300, "RPTID": [1]},
|
|
]}),
|
|
lambda b: b == 0,
|
|
lambda b: f"LRACK={b!r}",
|
|
)
|
|
round_trip(
|
|
"S2F37 → S2F38 (enable CEIDs 100,300)",
|
|
F.SecsS02F37({"CEED": True, "CEID": [100, 300]}),
|
|
lambda b: b == 0,
|
|
lambda b: f"ERACK={b!r}",
|
|
)
|
|
|
|
# Per equipment.yaml: RCMD=START → emit_ceid=300.
|
|
round_trip(
|
|
"S2F41 → S2F42 (RCMD=START)",
|
|
F.SecsS02F41({"RCMD": "START", "PARAMS": []}),
|
|
lambda b: isinstance(b, dict) and b.get("HCACK") == 0,
|
|
lambda b: f"HCACK={b.get('HCACK') if isinstance(b, dict) else b!r}",
|
|
)
|
|
|
|
# Wait specifically for CEID 300 — we may have seen CEID 100 from
|
|
# earlier control-state transitions, so a generic "any S6F11" gate
|
|
# races against the START-triggered event.
|
|
if event_arrival(300).wait(timeout=2.0):
|
|
check(
|
|
"S6F11 received after RCMD=START",
|
|
True,
|
|
f"observed CEIDs={events_received}",
|
|
)
|
|
else:
|
|
check("S6F11 received after RCMD=START", False,
|
|
f"timeout (observed={events_received})")
|
|
|
|
# ---- S5 alarms ----
|
|
round_trip("S5F5 → S5F6 (list all alarms)", F.SecsS05F05([]),
|
|
lambda b: isinstance(b, list) and len(b) >= 1,
|
|
lambda b: f"{len(b)} alarms")
|
|
round_trip("S5F7 → S5F8 (list enabled alarms)", F.SecsS05F07(),
|
|
lambda b: isinstance(b, list),
|
|
lambda b: f"{len(b)} enabled")
|
|
# Enable alarm 1: ALED bit 7 set (0x80) means enable.
|
|
round_trip("S5F3 → S5F4 (enable ALID=1)",
|
|
F.SecsS05F03({"ALED": 0x80, "ALID": 1}),
|
|
lambda b: b == 0,
|
|
lambda b: f"ACKC5={b!r}")
|
|
# Fire FAULT host-command -> S5F1 alarm-set.
|
|
round_trip("S2F41 → S2F42 (RCMD=FAULT)",
|
|
F.SecsS02F41({"RCMD": "FAULT", "PARAMS": []}),
|
|
lambda b: isinstance(b, dict) and b.get("HCACK") == 0,
|
|
lambda b: f"HCACK={b.get('HCACK') if isinstance(b, dict) else b!r}")
|
|
if s5f1_event.wait(timeout=2.0):
|
|
check(
|
|
"S5F1 received after RCMD=FAULT",
|
|
len(alarms_received) >= 1,
|
|
f"alarms={alarms_received}",
|
|
)
|
|
else:
|
|
check("S5F1 received after RCMD=FAULT", False, "timeout")
|
|
|
|
# ---- S2F21 legacy remote command (no params) ----
|
|
# Use the W=1 override (secsgem-py's stock S2F21 has W=0 and
|
|
# never awaits our reply); routes through the same
|
|
# HostCommandRegistry as S2F41.
|
|
round_trip("S2F21 → S2F22 (legacy RCMD=START)",
|
|
S02F21W("START"),
|
|
lambda b: b == 0,
|
|
lambda b: f"CMDA={b!r}")
|
|
|
|
# ---- S6 host-initiated report queries -------------------------
|
|
# S6F15 → S6F16: pull current event-report payload by CEID.
|
|
# CEID 300 was linked to RPTID 1 (SVID 2 = Clock) above.
|
|
round_trip(
|
|
"S6F15 → S6F16 (event report request CEID=300)",
|
|
F.SecsS06F15(300),
|
|
lambda b: isinstance(b, dict) and b.get("CEID") == 300,
|
|
lambda b: f"CEID={b.get('CEID') if isinstance(b, dict) else b!r}",
|
|
)
|
|
# S6F19 → S6F20: pull RPTID 1 directly (just values, no annotation).
|
|
round_trip(
|
|
"S6F19 → S6F20 (individual report request RPTID=1)",
|
|
F.SecsS06F19(1),
|
|
lambda b: isinstance(b, list) and len(b) >= 1,
|
|
lambda b: f"{len(b) if isinstance(b, list) else 0} values",
|
|
)
|
|
# S6F21 → S6F22: same but with (VID, value) annotation.
|
|
round_trip(
|
|
"S6F21 → S6F22 (annotated individual report RPTID=1)",
|
|
F.SecsS06F21(1),
|
|
lambda b: isinstance(b, list) and len(b) >= 1,
|
|
lambda b: f"{len(b) if isinstance(b, list) else 0} annotated values",
|
|
)
|
|
|
|
# ---- S7 recipes ----
|
|
# S7F1/F2: ask permission before sending a PP of LENGTH bytes.
|
|
round_trip(
|
|
"S7F1 → S7F2 (PP load inquire)",
|
|
F.SecsS07F01({"PPID": "RECIPE-Y", "LENGTH": 64}),
|
|
lambda b: b in (0, 1, 2, 3, 4, 5, 6),
|
|
lambda b: f"PPGNT={b!r}",
|
|
)
|
|
round_trip("S7F19 → S7F20 (current PPID list)", F.SecsS07F19(),
|
|
lambda b: isinstance(b, list),
|
|
lambda b: f"{len(b)} PPIDs")
|
|
round_trip(
|
|
"S7F3 → S7F4 (send PP RECIPE-X)",
|
|
F.SecsS07F03({"PPID": "RECIPE-X", "PPBODY": b"hello"}),
|
|
lambda b: b == 0,
|
|
lambda b: f"ACKC7={b!r}",
|
|
)
|
|
round_trip("S7F5 → S7F6 (read back RECIPE-X)", F.SecsS07F05("RECIPE-X"),
|
|
lambda b: isinstance(b, dict) and b.get("PPID") == "RECIPE-X",
|
|
lambda b: f"got {b!r}")
|
|
# S7F17/F18: delete the PP we just installed.
|
|
round_trip(
|
|
"S7F17 → S7F18 (delete RECIPE-X)",
|
|
F.SecsS07F17(["RECIPE-X"]),
|
|
lambda b: b == 0,
|
|
lambda b: f"ACKC7={b!r}",
|
|
)
|
|
|
|
# ---- S10 terminal display ----
|
|
round_trip(
|
|
"S10F1 → S10F2 (terminal one-liner)",
|
|
F.SecsS10F01({"TID": 0, "TEXT": "hello equipment"}),
|
|
lambda b: b == 0,
|
|
lambda b: f"ACKC10={b!r}",
|
|
)
|
|
round_trip(
|
|
"S10F3 → S10F4 (terminal multi-line)",
|
|
F.SecsS10F03({"TID": 0, "TEXT": "line A\nline B"}),
|
|
lambda b: b == 0,
|
|
lambda b: f"ACKC10={b!r}",
|
|
)
|
|
|
|
# ---- S1 control: go offline cleanly ----
|
|
round_trip("S1F15 → S1F16 (offline)", F.SecsS01F15(),
|
|
lambda b: b in (0, 1, 2),
|
|
lambda b: f"OFLACK={b!r}")
|
|
|
|
finally:
|
|
LOG.info("disabling host handler (separate)")
|
|
try:
|
|
client.disable()
|
|
except Exception as e: # noqa: BLE001
|
|
LOG.warning("disable raised: %s", e)
|
|
time.sleep(0.2)
|
|
|
|
if failures:
|
|
LOG.error("FAILURES (%d):", len(failures))
|
|
for fail in failures:
|
|
LOG.error(" - %s", fail)
|
|
return 1
|
|
LOG.info("all checks passed")
|
|
return 0
|
|
|
|
|
|
def main() -> int:
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--host", default="server")
|
|
ap.add_argument("--port", type=int, default=5000)
|
|
ap.add_argument("--session-id", type=int, default=0)
|
|
ap.add_argument("--log-level", default="INFO")
|
|
args = ap.parse_args()
|
|
logging.basicConfig(
|
|
level=args.log_level,
|
|
format="%(asctime)s %(levelname)s %(name)s: %(message)s",
|
|
)
|
|
# Silence the noisy secsgem 'communication' INFO logger (one line per packet)
|
|
# unless the user explicitly asked for DEBUG.
|
|
if args.log_level.upper() != "DEBUG":
|
|
logging.getLogger("communication").setLevel(logging.WARNING)
|
|
logging.getLogger("hsms_connection").setLevel(logging.WARNING)
|
|
return run(args.host, args.port, args.session_id)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|