raphael/secs-gem
Public Access
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

interop: secsgem-py cross-validation harness + lenient identifier parsing

Browse Source 
Adds a Docker-based interop harness that drives the C++ server with
secsgem-py 0.3.0 as the active host and probes a secsgem-py-passive
equipment from a minimal C++ active client.  Surfaces and fixes four
interoperability bugs uncovered by cross-testing:

  * SEMI E5 identifier formatcodes are a U1|U2|U4|U8 wildcard;
    secsgem-py picks the narrowest fitting width while our parsers
    only accepted U4.  `as_uN_scalar` / `as_iN_scalar` now accept
    any unsigned/signed width and range-check the downcast.
  * PPBODY (S7F3/F6) is "ASCII | Binary | List" per the spec;
    secsgem-py defaults to ASCII.  Added BINARY_OR_ASCII codegen
    item type with `as_text_or_binary` accessor.
  * S1F23/F24 Collection Event Namelist was unimplemented; added
    schema + `vids_for(ceid)` accessor on EventReportSubscriptions
    plus the dispatch handler.
  * S10F1 was registered as a host->equipment handler, but per
    SEMI E5 §12 S10F1 is equipment->host; S10F3 is the actual
    host->equipment Terminal Display Single.  Added an S10F3
    handler alongside (we keep S10F1 too for backward compat).

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Raphael Maenle
2026-06-08 23:17:18 +02:00
parent 9fbab92106
commit 2d60571a9c
13 changed files with 793 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
interop/.gitignore
+1
View File
@@ -0,0 +1 @@
__pycache__/
interop/Dockerfile
+7
View File
@@ -0,0 +1,7 @@
# Python image used to run secsgem-py against our C++ apps for cross-validation.
# Kept separate from the C++ build image so we don't bloat the toolchain layer.
FROM python:3.12-slim
RUN pip install --no-cache-dir secsgem==0.3.0
WORKDIR /interop
interop/README.md
+59
View File
@@ -0,0 +1,59 @@
# secsgem-py interop harness
Cross-validates our C++ SECS-II / HSMS / GEM implementation against
[secsgem-py](https://pypi.org/project/secsgem/) 0.3.0, the de-facto
Python reference. Everything runs in Docker — no Python or secsgem-py
on the host.
## What it tests
| Driver | Peer | Coverage |
| ------------------------------ | --------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------- |
| `host_vs_cpp_server.py` | C++ `secs_server` (passive) | HSMS select/separate, S1F1/F3/F11/F17/F23, S2F13/F17/F29/F33/F35/F37/F41, S5F3/F5/F7, S5F1 unsolicited, S6F11 unsolicited, S7F3/F5/F19, S10F1/F3, S1F15 |
| `secs_interop_probe` (C++) | `passive_equipment.py` (secsgem-py GemEquipmentHandler) | HSMS select, S1F13/F14, S1F1/F2, S1F3/F4, clean separate |
| `raw_gem300_harness.py` | C++ `secs_server` (passive) | GEM 300 streams secsgem-py upstream doesn't ship: S3F17/F18 (E87 carrier action), S16F5/F6 (E40 PRJobCommand), S16F27/F28 (E94 CJobCommand) — built with custom `SecsStreamFunction` subclasses + registered custom `DataItem`s |
24 named checks on the C++-server side; 4 explicit checks on the
C++-host side; 4 GEM-300 raw-frame checks. Implicit HSMS state-machine
and wire-level framing validation everywhere.
## Running
```bash
# Start C++ passive server, then drive it with secsgem-py host:
docker compose up -d server
docker compose run --rm interop python3 /app/interop/host_vs_cpp_server.py \
--host server --port 5000 --session-id 0
# Start Python passive equipment, then probe it with the C++ host:
docker compose up -d equipment_py
docker compose run --rm builder /app/build/secs_interop_probe \
--host equipment_py --port 5000 --device 0
```
Both exit 0 on success.
## What this caught
Real bugs surfaced by interop (now fixed):
1. **Strict U4 parsing rejected U1-encoded identifiers.** SEMI E5
declares DATAID, RPTID, VID, CEID, ALID, EXID, etc. as
`U1 | U2 | U4 | U8`; secsgem-py picks the smallest width that fits.
Our `as_u4_scalar`, `as_u2_scalar`, etc. were strict. Now lenient
with range-checked downcasts (`messages_helpers.hpp::any_unsigned_first`).
2. **PPBODY rejected when sent as ASCII.** SEMI lets PPBODY be
`ASCII | Binary | List`; secsgem-py defaults to ASCII. Added the
`BINARY_OR_ASCII` codegen item type plus a permissive
`as_text_or_binary` accessor, used for S7F3/F6.
3. **Missing S1F23 / S1F24 (Collection Event Namelist).** Added the
wire schema in `data/messages.yaml`, a `vids_for(ceid)` accessor on
the event-report store, and the dispatch handler in `secs_server.cpp`.
4. **Missing S10F3 handler (Terminal Display Single, host→equipment).**
Our server only registered S10F1; per SEMI E5, S10F1 is
equipment→host and S10F3 is the host→equipment counterpart. Added
the missing dispatch.
The C++ test suite still passes (278 cases / 1436 assertions) after
each of these changes — the fixes are purely permissive widenings, no
existing behaviour was broken.
interop/host_vs_cpp_server.py
+282
View File
@@ -0,0 +1,282 @@
"""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
LOG = logging.getLogger("interop")
F = secsgem.secs.functions
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)
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")
# ---- S7 recipes ----
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}")
# ---- 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())
interop/passive_equipment.py
+94
View File
@@ -0,0 +1,94 @@
"""Run secsgem-py as passive equipment so a C++ active host can drive it.
Listens on 0.0.0.0:<port>, waits to be reached by an active host.
secsgem-py's GemEquipmentHandler answers S1F13/F14 (establish
communications), S1F17/F18 (online request), S1F1/F2 (are-you-there)
out of the box. Status variables, equipment constants, and one
collection event are registered so that S1F3/S1F11/S1F23 round-trip
yield non-empty bodies.
"""
from __future__ import annotations
import argparse
import logging
import signal
import sys
import threading
import secsgem.common
import secsgem.gem
import secsgem.hsms
import secsgem.secs.variables as v
LOG = logging.getLogger("interop.equipment")
class InteropEquipment(secsgem.gem.GemEquipmentHandler):
def __init__(self, settings):
super().__init__(settings)
# One status variable so S1F3/F11 yield a non-empty list.
sv = secsgem.gem.StatusVariable(1, "ChamberTemp", "C", v.U4)
sv.value = 25
self.status_variables[1] = sv
# One equipment constant.
self.equipment_constants[10] = secsgem.gem.EquipmentConstant(
10, "EnergyCap", 0, 100, 50, "kW", v.U4
)
# One collection event (visible via S1F23).
self.collection_events[1001] = secsgem.gem.CollectionEvent(
1001, "DemoEvent", []
)
def main() -> int:
ap = argparse.ArgumentParser()
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",
)
if args.log_level.upper() != "DEBUG":
logging.getLogger("communication").setLevel(logging.WARNING)
logging.getLogger("hsms_connection").setLevel(logging.WARNING)
settings = secsgem.hsms.HsmsSettings(
address="0.0.0.0",
port=args.port,
session_id=args.session_id,
connect_mode=secsgem.hsms.HsmsConnectMode.PASSIVE,
device_type=secsgem.common.DeviceType.EQUIPMENT,
)
eq = InteropEquipment(settings)
eq.enable()
LOG.info("passive equipment listening on 0.0.0.0:%d session=%d",
args.port, args.session_id)
stop = threading.Event()
def shutdown(_signo, _frame):
LOG.info("signal received, stopping")
stop.set()
signal.signal(signal.SIGINT, shutdown)
signal.signal(signal.SIGTERM, shutdown)
try:
stop.wait()
finally:
try:
eq.disable()
except Exception as e: # noqa: BLE001
LOG.warning("disable raised: %s", e)
return 0
if __name__ == "__main__":
sys.exit(main())