raphael
8a55137e57
Interface cleanup so the report_* family matches the typo-safe ethos of eq.names instead of leaking raw protobuf errors on a misspelled value. - Milestone / ModuleState / JobState: importable str-enums (member == its wire name, so plain strings still work) — autocomplete + a typo-checked happy path. The clean rule: equipment-specific *names* live on eq.names; fixed protocol *value-sets* are enums. - _enum_value(): resolves an enum-or-string arg client-side and, on a bad value, raises ValueError with a close-match hint *before* the wire. Wired into report_job / report_substrate / report_module / request_control_state (all previously raised a raw protobuf ValueError). - Equipment is now a context manager (with Equipment(...) as eq: ...). - examples/wafer_tool.py: a cluster tool tracking one wafer through one module end-to-end (E90 + E157), showing the enums + context manager. - tests/test_enums.py: asserts the enums stay in lockstep with the proto and that the typo path is helpful. Wired into run_interop.sh (pyclient step). - Interop drives both the enum and string forms on the wire + the ValueError typo path. Docs (ch16/ch42) updated; names-vs-enums rule documented. All Python unit tests + 25 pyclient interop checks pass. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
377 lines
15 KiB
Python
377 lines
15 KiB
Python
"""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(eq.names.event.ProcessStarted) # typo-safe; plain strings also work
|
|
|
|
@eq.command # function name IS the RCMD name,
|
|
def START(cmd): # validated against equipment at import
|
|
run_recipe(cmd.params.get("PPID"))
|
|
eq.fire(eq.names.event.ProcessStarted)
|
|
|
|
eq.listen() # blocks, dispatching host commands to your handlers
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import difflib
|
|
import enum
|
|
import threading
|
|
from dataclasses import dataclass
|
|
from types import SimpleNamespace
|
|
from typing import Callable, Dict, Iterator, Optional, Union
|
|
|
|
import grpc
|
|
|
|
from ._enums import JobState, Milestone, ModuleState
|
|
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)
|
|
|
|
|
|
def _enum_value(proto_enum, value, kind: str) -> int:
|
|
"""Resolve a protocol-enum argument (an enum member like Milestone.ARRIVED
|
|
or a plain string like "ARRIVED") to its protobuf int, typo-safely — the
|
|
same close-match courtesy as eq.names, instead of a raw protobuf error."""
|
|
name = value.value if isinstance(value, enum.Enum) else str(value)
|
|
try:
|
|
return proto_enum.Value(name)
|
|
except ValueError:
|
|
valid = list(proto_enum.keys())
|
|
close = difflib.get_close_matches(name, valid, n=3)
|
|
hint = (f" Did you mean {', '.join(close)}?" if close
|
|
else f" Valid: {', '.join(valid)}.")
|
|
raise ValueError(f"unknown {kind} '{name}'.{hint}") from None
|
|
|
|
|
|
@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 ProcessJob:
|
|
"""An E40 process job the host wants run (or stopped/aborted). Do the
|
|
physical work, then report progress with Equipment.report_job()."""
|
|
|
|
id: str
|
|
action: str # "START" | "STOP" | "PAUSE" | "RESUME" | "ABORT"
|
|
recipe: str
|
|
carriers: list
|
|
|
|
|
|
@dataclass
|
|
class Health:
|
|
link: str # "DISCONNECTED" | "CONNECTED" | "SELECTED"
|
|
control_state: str # "HOST_OFFLINE" | "ONLINE_REMOTE" | ...
|
|
spool_depth: int
|
|
|
|
|
|
class _Names:
|
|
"""An autocomplete-able, typo-safe view of the equipment's real names
|
|
(from Describe). `eq.names.event.ProcessStarted` returns "ProcessStarted";
|
|
a wrong name raises AttributeError with close-match suggestions, and the
|
|
set shows up in REPL/IDE completion via __dir__."""
|
|
|
|
def __init__(self, kind: str, names):
|
|
self._kind = kind
|
|
self._names = set(names)
|
|
|
|
def __getattr__(self, attr: str) -> str:
|
|
if attr in self.__dict__.get("_names", ()):
|
|
return attr
|
|
close = difflib.get_close_matches(attr, self._names, n=3)
|
|
hint = f" Did you mean {', '.join(close)}?" if close else ""
|
|
raise AttributeError(
|
|
f"no {self._kind} named '{attr}'.{hint}")
|
|
|
|
def __dir__(self):
|
|
return sorted(self._names)
|
|
|
|
def __contains__(self, name: str) -> bool:
|
|
return name in self._names
|
|
|
|
|
|
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._job_handler: Optional[Callable[[ProcessJob], object]] = None
|
|
self._recipe_handler: Optional[Callable[[str, bytes], object]] = None
|
|
self._constant_handler: Optional[Callable[[str, object], object]] = None
|
|
self._listen_thread: Optional[threading.Thread] = None
|
|
self._stop = threading.Event()
|
|
self._names_cache: Optional[SimpleNamespace] = None
|
|
|
|
# ---- 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()}
|
|
|
|
@property
|
|
def names(self) -> SimpleNamespace:
|
|
"""The equipment's real names, by category, fetched once from the
|
|
daemon (Describe): eq.names.event.* / .alarm.* / .command.* / .var.* /
|
|
.constant.*. Autocomplete-friendly and typo-safe — handy instead of
|
|
bare string literals."""
|
|
if self._names_cache is None:
|
|
d = self._stub.Describe(pb.Empty())
|
|
self._names_cache = SimpleNamespace(
|
|
var=_Names("variable", d.variables),
|
|
event=_Names("event", d.events),
|
|
alarm=_Names("alarm", d.alarms),
|
|
command=_Names("command", d.commands),
|
|
constant=_Names("constant", d.constants))
|
|
return self._names_cache
|
|
|
|
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=_enum_value(pb.ControlState.State, desired, "control state"))
|
|
_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 command(self, fn: Callable[[Command], object]):
|
|
"""Bind a handler to the host command of the SAME NAME as the
|
|
function — no string literal:
|
|
|
|
@eq.command
|
|
def START(cmd): ... # binds the "START" RCMD
|
|
|
|
The function name is validated against the equipment's real command
|
|
set (via Describe), so a typo fails loudly at decoration time."""
|
|
name = fn.__name__
|
|
if name not in self.names.command:
|
|
close = difflib.get_close_matches(name, dir(self.names.command), n=3)
|
|
hint = f" Did you mean {', '.join(close)}?" if close else ""
|
|
raise NameError(f"no host command '{name}' to bind @eq.command to.{hint}")
|
|
self._handlers[name] = fn
|
|
return fn
|
|
|
|
def on_process_job(self, fn: Callable[[ProcessJob], object]):
|
|
"""Decorator: the host wants a process job run/stopped (E40).
|
|
Report progress with report_job()."""
|
|
self._job_handler = fn
|
|
return fn
|
|
|
|
def on_recipe(self, fn: Callable[[str, bytes], object]):
|
|
"""Decorator: the host downloaded a recipe (ppid, body)."""
|
|
self._recipe_handler = fn
|
|
return fn
|
|
|
|
def on_constant_change(self, fn: Callable[[str, object], object]):
|
|
"""Decorator: the host changed an equipment constant (name, value)."""
|
|
self._constant_handler = fn
|
|
return fn
|
|
|
|
def report_job(self, job_id: str, state: Union[JobState, str]) -> None:
|
|
"""Report physical job progress (E40). state: a JobState or its name —
|
|
SETTING_UP | PROCESSING | COMPLETE | ABORTED. The daemon drives the E40
|
|
state machine and notifies the host (S16F9)."""
|
|
req = pb.ProcessJobState(
|
|
job_id=job_id,
|
|
state=_enum_value(pb.ProcessJobState.State, state, "job state"))
|
|
_check(self._stub.ReportProcessJob(req))
|
|
|
|
def report_substrate(self, substrate_id: str,
|
|
milestone: Union[Milestone, str],
|
|
carrier_id: str = "", slot: int = 0) -> None:
|
|
"""E90 wafer tracking. milestone: a Milestone or its name — ARRIVED |
|
|
AT_WORK | PROCESSING | PROCESSED | AT_DESTINATION. ARRIVED records the
|
|
origin (carrier_id, slot) and creates the wafer; the rest drive its
|
|
E90 FSMs, and the daemon emits the standard CEIDs to the host."""
|
|
req = pb.SubstrateReport(
|
|
substrate_id=substrate_id,
|
|
milestone=_enum_value(pb.SubstrateReport.Milestone, milestone, "milestone"),
|
|
carrier_id=carrier_id, slot=slot)
|
|
_check(self._stub.ReportSubstrate(req))
|
|
|
|
def report_module(self, module_id: str,
|
|
state: Union[ModuleState, str]) -> None:
|
|
"""E157 module tracking. state: a ModuleState or its name —
|
|
NOT_EXECUTING | GENERAL_EXECUTING | STEP_EXECUTING | STEP_COMPLETED.
|
|
Auto-creates the module on first report; NOT_EXECUTING resets it."""
|
|
req = pb.ModuleReport(
|
|
module_id=module_id,
|
|
state=_enum_value(pb.ModuleReport.State, state, "module state"))
|
|
_check(self._stub.ReportModule(req))
|
|
|
|
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()
|
|
|
|
def __enter__(self) -> "Equipment":
|
|
return self
|
|
|
|
def __exit__(self, *exc) -> bool:
|
|
self.close()
|
|
return False
|
|
|
|
# ---- 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 hr.HasField("process_job") and self._job_handler:
|
|
j = hr.process_job
|
|
self._job_handler(ProcessJob(
|
|
j.job_id, pb.ProcessJob.Action.Name(j.action),
|
|
j.recipe, list(j.carriers)))
|
|
continue
|
|
if hr.HasField("process_program") and self._recipe_handler:
|
|
self._recipe_handler(hr.process_program.ppid,
|
|
hr.process_program.body)
|
|
continue
|
|
if hr.HasField("constant") and self._constant_handler:
|
|
self._constant_handler(hr.constant.name,
|
|
from_value(hr.constant.value))
|
|
continue
|
|
if not hr.HasField("command"):
|
|
continue # carriers etc.: future variants
|
|
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))
|