"""End-to-end validation of the secsgem_client Python package. The PUBLISHED client API (not raw stubs) plays the tool against a live secs_gemd, while secsgem-py — the reference GEM implementation — plays the fab host over HSMS. Every beautiful-API call is asserted against what the host actually receives on the wire: eq.set / eq["..."] -> S1F3-visible values, GetVariables round-trip eq.fire -> host receives S6F11 with the configured report eq.alarm / eq.clear -> host receives S5F1 set/clear @eq.command + eq.listen -> host's S2F41 gets HCACK=4, the name-bound handler runs, completion event reaches the host eq.names.* -> Describe-backed, typo-safe name lookup eq.report_substrate -> E90 wafer journey (+ unknown-wafer error path) eq.report_module -> E157 module walk/reset (+ illegal-jump error path) eq.control_state / request_control_state / eq.health Exits 0 on success. """ from __future__ import annotations import argparse import logging 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, Milestone, ModuleState, SecsGemError LOG = logging.getLogger("pyclient-interop") F = secsgem.secs.functions def run(grpc_addr: str, hsms_host: str, hsms_port: int) -> int: 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(label) # ---- the tool, via the published client API ---- eq = Equipment(grpc_addr) started = threading.Event() # @eq.command binds by function name, validated against the live equipment # (Describe) at decoration time — this is the path the host's S2F41 drives. @eq.command def START(cmd): # noqa: ANN001 started.set() eq.fire(eq.names.event.ProcessStarted) # typo-safe completion signal eq.listen(background=True) # ---- the host, via secsgem-py ---- settings = secsgem.hsms.HsmsSettings( address=hsms_host, port=hsms_port, session_id=0, connect_mode=secsgem.hsms.HsmsConnectMode.ACTIVE, device_type=secsgem.common.DeviceType.HOST) host = secsgem.gem.GemHostHandler(settings) ceid300 = threading.Event() s5f1 = {} s5f1_seen = threading.Event() def on_s6f11(_h, message): body = host.settings.streams_functions.decode(message).get() if isinstance(body, dict) and body.get("CEID") == 300: ceid300.set() host.send_response(F.SecsS06F12(0), message.header.system) def on_s5f1(_h, message): body = host.settings.streams_functions.decode(message).get() if isinstance(body, dict): s5f1.update(body) s5f1_seen.set() host.send_response(F.SecsS05F02(0), message.header.system) host.register_stream_function(6, 11, on_s6f11) host.register_stream_function(5, 1, on_s5f1) host.enable() try: if not host.waitfor_communicating(timeout=15): check("HSMS establish-communications", False) return 1 check("HSMS establish-communications", True) host.send_and_waitfor_response(F.SecsS01F17()) time.sleep(0.3) # ---- variables: kwargs, item syntax, read-back, errors ---- eq.set(ChamberPressure=2.5, WaferCounter=7) check("eq.set(kwargs) + eq.get round-trip", eq.get("ChamberPressure", "WaferCounter") == {"ChamberPressure": 2.5, "WaferCounter": 7}) eq["ChamberPressure"] = 1.25 check('eq["..."] item syntax', eq["ChamberPressure"] == 1.25) try: eq.set(NoSuchVariable=1) check("unknown variable raises SecsGemError", False) except SecsGemError as e: check("unknown variable raises SecsGemError", "NoSuchVariable" in str(e)) # ---- control state / health ---- check("eq.control_state", eq.control_state == "ONLINE_REMOTE", eq.control_state) h = eq.health() check("eq.health()", h.link == "SELECTED" and h.control_state == "ONLINE_REMOTE", str(h)) # ---- events: configure a report host-side, fire client-side ---- host.send_and_waitfor_response( F.SecsS02F33({"DATAID": 1, "DATA": [{"RPTID": 1, "VID": [101]}]})) host.send_and_waitfor_response( F.SecsS02F35({"DATAID": 1, "DATA": [{"CEID": 300, "RPTID": [1]}]})) host.send_and_waitfor_response(F.SecsS02F37({"CEED": True, "CEID": [300]})) eq.fire("ProcessStarted", ChamberPressure=2.75) check("eq.fire -> host receives S6F11", ceid300.wait(timeout=10)) # ---- alarms ---- host.send_and_waitfor_response(F.SecsS05F03({"ALED": 0x80, "ALID": 1})) eq.alarm("chiller_temp_high") got = s5f1_seen.wait(timeout=10) check("eq.alarm -> host receives S5F1 (set)", got and s5f1.get("ALID") == 1 and (int(s5f1.get("ALCD") or 0) & 0x80)) s5f1_seen.clear() eq.clear("chiller_temp_high") got = s5f1_seen.wait(timeout=10) check("eq.clear -> host receives S5F1 (clear)", got and not (int(s5f1.get("ALCD") or 0) & 0x80)) # ---- the command loop through @eq.on + eq.listen ---- ceid300.clear() rsp = host.send_and_waitfor_response( F.SecsS02F41({"RCMD": "START", "PARAMS": []})) body = host.settings.streams_functions.decode(rsp).get() check("host S2F41 -> HCACK=4", isinstance(body, dict) and int(body.get("HCACK") or -1) == 4) check("@eq.command('START') handler ran", started.wait(timeout=10)) check("handler's eq.fire reached the host (completion signal)", ceid300.wait(timeout=10)) # ---- Describe-backed names (every category, dir(), typo hint) ---- check("eq.names.event has ProcessStarted", "ProcessStarted" in eq.names.event) check("eq.names.command has START", "START" in eq.names.command) check("eq.names.var has ChamberPressure", "ChamberPressure" in eq.names.var) check("eq.names.alarm has chiller_temp_high", "chiller_temp_high" in eq.names.alarm) check("dir(eq.names.event) lists names for autocomplete", "ProcessStarted" in dir(eq.names.event)) try: _ = eq.names.event.ProcessStated # one-letter typo check("typo on eq.names raises with suggestion", False) except AttributeError as e: check("typo on eq.names raises with suggestion", "ProcessStarted" in str(e)) # close-match hint present # ---- @eq.command rejects an unknown command name at decoration time ---- try: @eq.command def NOT_A_REAL_COMMAND(cmd): # noqa: ANN001 pass check("@eq.command on unknown name raises NameError", False) except NameError: check("@eq.command on unknown name raises NameError", True) # ---- E90 substrate tracking: enum + string forms, then error paths ---- eq.report_substrate("WFR-PY-1", Milestone.ARRIVED, carrier_id="FOUP-PY", slot=4) # enum form eq.report_substrate("WFR-PY-1", "AT_WORK") # plain string, same thing eq.report_substrate("WFR-PY-1", Milestone.PROCESSING) eq.report_substrate("WFR-PY-1", Milestone.PROCESSED) eq.report_substrate("WFR-PY-1", Milestone.AT_DESTINATION) check("report_substrate full journey (enum + string) accepted", True) try: eq.report_substrate("WFR-PY-1", "AT_WROK") # typo: client-side reject check("misspelled milestone raises ValueError", False) except ValueError as e: check("misspelled milestone raises ValueError", "AT_WORK" in str(e)) try: eq.report_substrate("WFR-GHOST", Milestone.AT_WORK) # never ARRIVED check("report_substrate on unknown wafer raises", False) except SecsGemError as e: check("report_substrate on unknown wafer raises", "WFR-GHOST" in str(e)) # ---- E157 module tracking: walk and reset ---- eq.report_module("MOD-PY-1", ModuleState.GENERAL_EXECUTING) eq.report_module("MOD-PY-1", ModuleState.STEP_EXECUTING) eq.report_module("MOD-PY-1", ModuleState.STEP_COMPLETED) eq.report_module("MOD-PY-1", ModuleState.NOT_EXECUTING) check("report_module walk + reset accepted (no raise)", True) try: eq.report_module("MOD-PY-2", ModuleState.STEP_EXECUTING) # illegal from idle check("report_module illegal jump raises", False) except SecsGemError: check("report_module illegal jump raises", True) # ---- operator offline via the client ---- eq.request_control_state("HOST_OFFLINE") check("request_control_state(HOST_OFFLINE)", eq.control_state == "HOST_OFFLINE", eq.control_state) finally: host.disable() eq.close() if failures: LOG.error("FAILURES (%d): %s", len(failures), failures) return 1 LOG.info("all secsgem_client 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) args = ap.parse_args() logging.basicConfig(level=logging.INFO, format="%(message)s") logging.getLogger("communication").setLevel(logging.WARNING) logging.getLogger("hsms_connection").setLevel(logging.WARNING) return run(args.grpc, args.hsms_host, args.hsms_port) if __name__ == "__main__": raise SystemExit(main())