plesty.lib.device.composite_device ================================== .. py:module:: plesty.lib.device.composite_device .. autoapi-nested-parse:: Composite device wrapper for managing multiple devices together. Attributes ---------- .. autoapisummary:: plesty.lib.device.composite_device._logger plesty.lib.device.composite_device._DEFAULT_CLIENT_TIMEOUT_MS plesty.lib.device.composite_device._DEFAULT_CONNECT_ATTEMPTS plesty.lib.device.composite_device._DEFAULT_BACKOFF_S plesty.lib.device.composite_device._DEFAULT_CONNECT_DEADLINE_S plesty.lib.device.composite_device._MIN_HANDSHAKE_MS plesty.lib.device.composite_device._CLOSE_TIMEOUT_MS plesty.lib.device.composite_device._ADDRESS_ENV_SUFFIX plesty.lib.device.composite_device._CONFIG_KEYS plesty.lib.device.composite_device.StatusCallback plesty.lib.device.composite_device._BAD_STATES Exceptions ---------- .. autoapisummary:: plesty.lib.device.composite_device.DeviceUnreachableError Classes ------- .. autoapisummary:: plesty.lib.device.composite_device.ConnectionState plesty.lib.device.composite_device.CompositeDevice Functions --------- .. autoapisummary:: plesty.lib.device.composite_device.address_env_var plesty.lib.device.composite_device.connect_client plesty.lib.device.composite_device._describe_timeout Module Contents --------------- .. py:data:: _logger .. py:data:: _DEFAULT_CLIENT_TIMEOUT_MS :value: 50000 .. py:data:: _DEFAULT_CONNECT_ATTEMPTS :value: 4 .. py:data:: _DEFAULT_BACKOFF_S :value: 2.0 .. py:data:: _DEFAULT_CONNECT_DEADLINE_S :value: 10.0 .. py:data:: _MIN_HANDSHAKE_MS :value: 250 .. py:data:: _CLOSE_TIMEOUT_MS :value: 1000 .. py:data:: _ADDRESS_ENV_SUFFIX :value: 'ADDRESS' .. py:data:: _CONFIG_KEYS .. py:class:: ConnectionState Bases: :py:obj:`str`, :py:obj:`enum.Enum` Connection state of one sub-device, as seen by its composite. Initialize self. See help(type(self)) for accurate signature. .. py:attribute:: CONNECTING :value: 'connecting' First connect in progress. .. py:attribute:: CONNECTED :value: 'connected' Client open and answering. .. py:attribute:: DEGRADED :value: 'degraded' A call timed out; the client is being rebuilt. .. py:attribute:: DISCONNECTED :value: 'disconnected' connect or reconnect failed, or the composite shut down. :type: No usable client .. py:data:: StatusCallback .. py:data:: _BAD_STATES .. py:exception:: DeviceUnreachableError(dev: str, address: str, elapsed_s: float, cause: str = '') Bases: :py:obj:`RuntimeError` A sub-device server did not answer within its connect deadline. :ivar dev: Sub-device name. :ivar address: Server address that was tried. :ivar elapsed_s: Seconds spent trying. Build the error message from its parts. .. py:function:: 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. :param name: Sub-device attribute name (e.g. ``"hwp"``, ``"cam"``). :param prefix: Optional deployment prefix (e.g. ``"PL_SCAN"``). :returns: The upper-case variable name, e.g. ``PL_SCAN_CAM_ADDRESS``. .. py:function:: 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 :func:`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. :param address: ZMQ address of the device server. :param timeout_ms: Request timeout for the client, in milliseconds. :param attempts: Connection attempts before giving up. :param backoff_s: Base backoff; attempt *i* waits ``backoff_s * (i + 1)`` s. :param deadline_s: 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. .. py:function:: _describe_timeout(client: Any, per_call: Any) -> str Return the timeout a call ran under, for the retry warning. .. py:class:: 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 (:class:`_DeviceTCPIPClient`) sub-devices whose address is known, the composite adds a resilient :meth:`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 :class:`BaseDeviceSyncModel` or a connected :class:`_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 :func:`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 :func:`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 :meth:`call` / :meth:`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. :meth:`submit` then runs calls on the owning thread, which is also what lets several sub-devices act at the same time. :param devices: 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. :param env: Environment source for address lookups. Defaults to a side-effect-free :meth:`EnvSettings.load` (``.env`` plus the process environment), loaded only if an address is needed. :param env_prefix: Deployment prefix for derived variable names, e.g. ``"PL_SCAN"`` for ``PL_SCAN_CAM_ADDRESS``. :param thread_affinity: 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. .. py:attribute:: devices :type: list[str] :value: [] .. py:attribute:: _addresses :type: dict[str, str] .. py:attribute:: _connect_options :type: dict[str, dict[str, Any]] .. py:attribute:: _requires :type: dict[str, tuple[str, Ellipsis]] .. py:attribute:: _released :type: set[str] .. py:attribute:: _states :type: dict[str, ConnectionState] .. py:attribute:: _status_callbacks :type: list[StatusCallback] :value: [] .. py:attribute:: _env :value: None .. py:attribute:: _env_prefix :value: None .. py:attribute:: _executors :type: dict[str, concurrent.futures.ThreadPoolExecutor] .. py:attribute:: _owner .. py:method:: _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. :param name: Sub-device attribute name. :param config: Declarative configuration (see :meth:`__init__`). :returns: The server address and the keyword options for :func:`connect_client` (plus ``connect_deadline_s`` for the first connect). :raises ValueError: If *config* carries unknown keys or no address resolves. .. py:method:: _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. :param name: Sub-device name. :param address: Server address. :param options: Options from :meth:`_connect_options_from_config`. :returns: The connected client. :raises DeviceUnreachableError: If the server does not answer within ``connect_deadline_s``. .. py:method:: 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 ``_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. .. py:method:: status(dev: str | None = None) -> ConnectionState | dict[str, ConnectionState] Return the connection state of one sub-device, or of all of them. :param dev: Sub-device name; ``None`` returns a name → state mapping. :returns: The :class:`ConnectionState` of *dev*, or a copy of the whole map. :raises KeyError: If *dev* is not a sub-device of this composite. .. py:method:: 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. :param callback: Subscriber. :returns: The same *callback*. .. py:method:: _set_state(dev: str, new: ConnectionState, detail: str = '', expected: bool = False) -> None Record a state transition, log it, and notify subscribers (if it changed). :param dev: Sub-device name. :param new: The state entered. :param detail: Free text appended to the log line and passed to subscribers. :param expected: The transition was requested (``disconnect_all``), so a bad state is announced at INFO instead of WARNING. .. py:method:: _resolve_address(name: str, env_var: str | None = None) -> str Read sub-device *name*'s address from the environment. :param name: Sub-device attribute name, mapped to a variable by :func:`address_env_var` unless *env_var* names one explicitly. :param env_var: Explicit environment variable to read. :returns: The address string. :raises ValueError: If the variable is unset or empty. .. py:method:: 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 :class:`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. :param dev: Sub-device name. :param func: Remote method name to invoke. :param \*args: Positional arguments forwarded to the method. :param retries: Attempts before the last :class:`TimeoutError` propagates. :param sleep: Optional settle time (s) after a successful call. :param reconnect_wait_s: Wait (s) after a reconnect before retrying. :param \*\*kwargs: Keyword arguments forwarded to the method. :returns: The method's return value. :raises TimeoutError: If every attempt times out. .. py:method:: submit(dev: str, func: str, *args: Any, **kwargs: Any) -> Future[Any] Run :meth:`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. :param dev: Sub-device name. :param func: Remote method name to invoke. :param \*args: Positional arguments forwarded to :meth:`call`. :param \*\*kwargs: Keyword arguments forwarded to :meth:`call`. :returns: A future resolving to the method's return value. .. py:method:: _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. .. py:method:: reconnect(dev: str) -> None Rebuild a remote sub-device client, discarding its stale socket. :param dev: Sub-device name. :raises RuntimeError: If no address is known for *dev* (cannot rebuild). .. py:method:: identity() -> dict[str, Any] Query the identity of each sub-device and return a mapping of results. .. py:method:: 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. :param path: Target directory, forwarded to each sub-device's ``set_data_path`` (relative paths resolve per sub-device). :param subdirs: When ``True``, give each sub-device its own ``/`` subdirectory instead of the shared directory. :param devices: 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. .. py:method:: _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. .. py:method:: 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. .. py:method:: connect_all() -> None Connect all sub-devices that are BaseDeviceSyncModel instances. .. py:method:: 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. .. py:method:: 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. .. py:method:: __enter__() Connect all sub-devices and return self. .. py:method:: __exit__(exc_type, exc_val, exc_tb) Disconnect all sub-devices on context exit.