plesty.lib.device.composite_device

Composite device wrapper for managing multiple devices together.

Attributes

_logger

_DEFAULT_CLIENT_TIMEOUT_MS

_DEFAULT_CONNECT_ATTEMPTS

_DEFAULT_BACKOFF_S

_DEFAULT_CONNECT_DEADLINE_S

_MIN_HANDSHAKE_MS

_CLOSE_TIMEOUT_MS

_ADDRESS_ENV_SUFFIX

_CONFIG_KEYS

StatusCallback

_BAD_STATES

Exceptions

DeviceUnreachableError

A sub-device server did not answer within its connect deadline.

Classes

ConnectionState

Connection state of one sub-device, as seen by its composite.

CompositeDevice

A container device that manages multiple sub-devices as a single unit.

Functions

address_env_var(→ str)

Return the environment variable that holds sub-device name's address.

connect_client(→ plesty.lib.service._DeviceTCPIPClient)

Build a remote device client, retrying a transient connection handshake.

_describe_timeout(→ str)

Return the timeout a call ran under, for the retry warning.

Module Contents

plesty.lib.device.composite_device._logger
plesty.lib.device.composite_device._DEFAULT_CLIENT_TIMEOUT_MS = 50000
plesty.lib.device.composite_device._DEFAULT_CONNECT_ATTEMPTS = 4
plesty.lib.device.composite_device._DEFAULT_BACKOFF_S = 2.0
plesty.lib.device.composite_device._DEFAULT_CONNECT_DEADLINE_S = 10.0
plesty.lib.device.composite_device._MIN_HANDSHAKE_MS = 250
plesty.lib.device.composite_device._CLOSE_TIMEOUT_MS = 1000
plesty.lib.device.composite_device._ADDRESS_ENV_SUFFIX = 'ADDRESS'
plesty.lib.device.composite_device._CONFIG_KEYS
class plesty.lib.device.composite_device.ConnectionState

Bases: str, enum.Enum

Connection state of one sub-device, as seen by its composite.

Initialize self. See help(type(self)) for accurate signature.

CONNECTING = 'connecting'

First connect in progress.

CONNECTED = 'connected'

Client open and answering.

DEGRADED = 'degraded'

A call timed out; the client is being rebuilt.

DISCONNECTED = 'disconnected'

connect or reconnect failed, or the composite shut down.

Type:

No usable client

plesty.lib.device.composite_device.StatusCallback
plesty.lib.device.composite_device._BAD_STATES
exception plesty.lib.device.composite_device.DeviceUnreachableError(dev: str, address: str, elapsed_s: float, cause: str = '')

Bases: RuntimeError

A sub-device server did not answer within its connect deadline.

Variables:

  • dev – Sub-device name.

  • address – Server address that was tried.

  • elapsed_s – Seconds spent trying.

Parameters:

  • dev (str)

  • address (str)

  • elapsed_s (float)

  • cause (str)

Build the error message from its parts.

plesty.lib.device.composite_device.address_env_var(name: str, prefix: str | None = None) → str

Return the environment variable that holds sub-device name’s address.

Server addresses are credentials and live in the environment (an .env file), never in a checked-in config file. The variable name follows the convention already used across the platform — HWP_ADDRESS, PL_SCAN_CAM_ADDRESS — so a composite can find a sub-device’s address from its own attribute name.

Parameters:

  • name (str) – Sub-device attribute name (e.g. "hwp", "cam").

  • prefix (str | None) – Optional deployment prefix (e.g. "PL_SCAN").

Returns:

The upper-case variable name, e.g. PL_SCAN_CAM_ADDRESS.

Return type:

str

plesty.lib.device.composite_device.connect_client(address: str, timeout_ms: int = _DEFAULT_CLIENT_TIMEOUT_MS, attempts: int = _DEFAULT_CONNECT_ATTEMPTS, backoff_s: float = _DEFAULT_BACKOFF_S, deadline_s: float | None = None) → plesty.lib.service._DeviceTCPIPClient

Build a remote device client, retrying a transient connection handshake.

When several clients connect back-to-back the server occasionally does not answer the connect handshake in time; a plain build_client() would then abort the whole startup. This retries with a linear backoff, logging every attempt — a stalled connect must be visible while it stalls, not only once it has given up.

Parameters:

  • address (str) – ZMQ address of the device server.

  • timeout_ms (int) – Request timeout for the client, in milliseconds.

  • attempts (int) – Connection attempts before giving up.

  • backoff_s (float) – Base backoff; attempt i waits backoff_s * (i + 1) s.

  • deadline_s (float | None) – Total budget in seconds; when given, no attempt starts past it and each handshake is capped to what is left of it (the client’s request timeout is restored to timeout_ms on success).

Returns:

The connected client.

Raises:

RuntimeError – If every attempt fails or the deadline runs out.

Return type:

plesty.lib.service._DeviceTCPIPClient

plesty.lib.device.composite_device._describe_timeout(client: Any, per_call: Any) → str

Return the timeout a call ran under, for the retry warning.

Parameters:

  • client (Any)

  • per_call (Any)

Return type:

str

class plesty.lib.device.composite_device.CompositeDevice(devices: collections.abc.Mapping[str, Any] | collections.abc.Iterable[str], env: plesty.lib.utils.settings.EnvSettings | None = None, env_prefix: str | None = None, thread_affinity: bool = False)

A container device that manages multiple sub-devices as a single unit.

Sub-devices are reached by name (e.g. self.cam). For remote (_DeviceTCPIPClient) sub-devices whose address is known, the composite adds a resilient call() wrapper that retries on a request timeout, rebuilding the client between attempts to flush the stale ZMQ reply left in the socket — so a transient network hiccup does not abort a long-running orchestration.

Initialize from device instances, declarative configs, or bare names.

Each sub-device is given in one of four forms:

Form

Meaning

Device instance

A BaseDeviceSyncModel or a connected _DeviceTCPIPClient, used as-is.

str

The ZMQ address of its device server; the composite opens the client.

Mapping

A configuration (keys below).

None

An all-default configuration.

A bare name in a sequence is the same as None. Configuration keys: address (explicit ZMQ address), env (environment variable to read it from, overriding the derived name), and timeout_ms, attempts, backoff_s, passed to connect_client().

Real addresses stay in .env: unless one is given explicitly, the variable is derived from the sub-device’s own attribute name via address_env_var() ("hwp" → HWP_ADDRESS):

CompositeDevice(["hwp", "spec", "pm"])          # HWP_ADDRESS, ...
CompositeDevice(["cam", "led"], env_prefix="PL_SCAN")  # PL_SCAN_CAM_ADDRESS, ...
CompositeDevice({"hwp": {"timeout_ms": 5_000}, "stage": stage_device})

Every sub-device is reconnectable by call() / reconnect() without its address being repeated: a configured one registers the address it resolved, and a client passed as an instance reports the one it was opened with.

With thread_affinity, each sub-device gets its own single-thread executor, and its client is opened and used only on that thread — ZMQ sockets are not thread-safe, and even migrating one between threads is fragile. submit() then runs calls on the owning thread, which is also what lets several sub-devices act at the same time.

Parameters:

  • devices (collections.abc.Mapping[str, Any] | collections.abc.Iterable[str]) – Mapping of sub-device name to an instance, an address string, a config mapping, or None; or a sequence of names to configure entirely from the environment.

  • env (plesty.lib.utils.settings.EnvSettings | None) – Environment source for address lookups. Defaults to a side-effect-free EnvSettings.load() (.env plus the process environment), loaded only if an address is needed.

  • env_prefix (str | None) – Deployment prefix for derived variable names, e.g. "PL_SCAN" for PL_SCAN_CAM_ADDRESS.

  • thread_affinity (bool) – Give each sub-device its own thread and keep its client on it. Required for a composite whose sub-devices are driven concurrently.

Raises:

ValueError – If a config carries unknown keys, no address can be resolved for a sub-device, or a value is none of the four forms.

devices: list[str] = []
_addresses: dict[str, str]
_connect_options: dict[str, dict[str, Any]]
_requires: dict[str, tuple[str, Ellipsis]]
_released: set[str]
_states: dict[str, ConnectionState]
_status_callbacks: list[StatusCallback] = []
_env = None
_env_prefix = None
_executors: dict[str, concurrent.futures.ThreadPoolExecutor]
_owner
_connect_options_from_config(name: str, config: collections.abc.Mapping[str, Any]) → tuple[str, dict[str, Any]]

Resolve a declarative sub-device config into an address and connect options.

Parameters:

  • name (str) – Sub-device attribute name.

  • config (collections.abc.Mapping[str, Any]) – Declarative configuration (see __init__()).

Returns:

The server address and the keyword options for connect_client() (plus connect_deadline_s for the first connect).

Raises:

ValueError – If config carries unknown keys or no address resolves.

Return type:

tuple[str, dict[str, Any]]

_first_connect(name: str, address: str, options: collections.abc.Mapping[str, Any]) → plesty.lib.service._DeviceTCPIPClient

Open sub-device name’s client within its connect deadline.

Parameters:

  • name (str) – Sub-device name.

  • address (str) – Server address.

  • options (collections.abc.Mapping[str, Any]) – Options from _connect_options_from_config().

Returns:

The connected client.

Raises:

DeviceUnreachableError – If the server does not answer within connect_deadline_s.

Return type:

plesty.lib.service._DeviceTCPIPClient

preflight() → plesty.lib.traffic.preflight.Preflight

Check that every sub-device is the device it is supposed to be, and answers.

For each remote sub-device: the server must expose the methods listed under requires in its configuration (a wrong <NAME>_ADDRESS — say the stage’s port in PM_ADDRESS — connects fine and fails only at the first row without this), and identity() must answer. Local sub-devices are checked for the required attributes only; they connect in the experiment’s setup.

A server that answers is then asked what its own transport sees, and that report is kept as a part of this one. It is the difference between “the stage does not answer” and “the stage’s server is up, and the controller it drives is on the USB bus with no driver bound” — two problems with two different fixes, on two different machines. A server too old to expose preflight simply contributes no part; nothing about the rig’s own verdict changes.

Returns:

One Preflight for the rig, with a part per sub-device. .ready is the whole rig, .problems() is the name-to-problem mapping this used to return, and .summary() prints the rig and everything under it. Every result is logged.

Return type:

plesty.lib.traffic.preflight.Preflight

_remote_preflight(name: str, address: str, required: collections.abc.Iterable[str]) → plesty.lib.traffic.preflight.Preflight

Return what a device server says about itself, and about its instrument.

Parameters:

  • name (str) – The sub-device name in this composite.

  • address (str) – The server address it is reached at.

  • required (collections.abc.Iterable[str]) – The methods its configuration says it must expose.

Return type:

plesty.lib.traffic.preflight.Preflight

_instrument_part(name: str, exposed: set[str]) → tuple[plesty.lib.traffic.preflight.Preflight, Ellipsis]

Return the server’s own transport report, when it can give one.

Best effort by design: a server that predates this, or one whose transport cannot enumerate, leaves the sub-device reporting only what this side can see. Failing the rig over a diagnostic that is missing would make an upgrade of this library break every server not yet upgraded with it.

Parameters:

  • name (str) – The sub-device to ask.

  • exposed (set[str]) – The methods its describe reported.

Return type:

tuple[plesty.lib.traffic.preflight.Preflight, Ellipsis]

_local_preflight(name: str, device: Any, required: collections.abc.Iterable[str]) → plesty.lib.traffic.preflight.Preflight

Return what can be established about a sub-device in this process.

Parameters:

  • name (str) – The sub-device name in this composite.

  • device (Any) – The object itself.

  • required (collections.abc.Iterable[str]) – The methods its configuration says it must have.

Return type:

plesty.lib.traffic.preflight.Preflight

status(dev: str | None = None) → ConnectionState | dict[str, ConnectionState]

Return the connection state of one sub-device, or of all of them.

Parameters:

dev (str | None) – Sub-device name; None returns a name → state mapping.

Returns:

The ConnectionState of dev, or a copy of the whole map.

Raises:

KeyError – If dev is not a sub-device of this composite.

Return type:

ConnectionState | dict[str, ConnectionState]

on_status_change(callback: StatusCallback) → StatusCallback

Subscribe to sub-device connection state changes.

The callback runs synchronously on the thread that changes the state (a sub-device’s own thread under thread affinity) with (dev, old_state, new_state, detail); old_state is None for the first transition. Exceptions raised by a subscriber are logged and swallowed — a broken UI hook must not break the rig. The composite’s own log lines are emitted for every change independently of subscribers. Usable as a decorator.

Parameters:

callback (StatusCallback) – Subscriber.

Returns:

The same callback.

Return type:

StatusCallback

_set_state(dev: str, new: ConnectionState, detail: str = '', expected: bool = False) → None

Record a state transition, log it, and notify subscribers (if it changed).

Parameters:

  • dev (str) – Sub-device name.

  • new (ConnectionState) – The state entered.

  • detail (str) – Free text appended to the log line and passed to subscribers.

  • expected (bool) – The transition was requested (disconnect_all), so a bad state is announced at INFO instead of WARNING.

Return type:

None

_resolve_address(name: str, env_var: str | None = None) → str

Read sub-device name’s address from the environment.

Parameters:

  • name (str) – Sub-device attribute name, mapped to a variable by address_env_var() unless env_var names one explicitly.

  • env_var (str | None) – Explicit environment variable to read.

Returns:

The address string.

Raises:

ValueError – If the variable is unset or empty.

Return type:

str

call(dev: str, func: str, *args: Any, retries: int = 3, sleep: float = 0.0, reconnect_wait_s: float = 2.0, **kwargs: Any) → Any

Call a sub-device method, retrying and reconnecting on a timeout.

On a TimeoutError the sub-device client is rebuilt (discarding the stale ZMQ reply left in the socket) before the next attempt; the final timeout propagates if every attempt fails.

Parameters:

  • dev (str) – Sub-device name.

  • func (str) – Remote method name to invoke.

  • *args (Any) – Positional arguments forwarded to the method.

  • retries (int) – Attempts before the last TimeoutError propagates.

  • sleep (float) – Optional settle time (s) after a successful call.

  • reconnect_wait_s (float) – Wait (s) after a reconnect before retrying.

  • **kwargs (Any) – Keyword arguments forwarded to the method.

Returns:

The method’s return value.

Raises:

TimeoutError – If every attempt times out.

Return type:

Any

submit(dev: str, func: str, *args: Any, **kwargs: Any) → Future[Any]

Run call() on the sub-device’s own thread and return its future.

This is how a composite does two things at once: submit to each sub-device, then collect the futures — the calls overlap, and each socket is still only ever touched by its owning thread. Without thread_affinity the call runs inline and the returned future is already resolved, so the same code stays correct, just sequential.

Parameters:

  • dev (str) – Sub-device name.

  • func (str) – Remote method name to invoke.

  • *args (Any) – Positional arguments forwarded to call().

  • **kwargs (Any) – Keyword arguments forwarded to call().

Returns:

A future resolving to the method’s return value.

Return type:

Future[Any]

_run_owned(dev: str, func: str, *args: Any, **kwargs: Any) → Any

Mark this thread as the sub-device’s owner and run the call on it.

Parameters:

  • dev (str)

  • func (str)

  • args (Any)

  • kwargs (Any)

Return type:

Any

reconnect(dev: str) → None

Rebuild a remote sub-device client, discarding its stale socket.

Parameters:

dev (str) – Sub-device name.

Raises:

RuntimeError – If no address is known for dev (cannot rebuild).

Return type:

None

identity() → dict[str, Any]

Query the identity of each sub-device and return a mapping of results.

Routed through submit() like every other sub-device call, not called on the client directly. Under thread_affinity a sub-device’s socket belongs to its own thread, and asking here from a foreign one put a second thread on it: a magneto-PL rig whose power regulator was sampling the meter got the identity string back from measure_power (2026-09-01). The crash was the lucky case — two calls of the same shape crossing swap two readings and say nothing.

Return type:

dict[str, Any]

set_data_path(path: str, subdirs: bool = False, devices: collections.abc.Iterable[str] | None = None) → dict[str, str]

Route sub-device acquisitions into path.

An experiment uses this to point a whole instrument set at one session directory. Failures are not swallowed: a sub-device that rejects the path aborts the call, so a run never proceeds believing its data is routed. That is also why devices exists — in a rig where only the camera writes data, naming it is honest, while routing the stage and the powermeter would either fail or silently mean nothing.

Parameters:

  • path (str) – Target directory relative to each sub-device’s data root, forwarded to its set_data_path.

  • subdirs (bool) – When True, give each sub-device its own <path>/<name> subdirectory instead of the shared directory.

  • devices (collections.abc.Iterable[str] | None) – Sub-devices to route; None routes all of them.

Returns:

Mapping of sub-device name to the directory now in effect, relative to the Plesty share — sub-devices on different benches land in different folders of it.

Raises:

KeyError – If devices names a sub-device this composite has not got.

Return type:

dict[str, str]

_close_client(name: str, client: Any) → None

Close a remote client on its owning thread, quickly, never raising.

The server-side disconnect gets one short window (a dead server must not stall the teardown); undelivered messages are dropped (linger 0), so neither close() nor a later garbage collection of the client’s ZMQ context can block.

Parameters:

  • name (str)

  • client (Any)

Return type:

None

release() → None

Close every remote client and stop the threads; the composite is done.

For a composite that will not be used again — a rig that failed its preflight and is rebuilt, or one being torn down. Local sub-devices are not touched (disconnect_all handles those). Idempotent.

Return type:

None

connect_all() → None

Connect all sub-devices that are BaseDeviceSyncModel instances.

Return type:

None

disconnect_all() → None

Disconnect local sub-devices, close remote clients, stop the threads.

Remote clients are closed here too (server-side disconnect, linger 0): a client left open is garbage-collected later, and its ZMQ context can then block the interpreter — during a run, or at exit.

Return type:

None

shutdown() → None

Stop the per-device threads, if this composite owns any.

Idempotent, and harmless without thread_affinity — a composite that never started a thread has none to stop.

Return type:

None

__enter__()

Connect all sub-devices and return self.

__exit__(exc_type, exc_val, exc_tb)

Disconnect all sub-devices on context exit.