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() → dict[str, str]

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.

Returns:: Sub-device name → one-line problem, for the sub-devices that failed; empty when the rig is ready. Every result is logged.
Return type:: dict[str, str]

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.

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, forwarded to each sub-device’s set_data_path (relative paths resolve per sub-device).
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 resolved path now in effect.

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.