plestylib.device.funcs#

Classes#

FuncParam

Data class to hold function argument metadata.

FuncOutput

Data class to hold function output metadata.

FunctionSystem

Dynamic operation-to-function adapter for TCP message-based APIs.

Module Contents#

class plestylib.device.funcs.FuncParam#

Data class to hold function argument metadata.

name: str#
dtype: Any#
unit: str = ''#
default: Any = None#
required: bool = True#
options: list[Any] = None#
range: tuple[Any, Any] = None#
shape: tuple[Any, Ellipsis] | None = None#
item_dtype: Any = None#
description: str = ''#
class plestylib.device.funcs.FuncOutput#

Data class to hold function output metadata.

name: str#
dtype: Any#
unit: str = ''#
required: bool = True#
code_mapping: dict[Any, str] = None#
range: tuple[Any, Any] = None#
shape: tuple[Any, Ellipsis] | None = None#
item_dtype: Any = None#
headers: list[plestylib.data.TableHeader] | None = None#
description: str = ''#
class plestylib.device.funcs.FunctionSystem(op_schema=None, solver: Callable[[dict[str, Any]], dict[str, Any]] = None)#

Dynamic operation-to-function adapter for TCP message-based APIs.

This class allows developers to register operations as Python callables, where: - iparams is a list of FuncParam objects describing input schema. - oparams is a list of FuncOutput objects describing output schema. - the runtime request format is {'op': <operation>, 'parameters': <dict>}.

Registered functions can be called with a parameter dictionary, keyword arguments, or positional arguments (mapped by iparams order). Each call returns a response dictionary from the configured executor.

Parameters:

solver (Callable[[dict[str, Any]], dict[str, Any]]) – Callable used to send a message dictionary and return a response dictionary.

_functions#
_solver = None#
bind_op_solver(solver: Callable[[dict[str, Any]], dict[str, Any]])#

Bind or replace the callable used to execute operation requests.

Parameters:

solver (Callable[[dict[str, Any]], dict[str, Any]]) – Callable receiving request dictionary and returning response dictionary.

bind_op_response_parser(op_name: str, parser: plestylib.device.device_utils.ResponseParser)#

Bind a response parser to a registered operation.

Parameters:
register_func(func_name: str, iparams: list[FuncParam] = None, oparams: list[FuncOutput] = None, resp_parser: plestylib.device.device_utils.ResponseParser = None, **kwargs)#

Register an operation as a callable function on this instance.

Parameters:

  • func_name (str) – Operation name and exported function name.

  • iparams (list[FuncParam]) – Input schema as [FuncParam(...), ...].

  • oparams (list[FuncOutput]) – Output schema as [FuncOutput(...), ...].

  • resp_parser (plestylib.device.device_utils.ResponseParser) – Optional parser applied to executor response.

  • kwargs – Registration options for future extension. These options are stored as metadata and are not sent in runtime requests.

_parse_iparams_from_schema(iparams_schema: Any) → list[FuncParam]#

Parse operation input schema into FuncParam objects.

Parameters:

iparams_schema (Any)

Return type:

list[FuncParam]

_parse_oparams_from_schema(oparams_schema: Any) → list[FuncOutput]#

Parse operation output schema into FuncOutput objects.

Parameters:

oparams_schema (Any)

Return type:

list[FuncOutput]

register_from_op_schema(schema_path: str, resp_parsers: dict[str, plestylib.device.device_utils.ResponseParser] = None, **registration_options) → list[str]#

Register multiple operations from a JSON schema file.

Expected schema format:

{
    "operation_name": {
        "iparams": {"arg_name": {"type": "float"}},
        "oparams": {"out_name": {"type": "str"}}
    }
}
Parameters:

  • schema_path (str) – Path to JSON schema file.

  • resp_parsers (dict[str, plestylib.device.device_utils.ResponseParser]) – Optional map {operation_name: ResponseParser}.

  • **registration_options – Additional options forwarded to register_func.

Returns:

Names of registered operations.

Return type:

list[str]

_normalize_iparams(iparams: list[Any]) → list[FuncParam]#

Normalize input schema into FuncParam objects.

Parameters:

iparams (list[Any])

Return type:

list[FuncParam]

_normalize_oparams(oparams: list[Any]) → list[FuncOutput]#

Normalize output schema into FuncOutput objects.

Parameters:

oparams (list[Any])

Return type:

list[FuncOutput]

static _type_name(dtype: Any) → str#

Return a human-readable type name.

Parameters:

dtype (Any)

Return type:

str

static _is_plesty_array_dtype(dtype: Any) → bool#
Parameters:

dtype (Any)

Return type:

bool

static _is_plesty_table2d_dtype(dtype: Any) → bool#
Parameters:

dtype (Any)

Return type:

bool

_parse_table_headers(headers_cfg: Any) → list[plestylib.data.TableHeader] | None#
Parameters:

headers_cfg (Any)

Return type:

list[plestylib.data.TableHeader] | None

_cast_plesty_array(value: Any) → plestylib.data.PlestyArray#
Parameters:

value (Any)

Return type:

plestylib.data.PlestyArray

_cast_table_value(value: Any, dtype: Any)#
Parameters:

  • value (Any)

  • dtype (Any)

_cast_plesty_table2d(value: Any, output_meta: FuncOutput) → plestylib.data.PlestyTable2D#
Parameters:
Return type:

plestylib.data.PlestyTable2D

_build_operation_doc(func_name, iparams, oparams, resp_parser) → str#

Build detailed user-facing documentation text for a registered operation.

Return type:

str

func_summary(style: str = 'short', filename=None) → str | dict[str, dict[str, Any]]#

Summarize all registered operations.

Parameters:

  • style (str) – If “short”, return a brief summary. If “md”, return a detailed documentation in Markdown format. If “dict”, return a structured dictionary. If “google”, return a detailed documentation in Google style docstring format with function signatures.

  • filename – Optional path to save the summary text.

Returns:

Operation documentation.

Return type:

str | dict[str, dict[str, Any]]

Notes

The summary is built from registration metadata and can be used as user-facing documentation for available operations.

_normalize_args(func_name: str, iparams: list[FuncParam], args, kwargs) → dict[str, Any]#

Normalize incoming call arguments into a single parameters dictionary.

Accepts one dictionary positional argument, keyword arguments, or positional arguments mapped by iparams order.

Parameters:
Return type:

dict[str, Any]

_validate_input(func_name: str, iparams: list[FuncParam], params: dict[str, Any]) → dict[str, Any]#

Validate and cast input values based on registered iparams schema.

Raises:

  • KeyError – If unexpected keys are provided.

  • ValueError – If type casting fails.

Parameters:

  • func_name (str)

  • iparams (list[FuncParam])

  • params (dict[str, Any])

Return type:

dict[str, Any]

_validate_output(func_name: str, oparams: list[FuncOutput], response: dict[str, Any]) → dict[str, Any]#

Validate and cast response fields against registered oparams schema.

Raises:

  • KeyError – If expected response keys are missing.

  • ValueError – If type casting fails.

Parameters:

  • func_name (str)

  • oparams (list[FuncOutput])

  • response (dict[str, Any])

Return type:

dict[str, Any]

_create_func(func_name, iparams, oparams)#

Create the executable operation function for a registered operation.