Function System

The function system is the operation layer used by device APIs in PlestyLib.

It lets you define operation signatures with structured input and output metadata, then automatically exposes callable methods with validation, casting, and optional response parsing.

Core implementation: FunctionSystem in plesty.lib.device.funcs.

What It Does

FunctionSystem provides:

  1. Dynamic registration of operations as instance methods.

  2. Input schema validation through FuncParam metadata.

  3. Output schema validation through FuncOutput metadata.

  4. Automatic casting for basic types and Plesty data types.

  5. Optional response parser binding per operation.

  6. Auto-generated operation docs and summaries.

Core Data Models

FuncParam

Input field metadata:

  1. name

  2. dtype

  3. default

  4. required

  5. options

  6. range

  7. shape

  8. item_dtype

  9. unit

  10. description

FuncOutput

Output field metadata:

  1. name

  2. dtype

  3. required

  4. code_mapping

  5. range

  6. shape

  7. item_dtype

  8. headers (for PlestyTable2D)

  9. unit

  10. description

FuncMeta

Lightweight metadata that links a PLESTY operation to a single device-protocol command, enabling generic solvers (such as SCPISolver) to dispatch simple operations without hand-written logic.

FuncMeta is built automatically from a "command" field in op_schema.json and injected into the request dictionary by FunctionSystem. You rarely construct it manually.

Fields:

Field

Type

Description

name

str

Operation name as registered in the function schema

command

str

Raw device command string to send (e.g. "MEAS:POW?")

output_key

str

Key under which the parsed value appears in the response dict

output_dtype

type

Python type used to cast the raw string response (default: float)

When to use "command" in op_schema.json:

Add a "command" field to any single-command operation whose response is a scalar:

{
  "measure_power": {
    "command": "MEAS:SCAL:POW",
    "output_key": "power",
    "output_dtype": "float",
    "iparams": {},
    "oparams": {
      "power": { "type": "float", "unit": "W", "required": true }
    }
  }
}

When FunctionSystem processes this entry it builds a FuncMeta and stores it in the request dictionary. A compatible solver (e.g. SCPISolver.solve_func) reads it and returns {"power": <cast_value>} automatically.

Complex operations that require multi-step exchanges or custom response parsing should omit "command" and implement dispatch in a custom OpSolver.

Operation Execution Model

Each registered operation becomes a callable method on the object.

The call pipeline is:

  1. Normalize call arguments.

  2. Validate and cast inputs (iparams).

  3. Build request dictionary.

  4. Execute via bound solver (self._solver).

  5. Optionally parse response (resp_parser).

  6. Validate and cast outputs (oparams).

Runtime request payload sent to solver:

{
	"op": "operation_name",
	"params": {...}
}

Accepted Call Styles

Registered operation methods support all three styles:

  1. func({"a": 1, "b": 2})

  2. func(a=1, b=2)

  3. func(1, 2) mapped by iparams order

Register Operations Programmatically

from plesty.lib.device.funcs import FunctionSystem, FuncParam, FuncOutput


def fake_solver(request: dict) -> dict:
	if request["op"] == "set_power":
		return {"ok": True, "applied": request["params"]["value"]}
	return {"ok": False}


fs = FunctionSystem(solver=fake_solver)

fs.register_func(
	"set_power",
	iparams=[
		FuncParam(name="value", dtype=float, required=True, range=(0.0, 10.0), unit="watt")
	],
	oparams=[
		FuncOutput(name="ok", dtype=bool, required=True),
		FuncOutput(name="applied", dtype=float, required=True, unit="watt"),
	],
)

print(fs.set_power(value=2.5))

Register from Operation Schema

Use register_from_op_schema to load operations from a JSON schema file.

Example schema:

{
  "set_wavelength": {
	"iparams": {
	  "value": {
		"type": "float",
		"unit": "nm",
		"required": true,
		"range": [400.0, 1700.0],
		"description": "Target wavelength"
	  }
	},
	"oparams": {
	  "ok": {
		"type": "bool",
		"required": true
	  }
	}
  }
}

Load from file:

from plesty.lib.device.funcs import FunctionSystem

fs = FunctionSystem(solver=my_solver)
registered = fs.register_from_op_schema("op_schema.json")
print(registered)

Notes:

  1. register_from_op_schema expects a file path.

  2. Keys starting with _ are attached as attributes and skipped as callable operations.

Response Parsers

Bind a parser when raw solver responses need custom transformation:

from plesty.lib.device.device_utils import ResponseParser


class MyParser(ResponseParser):
	def __call__(self, response, param=None, **kwargs):
		return {"ok": bool(response.get("status") == "ok")}


fs.bind_op_response_parser("set_wavelength", MyParser())

Parser rules:

  1. Parser receives raw response and operation output metadata.

  2. Parser must return a dictionary.

  3. Output validation still runs after parser unless no oparams are declared.

Output Type Casting

FunctionSystem supports advanced output casting:

  1. Basic types (int, float, bool, str).

  2. PlestyArray from list/tuple/ndarray.

  3. PlestyTable2D from row-dict list or column dictionary, using declared table headers.

For PlestyTable2D, define headers in schema or metadata, otherwise casting fails.

Documentation and Summaries

You can export operation summaries using func_summary:

  1. style="short" concise signature list.

  2. style="md" markdown documentation.

  3. style="dict" machine-readable dictionary.

  4. style="google" generated stub module with Google-style docstrings.

Example:

print(fs.func_summary(style="short"))
fs.func_summary(style="md", filename="ops.md")

Integration with BaseDeviceSyncModel

BaseDeviceSyncModel initializes FunctionSystem and accepts op_schema in its constructor.

Example:

from plesty.lib.device.base_device_sync import BaseDeviceSyncModel


class MyDevice(BaseDeviceSyncModel):
	def __init__(self, id: str, op_schema: str):
		super().__init__(id=id, op_schema=op_schema)

	def connect(self):
		...

	def disconnect(self):
		...

	def check_errors(self) -> list[str]:
		return []

	def check_operatability(self) -> bool:
		return True

	def _write_(self, key, value) -> bool:
		return True

	def _query_(self, key) -> str:
		return "0"

Common Errors

  1. RuntimeError: no solver bound.

  2. TypeError: wrong positional count or non-dict response.

  3. KeyError: unexpected input keys or missing required keys.

  4. ValueError: failed cast, range check, or options check.

Best Practices

  1. Define precise dtype, required, range, and options in iparams.

  2. Define oparams whenever response contract is known.

  3. Keep transport/protocol logic in solver, not in operation wrapper functions.

  4. Use response parsers for protocol-specific normalization only.

  5. Export func_summary(style="md") as part of API documentation workflows.