"""The Equipment client: a complete GEM tool integration in plain Python. The secs_gemd daemon owns everything SEMI — the HSMS link to the host, the GEM state machines, message formats, timers, spooling. This client only ever does two things: tell the equipment about itself (variables, events, alarms) and react to what the host asks (commands). Names are the ones from your equipment.yaml; values are plain Python. from secsgem_client import Equipment eq = Equipment("localhost:50051") eq.set(ChamberPressure=2.5) eq.fire("ProcessStarted") @eq.on("START") def start(cmd): run_recipe(cmd.params.get("PPID")) eq.listen() # blocks, dispatching host commands to your handlers """ from __future__ import annotations import threading from dataclasses import dataclass from typing import Callable, Dict, Iterator, Optional import grpc from ._proto import equipment_pb2 as pb from ._proto import equipment_pb2_grpc as rpc from ._values import from_value, to_value class SecsGemError(RuntimeError): """The daemon declined a request (unknown name, bad value, wrong state).""" def __init__(self, code: int, message: str): super().__init__(message or pb.Ack.Code.Name(code)) self.code = code def _check(ack: pb.Ack) -> None: if ack.code != pb.Ack.ACCEPT: raise SecsGemError(ack.code, ack.message) @dataclass class Command: """A remote command from the host (e.g. START). The host has already been told "accepted, will finish later" — report the real outcome by firing an event (success) or raising an alarm (failure) from your handler.""" id: str name: str params: Dict[str, object] _eq: "Equipment" def done(self, ok: bool = True) -> None: """Mark the command complete (for the daemon's audit trail). Called automatically after your handler returns; call early if you want.""" self._eq._complete(self.id, ok) self._done = True @dataclass class Health: link: str # "DISCONNECTED" | "CONNECTED" | "SELECTED" control_state: str # "HOST_OFFLINE" | "ONLINE_REMOTE" | ... spool_depth: int class Equipment: """A connection to the secs_gemd daemon — your equipment, by name.""" def __init__(self, address: str = "localhost:50051", connect_timeout: float = 10.0): self._channel = grpc.insecure_channel(address) grpc.channel_ready_future(self._channel).result(timeout=connect_timeout) self._stub = rpc.EquipmentStub(self._channel) self._handlers: Dict[str, Callable[[Command], object]] = {} self._listen_thread: Optional[threading.Thread] = None self._stop = threading.Event() # ---- report state to the host ------------------------------------------- def set(self, values: Optional[Dict[str, object]] = None, **kwargs) -> None: """Update status/data variables by name: eq.set(ChamberPressure=2.5). The host sees the new values immediately when it polls.""" merged = dict(values or {}) merged.update(kwargs) req = pb.VariableUpdate() for name, v in merged.items(): req.values[name].CopyFrom(to_value(v)) _check(self._stub.SetVariables(req)) def get(self, *names: str) -> Dict[str, object]: """Read variables back from the equipment. No names = all of them.""" try: snap = self._stub.GetVariables(pb.VariableQuery(names=list(names))) except grpc.RpcError as e: raise SecsGemError(pb.Ack.PARAMETER_INVALID, e.details()) from None return {k: from_value(v) for k, v in snap.values.items()} def __setitem__(self, name: str, value) -> None: self.set({name: value}) def __getitem__(self, name: str): return self.get(name)[name] def fire(self, event: str, **data) -> None: """Fire a collection event by name; the daemon assembles the configured report and sends it to the host. kwargs set variable values for this event first: eq.fire("WaferComplete", Thickness=1.2).""" req = pb.Event(name=event) for name, v in data.items(): req.data[name].CopyFrom(to_value(v)) _check(self._stub.FireEvent(req)) def alarm(self, name: str) -> None: """Raise an alarm by its config name (or stringified ALID).""" _check(self._stub.SetAlarm(pb.Alarm(name=name))) def clear(self, name: str) -> None: """Clear a previously raised alarm.""" _check(self._stub.ClearAlarm(pb.Alarm(name=name))) # ---- control state & health --------------------------------------------- @property def control_state(self) -> str: """The GEM control state, e.g. "ONLINE_REMOTE".""" cs = self._stub.GetControlState(pb.Empty()) return pb.ControlState.State.Name(cs.state) def request_control_state(self, desired: str) -> None: """Operator-panel transition, e.g. eq.request_control_state("HOST_OFFLINE") before maintenance. Raises if the E30 table says no from here.""" req = pb.ControlStateRequest( desired=pb.ControlState.State.Value(desired)) _check(self._stub.RequestControlState(req)) def health(self) -> Health: """One health snapshot (link / control state / spool depth).""" for h in self._stub.WatchHealth(pb.Empty()): return Health(pb.Health.LinkState.Name(h.link), pb.ControlState.State.Name(h.control_state), h.spool_depth) raise SecsGemError(pb.Ack.CANNOT_DO_NOW, "health stream ended") def watch_health(self) -> Iterator[Health]: """Yields a Health snapshot on every link/state change.""" for h in self._stub.WatchHealth(pb.Empty()): yield Health(pb.Health.LinkState.Name(h.link), pb.ControlState.State.Name(h.control_state), h.spool_depth) # ---- react to the host ---------------------------------------------------- def on(self, command: str): """Decorator: run this function when the host sends `command`. Use "*" to catch commands with no specific handler.""" def register(fn: Callable[[Command], object]): self._handlers[command] = fn return fn return register def listen(self, background: bool = False) -> None: """Consume host requests and dispatch them to @eq.on handlers. Blocks; pass background=True to run on a daemon thread instead.""" if background: self._listen_thread = threading.Thread(target=self._listen_loop, daemon=True) self._listen_thread.start() return self._listen_loop() def close(self) -> None: self._stop.set() self._channel.close() # ---- internals ------------------------------------------------------------- def _listen_loop(self) -> None: try: for hr in self._stub.Subscribe(pb.SubscribeRequest(client="secsgem_client")): if self._stop.is_set(): return if not hr.HasField("command"): continue # future HostRequest variants (jobs, carriers, ...) cmd = hr.command handler = self._handlers.get(cmd.name) or self._handlers.get("*") command = Command(cmd.id, cmd.name, {k: from_value(v) for k, v in cmd.params.items()}, self) ok = True try: if handler is not None: handler(command) except Exception: ok = False raise finally: if not getattr(command, "_done", False): self._complete(cmd.id, ok) except grpc.RpcError: if not self._stop.is_set(): raise def _complete(self, command_id: str, ok: bool) -> None: ack = pb.Ack(code=pb.Ack.ACCEPT if ok else pb.Ack.REJECTED) self._stub.CompleteCommand(pb.CommandResult(id=command_id, ack=ack))