plesty-lib · SCPI Dispatch

FuncMeta & SCPISolver.solve_func

Schema-driven operation dispatch — map a JSON command entry to a typed device response in zero hand-written logic.

plesty-lib
⚠️ The Problem
Before FuncMeta

30 operations. 30 identical methods. 30 places to typo a command string.

The classic PLESTY device OpSolver was a single solve_op() method containing a long chain of if/elif blocks. Each branch manually called send_command("MEAS:POW?"), stripped whitespace, cast to float, and returned a dict.

For devices with many query operations (power meters, spectrometers, lock-in amplifiers) this produced 30–60 near-identical code paths. Any typo in a command string — "MEAS.POW?" instead of "MEAS:POW?" — was a runtime bug that only surfaced when that specific operation was called. There was no schema to cross-check against. There was no type contract. And because each developer wrote their own variant, modules were inconsistent with each other.

📦 What FuncMeta Is
plesty.lib.device.funcs.FuncMeta

A lightweight @dataclass that links a PLESTY operation name to its device-protocol command. The FunctionSystem builds one automatically from any schema operation that has a "command" field, then injects it into the request dictionary before calling solve_op().

FieldTypeDescription
namestrOperation name as registered in the function schema (e.g. "get_power")
commandstrRaw device command string to send (e.g. "MEAS:POW?")
output_keystrKey under which the parsed value is returned (e.g. "power")
output_dtypetypePython type used to cast the raw string response. Defaults to float.
Source — plesty/lib/device/funcs.py
from dataclasses import dataclass, field

@dataclass
class FuncMeta:
    """Link a PLESTY operation to its device-protocol command."""
    name: str
    command: str
    output_key: str
    output_dtype: type = field(default=float)
⚙️ What solve_func Does
SCPISolver.solve_func — dispatch flow
inputs
op_name
send_fn
func_meta
→
if func_meta is None
return None
else →
send command
send_fn(func_meta.command)
→
cast response
output_dtype(raw.strip())
→
return
{output_key: value}

When func_meta is None — meaning the schema entry has no "command" field — solve_func returns None and the caller's solve_op() handles the operation with custom logic.

Source — plesty/lib/solver/scpi.py
def solve_func(
    self,
    op_name: str,
    send_fn: Callable[[str], Any],
    func_meta: "FuncMeta | None",
) -> "dict[str, Any] | None":
    if func_meta is None:
        return None                            # caller handles custom logic
    raw = send_fn(func_meta.command)           # send SCPI command
    try:
        value = func_meta.output_dtype(str(raw).strip())   # cast response
    except (ValueError, TypeError) as exc:
        raise ValueError(
            f"solve_func: cannot cast response {raw!r} to"
            f" {func_meta.output_dtype.__name__} for op '{op_name}'"
        ) from exc
    return {func_meta.output_key: value}       # return typed dict
📋 Tutorial — Using FuncMeta in Your Device
1

Add a "command" field to your operation schema

In schema_func.json, any operation with a "command" key is automatically handled by solve_func. Operations without "command" fall through to your OpSolver.

// plesty/mydevice/schema_func.json
{
  "operations": {
    "get_power": {
      "command": "MEAS:POW?",           // ← FuncMeta built from this
      "output_key": "power",
      "output_dtype": "float",
      "description": "Measure optical power in Watts."
    },
    "get_wavelength": {
      "command": "SENS:CORR:WAV?",       // ← also handled automatically
      "output_key": "wavelength",
      "output_dtype": "float",
      "description": "Read the correction wavelength in nm."
    },
    "get_zeroed_power": {
      // no "command" field — falls through to OpSolver.solve_op()
      "description": "Return zero-corrected power reading."
    }
  }
}
2

FunctionSystem builds FuncMeta automatically

At device initialisation, FunctionSystem scans your schema. For every operation with a "command" field it constructs a FuncMeta instance and registers it. No code change needed — this happens inside plesty-lib.

When solve_op() is called, FunctionSystem injects the matching FuncMeta (or None) into request["func_meta"].

3

Call solve_func first in your OpSolver

Pass request.get("func_meta") to solve_func. If it returns a dict, you're done. If it returns None, handle the operation manually.

class _MyDeviceOpSolver(OpSolver):
    def __init__(self, tm: TrafficManager) -> None:
        self._tm = tm
        self._scpi = SCPISolver()

    def solve_op(self, op_name: str, request: dict) -> dict:
        # Step 1: try schema-driven dispatch
        result = self._scpi.solve_func(
            op_name,
            self._tm.send_command,
            request.get("func_meta"),
        )
        if result is not None:
            return result   # done — no Python method needed

        # Step 2: handle custom operations
        if op_name == "get_zeroed_power":
            self._tm.send_command("SENS:CORR:COLL:ZERO:INIT")
            raw = self._tm.send_command("MEAS:POW?")
            return {"power": float(raw.strip())}

        raise NotImplementedError(f"Unknown operation: {op_name}")
4

For complex operations — omit "command" and implement in OpSolver

Operations requiring multi-step SCPI exchange, binary responses, or conditional device state should not have a "command" field in the schema. solve_func will return None for them, and your if block in solve_op handles them with full custom logic.

↔️ Before & After
✗ Before — manual dispatch
class Pm100dOpSolver(OpSolver):
  def solve_op(self, op_name, request):
    if op_name == "get_power":
      raw = self._tm.send_command("MEAS:POW?")
      return {"power": float(raw.strip())}
    elif op_name == "get_wavelength":
      raw = self._tm.send_command("SENS:CORR:WAV?")
      return {"wavelength": float(raw.strip())}
    elif op_name == "get_zeroed_power":
      # custom multi-step logic
      self._tm.send_command(
          "SENS:CORR:COLL:ZERO:INIT")
      raw = self._tm.send_command("MEAS:POW?")
      return {"power": float(raw.strip())}
    ...  # 27 more elif blocks
✓ After — schema-driven
class _Pm100dOpSolver(OpSolver):
  def solve_op(self, op_name, request):
    # Simple ops: handled automatically
    result = self._scpi.solve_func(
      op_name,
      self._tm.send_command,
      request.get("func_meta"),
    )
    if result is not None:
      return result

    # Only complex ops reach here
    if op_name == "get_zeroed_power":
      self._tm.send_command(
          "SENS:CORR:COLL:ZERO:INIT")
      raw = self._tm.send_command("MEAS:POW?")
      return {"power": float(raw.strip())}

    raise NotImplementedError(op_name)
🔧 When to Use a Custom OpSolver

FuncMeta handles one command → one scalar. Anything else needs OpSolver.

Omit "command" from the schema entry and implement the logic in solve_op() when your operation:

MULTI-STEP
Requires sending multiple SCPI commands in sequence before reading a response.
NON-SCALAR
Returns a list, array, or structured object that can't be cast with a single type call.
CONDITIONAL
Command string depends on device state, operation arguments, or runtime conditions.
STATEFUL
Needs to read device registers, wait for completion, or retry on error.
📑 API Reference
FuncMeta — plesty.lib.device.funcs
FieldDescription
name: strOperation name matching the schema key.
command: strRaw SCPI command to send (e.g. "MEAS:POW?").
output_key: strDict key for the returned value (e.g. "power").
output_dtype: typeCast target for the raw string response. Default: float.
SCPISolver.solve_func — plesty.lib.solver.scpi
Parameter / ReturnDescription
op_name: strOperation name — used only in error messages.
send_fn: CallableCallable that sends a command string and returns the device response.
func_meta: FuncMeta | NoneInjected by FunctionSystem. None = no schema command, caller must handle.
→ dict | None{output_key: cast_value} on success; None when func_meta is absent.
raises ValueErrorWhen the raw response cannot be cast to output_dtype.