plesty.lib.device.base_device_sync

Base synchronous device model for Plesty.

Classes

BaseDeviceSyncModel

Abstract base class for synchronous device models in Plesty.

Module Contents

class plesty.lib.device.base_device_sync.BaseDeviceSyncModel(id: str, op_schema=None, param_schema=None, op_solver=None, cmd_solver=None, data_path: str | os.PathLike[str] | None = None, allowed_data_roots: collections.abc.Iterable[str | os.PathLike[str]] | None = None, **kwargs)

Bases: abc.ABC, plesty.lib.device.params.ConfigSystem, plesty.lib.device.funcs.FunctionSystem, plesty.lib.device.telemetry.TelemetrySystem

Abstract base class for synchronous device models in Plesty.

Initialize the device model with optional schemas and solvers.

Parameters:

id (str) – Device identifier, combined with the host IP into the resource key.
op_schema – Optional operation schema for the function system.
param_schema – Optional parameter schema for the configuration system.
op_solver – Optional operation solver bound to the function system.
cmd_solver – Optional command solver bound to the configuration system.
data_path (str | os.PathLike[str] | None) – Default directory for acquired data, used until a client calls set_data_path(). Device modules read it from their own configuration source (YAML, .env) and pass it in — the library does not prescribe a format.
allowed_data_roots (collections.abc.Iterable[str | os.PathLike[str]] | None) – Directories that client-supplied paths must resolve under. Defaults to [data_path], so a device can only be redirected inside its own data directory unless wider roots are declared explicitly.
**kwargs – Extra keyword arguments, accepted and ignored so subclasses can pass their own construction options up the chain.

Raises:

ValueError – If data_path does not resolve under allowed_data_roots.

id

_resources: list[Any] = []

_resource_manager: plesty.lib.service.resource_manager.ResourceManager | None = None

_allowed_data_roots: list[pathlib.Path]

_root_spellings: list[tuple[pathlib.Path, pathlib.Path]]

_default_data_path: pathlib.Path | None = None

_data_path: pathlib.Path | None = None

_data_path_display: pathlib.Path | None

register_resources(resources: list[Any]) → None

Declare the physical resources this device owns.

Extends the device’s resource inventory and registers the same set with the embedded ResourceManager so that direct (non-TCP) clients can request exclusive access via allocate_resource().

Parameters:: resources (list[Any]) – Resource identifiers in any format accepted by ResourceManager (flat list, set, or nested dict tree).
Return type:: None

allocate_resource(resources: Any, client_id: str) → bool

Request exclusive access to one or more device resources.

Delegates to the embedded ResourceManager. Returns True if the allocation succeeded, False if any requested resource is already held by another client. When no resources have been registered, all requests are granted.

Parameters:

resources (Any) – Resource identifiers (str, list, or dict tree).
client_id (str) – Stable identifier for the requesting client.

Return type:

bool

release_resource(client_id: str) → None

Release all resources previously allocated to client_id.

Parameters:: client_id (str) – Identifier of the client whose allocation should be freed.
Return type:: None

client_resources(client_id: str) → list[str]

Return the canonical resource paths currently allocated to client_id.

Parameters:: client_id (str) – Identifier of the client to query.
Return type:: list[str]

_is_allowed_data_path(path: pathlib.Path) → bool

Return whether path resolves under one of the allowed data roots.

Parameters:: path (pathlib.Path)
Return type:: bool

_display_path(resolved: pathlib.Path) → pathlib.Path

Re-base resolved onto the configured spelling of its data root.

Path.resolve() rewrites a mapped Windows drive to its UNC form, which the operator never configured and some vendor software rejects. A path under a known root is expressed under that root’s configured spelling; anything else is returned as it came.

Parameters:: resolved (pathlib.Path)
Return type:: pathlib.Path

_resolve_data_path(path: str | os.PathLike[str]) → pathlib.Path

Resolve path to an absolute directory, expanding ~ and following links.

A relative path is taken relative to the configured default data path (or, when only roots are configured, the first allowed root), so a client can route a run with set_data_path("session-42") without knowing the server’s filesystem layout.

Parameters:: path (str | os.PathLike[str]) – Absolute or relative directory supplied by the caller.
Raises:: ValueError – If path is relative and no data root is configured.
Return type:: pathlib.Path

set_data_path(path: str | os.PathLike[str], **kwargs) → str

Set the directory where this device stores acquired data.

The path is resolved (~ expanded, .. collapsed, symlinks followed) and must lie under one of the device’s allowed data roots; it is created if missing. All clients of a device server share one data path — the last call wins, matching the single-operator resource-lock model.

Parameters:

path (str | os.PathLike[str]) – Target directory, absolute or relative to the device’s default data path.
**kwargs – Optional extras such as resources, ignored here.

Returns:

The directory now in effect, in the configured spelling of its data root — a mapped drive stays a mapped drive; resolution is used for the containment checks only.

Raises:

ValueError – If the resolved path is outside the allowed roots, or the device has no data path configured at all.

Return type:

str

get_data_path(**kwargs) → str

Return the directory currently used for acquired data.

Device code that writes acquisition files should read the directory from here rather than from its own configuration, so client-set paths take effect.

Parameters:: **kwargs – Optional extras such as resources, ignored here.
Returns:: the last one set by a client, or the device’s configured default. The directory is created if missing — the configured default is otherwise never created anywhere (construction is side-effect-free), and vendor software validating the directory rejects one that does not exist (plesty-lib#31).
Return type:: The directory in effect, in its configured spelling
Raises:: RuntimeError – If the device has no data path configured.

abstractmethod init(main=None) → None

Initialize the device. This should be called before any other operations.

Return type:: None

property is_operatable: bool

Check if the device is currently operatable (connected and responsive).

Return type:: bool

property state: dict

Return a dictionary representing the current device state with all parameters.

Return type:: dict

get_state() → dict[str, Any]

Return a dictionary of all configuration parameters and their current values.

Return type:: dict[str, Any]

synchronize_param_from_device(keys: list[str] | None = None, sync_constraints: bool = False) → None

Synchronizing the model’s configuration with the actual device.

Parameters:

keys (list[str] | None)
sync_constraints (bool)

Return type:

None

doc_model() → plesty.lib.device.doc.DeviceDocModel

Return this device’s structured, render-agnostic documentation model.

The model bundles the configuration parameters (grouped) and every callable operation (schema-registered plus @expose_to_api custom methods, unified). Documentation renderers consume it instead of re-deriving the metadata.

Returns:: The DeviceDocModel for this device.
Return type:: plesty.lib.device.doc.DeviceDocModel

summary(style: str = 'short', filename=None) → str

Summarize both configuration parameters and registered functions.

Parameters:

style (str) – One of short, md, or google.
filename – Optional file path to save the rendered summary.

Returns:

A rendered summary string combining parameter and function sections.

Return type:

str

get_query_command(key: str) → str | None

Return the command that would be sent to query key, sending nothing.

An accessor, not an action — which is the whole difference from query(). query(key) performs the exchange, consumes the reply and returns the value; this returns the command, touches no I/O and leaves the session exactly as it found it. The naming follows the split already in this class: get_* reads what the device declares, a bare verb talks to it.

It exists so a caller can transmit a query without reading its answer — the orphaned reply the buffer-drain gate needs. That takes the command and the transport’s orphan(), and the command is the half only the device knows.

Parameters:: key (str) – Configuration key.
Returns:: The command, or None when this device cannot produce one — the base implementation, since a solver is bound by the subclass. The drain gate then skips and says so.
Return type:: str | None

query_param_range(key: str) → tuple[float | int | None, float | int | None]

Query the device for the valid range of a numeric parameter.

Parameters:: key (str)
Return type:: tuple[float | int | None, float | int | None]

query_param_options(key: str) → list[Any] | None

Query the device for the valid options of a categorical parameter.

Parameters:: key (str)
Return type:: list[Any] | None

abstractmethod connect() → Any

Placeholder method for connecting to the device.

Return type:: Any

abstractmethod disconnect() → Any

Placeholder method for disconnecting from the device.

Return type:: Any

abstractmethod identity() → str

Query the device for its identity string.

Return type:: str

abstractmethod check_errors() → list[str]

Query the device for any error messages.

Return type:: list[str]

abstractmethod check_operatability() → bool

Check if the device is currently operatable (connected and responsive).

Return type:: bool

__enter__(): Enter the device context: acquire the resource lock, init and connect.

__exit__(exc_type, exc_value, tb)

Exit the device context, disconnecting and releasing resources.

The resource lock is released even when disconnect() fails. A transport that throws on the way out — a clear refused by a wedged session, a close on an already-dead handle — must not leave the device marked as in use: every later attempt then fails with “Resource already in use” against an instrument nothing is holding, and the process has to be restarted to clear it.

response_to_target_type(key, response) → Any

Parse a raw device response into the target type for the given parameter key.

Return type:: Any

_user_permission_check(user_resources: list[Any], key: str) → bool

Check if the client is permitted to access key given its allocated resources.

The default implementation is permissive (returns True). Subclasses should override this method with device-specific logic, and may call client_resources() or inspect _resource_manager to query current allocations.

Parameters:

user_resources (list[Any]) – Resources reported by the client for this request.
key (str) – The configuration parameter or function key being accessed.

Return type:

bool

write(key: str, value: str | float | int | bool, **kwargs) → bool

Write a configuration parameter value to the device after validation.

Parameters:

key (str) – Name of the parameter to write (as listed on the Parameters page).
value (str | float | int | bool) – New value; validated against the parameter’s type and range before being sent to the device.
**kwargs – Optional extras such as resources for per-client access control.

Returns:

True if the write succeeded, False otherwise (e.g. a permission check failed).

Return type:

bool

query(key: str, **kwargs) → Any

Query a configuration parameter value from the device.

Parameters:

key (str) – Name of the parameter to read (as listed on the Parameters page).
**kwargs – Optional extras such as resources for per-client access control.

Returns:

The parameter’s current value, converted to its declared type.

Return type:

Any

abstractmethod _write_(key: str, value: str | float | int | bool) → bool

Placeholder method for writing a command to the device.

Parameters:

key (str)
value (str | float | int | bool)

Return type:

bool

abstractmethod _query_(key: str) → str

Placeholder method for querying the device.

Parameters:: key (str)
Return type:: str

reset() → None

Reset the device to a known state.

Return type:: None

clear() → None

Clear the device state.

Return type:: None