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 plestylib.device.funcs.
What It Does#
FunctionSystem provides:
Dynamic registration of operations as instance methods.
Input schema validation through
FuncParammetadata.Output schema validation through
FuncOutputmetadata.Automatic casting for basic types and Plesty data types.
Optional response parser binding per operation.
Auto-generated operation docs and summaries.
Core Data Models#
FuncParam#
Input field metadata:
namedtypedefaultrequiredoptionsrangeshapeitem_dtypeunitdescription
FuncOutput#
Output field metadata:
namedtyperequiredcode_mappingrangeshapeitem_dtypeheaders(forPlestyTable2D)unitdescription
Operation Execution Model#
Each registered operation becomes a callable method on the object.
The call pipeline is:
Normalize call arguments.
Validate and cast inputs (
iparams).Build request dictionary.
Execute via bound solver (
self._solver).Optionally parse response (
resp_parser).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:
func({"a": 1, "b": 2})func(a=1, b=2)func(1, 2)mapped byiparamsorder
Register Operations Programmatically#
from plestylib.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 plestylib.device.funcs import FunctionSystem
fs = FunctionSystem(solver=my_solver)
registered = fs.register_from_op_schema("op_schema.json")
print(registered)
Notes:
register_from_op_schemaexpects a file path.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 plestylib.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:
Parser receives raw response and operation output metadata.
Parser must return a dictionary.
Output validation still runs after parser unless no
oparamsare declared.
Output Type Casting#
FunctionSystem supports advanced output casting:
Basic types (
int,float,bool,str).PlestyArrayfrom list/tuple/ndarray.PlestyTable2Dfrom 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:
style="short"concise signature list.style="md"markdown documentation.style="dict"machine-readable dictionary.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 plestylib.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#
RuntimeError: no solver bound.TypeError: wrong positional count or non-dict response.KeyError: unexpected input keys or missing required keys.ValueError: failed cast, range check, or options check.
Best Practices#
Define precise
dtype,required,range, andoptionsiniparams.Define
oparamswhenever response contract is known.Keep transport/protocol logic in solver, not in operation wrapper functions.
Use response parsers for protocol-specific normalization only.
Export
func_summary(style="md")as part of API documentation workflows.