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

feat(client): the Python client — a GEM tool in plain Python (Phase C)

Browse Source 
clients/python: pip-installable "secsgem-client", pure Python (stubs
pre-generated from equipment.proto, import made package-relative; no
compiled extension, no SEMI knowledge, no C++ toolchain). The API the whole
effort aimed at:

    eq = Equipment("localhost:50051")
    eq.set(ChamberPressure=2.5); eq["WaferCounter"] = 7
    eq.fire("ProcessStarted", ChamberPressure=2.75)
    eq.alarm("chiller_temp_high"); eq.clear("chiller_temp_high")
    @eq.on("START")
    def start(cmd): ...           # auto-CompleteCommand after return
    eq.listen(background=True)
    eq.control_state; eq.request_control_state("HOST_OFFLINE"); eq.health()

Errors raise SecsGemError carrying the daemon's message ("no variable named
..."). bool checked before int in conversion (isinstance(True, int)).
examples/mini_tool.py is a complete GEM tool in ~25 lines.

PROOF — interop/pyclient_interop.py drives the PUBLISHED package (not raw
stubs) against a live secs_gemd with secsgem-py as the fab host: 13 checks
all green on first run — set/get round-trips, item syntax, SecsGemError on
unknown names, control state, health, fire->S6F11 on the host's wire,
alarm/clear->S5F1 with correct set bit, the full command loop (host S2F41 ->
HCACK=4 -> @eq.on handler -> completion event back at the host), operator
offline. Conversion layer unit-tested standalone; both wired into
tools/run_interop.sh as the pyclient step.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Raphael Maenle
2026-06-10 22:57:55 +02:00
parent 912304966f
commit 8686654b15
19 changed files with 1235 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
clients/python/secsgem_client/_proto/equipment_pb2_grpc.py
+546
View File
@@ -0,0 +1,546 @@
# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT!
"""Client and server classes corresponding to protobuf-defined services."""
import grpc
from . import equipment_pb2 as equipment__pb2
class EquipmentStub(object):
"""=============================================================================
SECS/GEM Equipment API
The daemon *is* one piece of SEMI equipment: it owns the HSMS link to the
host (MES) and speaks GEM on your behalf. Your tool software connects as a
client and only ever does two things:
• tell the equipment about itself — set variables, fire events, raise alarms
• react to what the host asks for — receive commands/jobs, answer them
Everything SECS lives inside the daemon: message framing, report definitions,
the GEM state machines, timers, spooling. You need no SEMI knowledge to use
this API. Items are addressed by the human names from your equipment config
(e.g. "chamber_pressure"), never by numeric SVID / CEID / ALID.
CAPABILITY TIERS — wire up only what your equipment is:
• Universal — variables, events, alarms, control state, commands. Every tool.
• Carriers — E87 carrier/load-port flows. Only carrier-based equipment.
• Recipes — S7 process-program transfer. Only recipe-driven equipment.
• Jobs — E40 process jobs. Only job-based process/front-end equipment.
If a tier doesn't apply, you simply never receive its HostRequest variants and
never call its report RPCs.
=============================================================================
---- Universal: report state to the host --------------------------------
"""
def __init__(self, channel):
"""Constructor.
Args:
channel: A grpc.Channel.
"""
self.SetVariables = channel.unary_unary(
'/secsgem.v1.Equipment/SetVariables',
request_serializer=equipment__pb2.VariableUpdate.SerializeToString,
response_deserializer=equipment__pb2.Ack.FromString,
)
self.GetVariables = channel.unary_unary(
'/secsgem.v1.Equipment/GetVariables',
request_serializer=equipment__pb2.VariableQuery.SerializeToString,
response_deserializer=equipment__pb2.VariableSnapshot.FromString,
)
self.FireEvent = channel.unary_unary(
'/secsgem.v1.Equipment/FireEvent',
request_serializer=equipment__pb2.Event.SerializeToString,
response_deserializer=equipment__pb2.Ack.FromString,
)
self.SetAlarm = channel.unary_unary(
'/secsgem.v1.Equipment/SetAlarm',
request_serializer=equipment__pb2.Alarm.SerializeToString,
response_deserializer=equipment__pb2.Ack.FromString,
)
self.ClearAlarm = channel.unary_unary(
'/secsgem.v1.Equipment/ClearAlarm',
request_serializer=equipment__pb2.Alarm.SerializeToString,
response_deserializer=equipment__pb2.Ack.FromString,
)
self.GetControlState = channel.unary_unary(
'/secsgem.v1.Equipment/GetControlState',
request_serializer=equipment__pb2.Empty.SerializeToString,
response_deserializer=equipment__pb2.ControlState.FromString,
)
self.RequestControlState = channel.unary_unary(
'/secsgem.v1.Equipment/RequestControlState',
request_serializer=equipment__pb2.ControlStateRequest.SerializeToString,
response_deserializer=equipment__pb2.Ack.FromString,
)
self.Subscribe = channel.unary_stream(
'/secsgem.v1.Equipment/Subscribe',
request_serializer=equipment__pb2.SubscribeRequest.SerializeToString,
response_deserializer=equipment__pb2.HostRequest.FromString,
)
self.CompleteCommand = channel.unary_unary(
'/secsgem.v1.Equipment/CompleteCommand',
request_serializer=equipment__pb2.CommandResult.SerializeToString,
response_deserializer=equipment__pb2.Ack.FromString,
)
self.ReportProcessJob = channel.unary_unary(
'/secsgem.v1.Equipment/ReportProcessJob',
request_serializer=equipment__pb2.ProcessJobState.SerializeToString,
response_deserializer=equipment__pb2.Ack.FromString,
)
self.ReportCarrier = channel.unary_unary(
'/secsgem.v1.Equipment/ReportCarrier',
request_serializer=equipment__pb2.CarrierState.SerializeToString,
response_deserializer=equipment__pb2.Ack.FromString,
)
self.WatchHealth = channel.unary_stream(
'/secsgem.v1.Equipment/WatchHealth',
request_serializer=equipment__pb2.Empty.SerializeToString,
response_deserializer=equipment__pb2.Health.FromString,
)
class EquipmentServicer(object):
"""=============================================================================
SECS/GEM Equipment API
The daemon *is* one piece of SEMI equipment: it owns the HSMS link to the
host (MES) and speaks GEM on your behalf. Your tool software connects as a
client and only ever does two things:
• tell the equipment about itself — set variables, fire events, raise alarms
• react to what the host asks for — receive commands/jobs, answer them
Everything SECS lives inside the daemon: message framing, report definitions,
the GEM state machines, timers, spooling. You need no SEMI knowledge to use
this API. Items are addressed by the human names from your equipment config
(e.g. "chamber_pressure"), never by numeric SVID / CEID / ALID.
CAPABILITY TIERS — wire up only what your equipment is:
• Universal — variables, events, alarms, control state, commands. Every tool.
• Carriers — E87 carrier/load-port flows. Only carrier-based equipment.
• Recipes — S7 process-program transfer. Only recipe-driven equipment.
• Jobs — E40 process jobs. Only job-based process/front-end equipment.
If a tier doesn't apply, you simply never receive its HostRequest variants and
never call its report RPCs.
=============================================================================
---- Universal: report state to the host --------------------------------
"""
def SetVariables(self, request, context):
"""Update one or more status/data variables by name. The daemon remembers the
values, so the host always sees the latest when it polls (S1F3).
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def GetVariables(self, request, context):
"""Read back what the daemon currently holds (useful on tool restart/reconnect).
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def FireEvent(self, request, context):
"""Fire a collection event by name. The daemon assembles the configured report
and sends S6F11. Values in `data` override current values for this one event.
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def SetAlarm(self, request, context):
"""Raise (S5F1 set) or clear an alarm by name.
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def ClearAlarm(self, request, context):
"""Missing associated documentation comment in .proto file."""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def GetControlState(self, request, context):
"""---- Universal: control state -------------------------------------------
Current GEM control state (ONLINE/LOCAL/REMOTE/OFFLINE).
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def RequestControlState(self, request, context):
"""Request a transition — e.g. an operator panel taking the tool OFFLINE for
maintenance, or back ONLINE. The daemon applies E30 rules and may decline.
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def Subscribe(self, request, context):
"""---- Universal: react to the host ---------------------------------------
Subscribe to everything the host asks of this equipment. The daemon streams
HostRequest messages for as long as the call stays open.
Delivery contract (v1):
- firehose: every subscriber receives every host request;
- NO buffering: a command arriving while no client is subscribed is
answered with its declarative ack from the equipment config (the
pre-daemon behaviour) and is NOT replayed on reconnect — the daemon
never tells the host "will finish later" for work no tool will do;
- when a Command does arrive here, the host has ALREADY been answered
with S2F42 HCACK=4; report the real outcome via FireEvent/SetAlarm.
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def CompleteCommand(self, request, context):
"""Report the outcome of a Command delivered on the stream, quoting its `id`.
NOTE the contract (SEMI-conformant, non-blocking): the daemon has ALREADY
answered the host with S2F42 HCACK=4 ("accepted, will finish later") when
it pushed the command onto the stream — the host's transaction is closed.
CompleteCommand therefore correlates/audits the command lifecycle; the
host learns the real outcome via the events/alarms you fire (FireEvent /
SetAlarm), exactly as E30 intends. A synchronous gating mode (tool decides
the HCACK before S2F42 goes out) is a possible v2 extension.
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def ReportProcessJob(self, request, context):
"""---- Jobs / Carriers: report progress of work the host asked for --------
Keyed by the durable id (job_id / carrier_id), not a per-message id — these
are long-lived objects you report against as the physical work proceeds.
E40 — job-based tools
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def ReportCarrier(self, request, context):
"""E87 — carrier-based tools
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def WatchHealth(self, request, context):
"""---- Diagnostics --------------------------------------------------------
Live daemon/link status: distinguishes "host went offline" from "cable
unplugged" from "spool filling up". Streams a snapshot on every change.
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def add_EquipmentServicer_to_server(servicer, server):
rpc_method_handlers = {
'SetVariables': grpc.unary_unary_rpc_method_handler(
servicer.SetVariables,
request_deserializer=equipment__pb2.VariableUpdate.FromString,
response_serializer=equipment__pb2.Ack.SerializeToString,
),
'GetVariables': grpc.unary_unary_rpc_method_handler(
servicer.GetVariables,
request_deserializer=equipment__pb2.VariableQuery.FromString,
response_serializer=equipment__pb2.VariableSnapshot.SerializeToString,
),
'FireEvent': grpc.unary_unary_rpc_method_handler(
servicer.FireEvent,
request_deserializer=equipment__pb2.Event.FromString,
response_serializer=equipment__pb2.Ack.SerializeToString,
),
'SetAlarm': grpc.unary_unary_rpc_method_handler(
servicer.SetAlarm,
request_deserializer=equipment__pb2.Alarm.FromString,
response_serializer=equipment__pb2.Ack.SerializeToString,
),
'ClearAlarm': grpc.unary_unary_rpc_method_handler(
servicer.ClearAlarm,
request_deserializer=equipment__pb2.Alarm.FromString,
response_serializer=equipment__pb2.Ack.SerializeToString,
),
'GetControlState': grpc.unary_unary_rpc_method_handler(
servicer.GetControlState,
request_deserializer=equipment__pb2.Empty.FromString,
response_serializer=equipment__pb2.ControlState.SerializeToString,
),
'RequestControlState': grpc.unary_unary_rpc_method_handler(
servicer.RequestControlState,
request_deserializer=equipment__pb2.ControlStateRequest.FromString,
response_serializer=equipment__pb2.Ack.SerializeToString,
),
'Subscribe': grpc.unary_stream_rpc_method_handler(
servicer.Subscribe,
request_deserializer=equipment__pb2.SubscribeRequest.FromString,
response_serializer=equipment__pb2.HostRequest.SerializeToString,
),
'CompleteCommand': grpc.unary_unary_rpc_method_handler(
servicer.CompleteCommand,
request_deserializer=equipment__pb2.CommandResult.FromString,
response_serializer=equipment__pb2.Ack.SerializeToString,
),
'ReportProcessJob': grpc.unary_unary_rpc_method_handler(
servicer.ReportProcessJob,
request_deserializer=equipment__pb2.ProcessJobState.FromString,
response_serializer=equipment__pb2.Ack.SerializeToString,
),
'ReportCarrier': grpc.unary_unary_rpc_method_handler(
servicer.ReportCarrier,
request_deserializer=equipment__pb2.CarrierState.FromString,
response_serializer=equipment__pb2.Ack.SerializeToString,
),
'WatchHealth': grpc.unary_stream_rpc_method_handler(
servicer.WatchHealth,
request_deserializer=equipment__pb2.Empty.FromString,
response_serializer=equipment__pb2.Health.SerializeToString,
),
}
generic_handler = grpc.method_handlers_generic_handler(
'secsgem.v1.Equipment', rpc_method_handlers)
server.add_generic_rpc_handlers((generic_handler,))
# This class is part of an EXPERIMENTAL API.
class Equipment(object):
"""=============================================================================
SECS/GEM Equipment API
The daemon *is* one piece of SEMI equipment: it owns the HSMS link to the
host (MES) and speaks GEM on your behalf. Your tool software connects as a
client and only ever does two things:
• tell the equipment about itself — set variables, fire events, raise alarms
• react to what the host asks for — receive commands/jobs, answer them
Everything SECS lives inside the daemon: message framing, report definitions,
the GEM state machines, timers, spooling. You need no SEMI knowledge to use
this API. Items are addressed by the human names from your equipment config
(e.g. "chamber_pressure"), never by numeric SVID / CEID / ALID.
CAPABILITY TIERS — wire up only what your equipment is:
• Universal — variables, events, alarms, control state, commands. Every tool.
• Carriers — E87 carrier/load-port flows. Only carrier-based equipment.
• Recipes — S7 process-program transfer. Only recipe-driven equipment.
• Jobs — E40 process jobs. Only job-based process/front-end equipment.
If a tier doesn't apply, you simply never receive its HostRequest variants and
never call its report RPCs.
=============================================================================
---- Universal: report state to the host --------------------------------
"""
@staticmethod
def SetVariables(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(request, target, '/secsgem.v1.Equipment/SetVariables',
equipment__pb2.VariableUpdate.SerializeToString,
equipment__pb2.Ack.FromString,
options, channel_credentials,
insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
@staticmethod
def GetVariables(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(request, target, '/secsgem.v1.Equipment/GetVariables',
equipment__pb2.VariableQuery.SerializeToString,
equipment__pb2.VariableSnapshot.FromString,
options, channel_credentials,
insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
@staticmethod
def FireEvent(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(request, target, '/secsgem.v1.Equipment/FireEvent',
equipment__pb2.Event.SerializeToString,
equipment__pb2.Ack.FromString,
options, channel_credentials,
insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
@staticmethod
def SetAlarm(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(request, target, '/secsgem.v1.Equipment/SetAlarm',
equipment__pb2.Alarm.SerializeToString,
equipment__pb2.Ack.FromString,
options, channel_credentials,
insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
@staticmethod
def ClearAlarm(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(request, target, '/secsgem.v1.Equipment/ClearAlarm',
equipment__pb2.Alarm.SerializeToString,
equipment__pb2.Ack.FromString,
options, channel_credentials,
insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
@staticmethod
def GetControlState(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(request, target, '/secsgem.v1.Equipment/GetControlState',
equipment__pb2.Empty.SerializeToString,
equipment__pb2.ControlState.FromString,
options, channel_credentials,
insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
@staticmethod
def RequestControlState(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(request, target, '/secsgem.v1.Equipment/RequestControlState',
equipment__pb2.ControlStateRequest.SerializeToString,
equipment__pb2.Ack.FromString,
options, channel_credentials,
insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
@staticmethod
def Subscribe(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_stream(request, target, '/secsgem.v1.Equipment/Subscribe',
equipment__pb2.SubscribeRequest.SerializeToString,
equipment__pb2.HostRequest.FromString,
options, channel_credentials,
insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
@staticmethod
def CompleteCommand(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(request, target, '/secsgem.v1.Equipment/CompleteCommand',
equipment__pb2.CommandResult.SerializeToString,
equipment__pb2.Ack.FromString,
options, channel_credentials,
insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
@staticmethod
def ReportProcessJob(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(request, target, '/secsgem.v1.Equipment/ReportProcessJob',
equipment__pb2.ProcessJobState.SerializeToString,
equipment__pb2.Ack.FromString,
options, channel_credentials,
insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
@staticmethod
def ReportCarrier(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(request, target, '/secsgem.v1.Equipment/ReportCarrier',
equipment__pb2.CarrierState.SerializeToString,
equipment__pb2.Ack.FromString,
options, channel_credentials,
insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
@staticmethod
def WatchHealth(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_stream(request, target, '/secsgem.v1.Equipment/WatchHealth',
equipment__pb2.Empty.SerializeToString,
equipment__pb2.Health.FromString,
options, channel_credentials,
insecure, call_credentials, compression, wait_for_ready, timeout, metadata)