plesty.lib.service ================== .. py:module:: plesty.lib.service .. autoapi-nested-parse:: Service package providing TCP/IP server and client factory functions. Submodules ---------- .. toctree:: :maxdepth: 1 /reference/plesty/lib/service/resource_manager/index /reference/plesty/lib/service/tcp_ip_client/index /reference/plesty/lib/service/tcp_ip_server/index Classes ------- .. autoapisummary:: plesty.lib.service._DeviceTCPIPClient plesty.lib.service._DeviceTCPIPServer plesty.lib.service._AsyncWrapperSafe plesty.lib.service._AsyncDeviceThread Functions --------- .. autoapisummary:: plesty.lib.service.build_server plesty.lib.service.build_client Package Contents ---------------- .. py:class:: _DeviceTCPIPClient(address='tcp://localhost:5555', timeout=5000, resources=None, client_id: str | None = None, connect_kwargs: dict | None = None) ZeroMQ-based client for remote device control. This client provides a transparent interface to a remote device server. Methods can be invoked as if the device were local. The server address it was opened with stays available as :attr:`address`. Features: - Standard query/write interface - Arbitrary remote function calls - Automatic method discovery and binding - Timeout support - Structured error handling Initialize the device client. :param address: Server address. :param timeout: Receive timeout in milliseconds. :param resources: Resource request as a string, flat collection, or nested dict tree. :param client_id: Optional stable client identity for server-side routing. :param connect_kwargs: Optional metadata sent during connect handshake. .. py:attribute:: ctx .. py:attribute:: socket .. py:attribute:: client_id :value: 'client-00000000000000000000000000000000' .. py:attribute:: address :value: 'tcp://localhost:5555' .. py:attribute:: resources .. py:attribute:: connect_kwargs .. py:attribute:: default_timeout :value: 5000 .. py:attribute:: response_grace_ms :value: 1000 .. py:attribute:: _connected :value: False .. py:attribute:: connection_info :value: None .. py:method:: set_default_timeout(timeout_ms: int) -> None Change the receive timeout used by requests without their own. :param timeout_ms: New default receive timeout in milliseconds. .. py:method:: connect() -> None Connect to the server, discover methods, and bind them locally. .. py:method:: _close_socket() -> None Close the socket and context without waiting for undelivered messages. The handshake to an absent server sits in the DEALER's queue forever; with the default infinite LINGER, ``ctx.term()`` then blocks forever too — a connect to a dead server never returned, and no deadline could bound it. Linger 0 drops what could not be delivered. .. py:method:: _send(payload) Send a request to the server and wait for response. :param payload: Dictionary payload. :returns: Result from server. :rtype: Any :raises RuntimeError: If server returns an error. :raises TimeoutError: If no response is received. .. py:method:: connect_to_server() -> Any Send a handshake so the server can register client metadata. .. py:method:: disconnect_from_server() -> Any Release server-side allocations for this client. .. py:method:: _resolve_recv_timeout_ms(payload_timeout_s) Convert per-request operation timeout (seconds) into socket receive timeout (ms). A small grace window is added so the client still receives timeout/error responses sent by the server right after operation timeout is reached. .. py:method:: query(param: Any, timeout: Any = None) -> Any Query a device parameter. :param param: Parameter name. :param timeout: Optional timeout in seconds (server-side). :returns: Parameter value. :rtype: Any .. py:method:: write(param: Any, value: Any, timeout: Any = None) -> Any Write a device parameter. :param param: Parameter name. :param value: Value to set. :param timeout: Optional timeout in seconds (server-side). :returns: Result from device. :rtype: Any .. py:method:: call(func: Any, *args: Any, timeout: Any = None, **kwargs: Any) -> Any Call a remote device function. :param func: Function name. :param args: Positional arguments. :param timeout: Optional timeout in seconds (server-side). :param kwargs: Keyword arguments. :returns: Function result. :rtype: Any .. py:method:: describe() -> dict[str, Any] Retrieve available device methods from server. :returns: Device description. :rtype: dict .. py:method:: close() -> None Release server-side allocations and close local ZMQ resources. .. py:method:: __enter__() Enter the client context manager. .. py:method:: __exit__(exc_type, exc_value, traceback) Exit the client context manager, closing the connection. .. py:method:: _build_methods() Dynamically attach remote methods as local methods. .. py:method:: _make_proxy(name) Create a proxy method for a remote function. :param name: Function name. :returns: Proxy method. :rtype: callable .. py:class:: _DeviceTCPIPServer(device, wrapper_cls, address='tcp://*:5555', resources=None) ZeroMQ-based asynchronous device server. This server exposes a synchronous device over TCP/IP using a JSON-based RPC protocol. The device is wrapped using an async wrapper (e.g., AsyncWrapperSafe or AsyncDeviceThread) to ensure safe, serialized access. The server supports multiple concurrent clients via a ROUTER socket. Protocol: Request payload: .. code-block:: json { "type": "query | write | call | help | describe", "timeout": 3.0 } Response payload: .. code-block:: json {"status": "ok", "result": "..."} Error response payload: .. code-block:: json {"status": "error", "error": "message", "type": "ExceptionType"} Initialize the device server. :param device: Synchronous device instance. :param wrapper_cls: Async wrapper class (e.g., AsyncDeviceThread). :param address: ZMQ bind address. :param resources: Optional flat collection or nested dict tree of lockable resources. .. py:attribute:: resource_manager .. py:attribute:: client_metadata :type: dict[str, dict] .. py:attribute:: device .. py:attribute:: ctx .. py:attribute:: socket .. py:attribute:: host .. py:attribute:: port .. py:property:: address :type: str Get the server's bind address. .. py:property:: is_running :type: bool Check if the server is currently running. .. py:method:: _execute(coro, timeout=None) :async: Execute a coroutine with optional timeout. :param coro: Coroutine to execute. :param timeout: Timeout in seconds. :returns: Result of coroutine. :raises asyncio.TimeoutError: If timeout is exceeded. .. py:method:: _client_id(identity: bytes) -> str :staticmethod: Return a stable string form of a ZeroMQ ROUTER identity frame. .. py:method:: _log_orphan_reply(identity: bytes, request_str: str, exc: Exception) -> None Report a reply that had nowhere to go, at the severity it deserves. Two very different things produce an unroutable reply, and until now they produced the same warning. A ``disconnect`` is routine: the client asks to be released and its composite gives that acknowledgement a one-second window before dropping the socket, so a server that is a moment slow — because it is finishing an acquisition, the loop being sequential — answers a client that has already gone. Nothing is lost, and a teardown per run turns the log into noise nobody reads. Anything else is a lost answer to work the device has already done: an exposure taken, a parameter written, a stage moved. The caller saw a timeout and will decide the operation failed. That deserves a warning, and it deserves to name the operation, because "some reply was lost" is not something anyone can act on afterwards. :param identity: ROUTER identity frame of the vanished client. :param request_str: The request being answered, as received. :param exc: The routing error. .. py:method:: _requested_target(request_str: str) -> str :staticmethod: Return what a request acted on (function or parameter), for the log. .. py:method:: _handle(request_str, identity: bytes) :async: Handle a single client request. :param request_str: JSON-encoded request string. :param identity: ZeroMQ ROUTER identity frame bytes. :returns: JSON-serializable response. :rtype: dict .. py:method:: _encode(response: dict, identity: bytes) -> bytes Serialize a response, turning a failure into that client's error. ``_handle`` catches everything a device call can raise and answers with an error dict — but the encoding happened in :meth:`run`, outside that boundary. A result JSON cannot carry therefore raised in the server loop, hit its outer handler, and shut the server down: one client calling one method that returns an ndarray, a dataclass or a set took the instrument away from every other client, and the failure looked like a timeout because nothing answered. That is finding 3 of the 2026-08-04 round, and every device that returns anything richer than a JSON primitive is one call away from it. A method whose return value cannot cross the protocol is a defect in that method. It is not a reason to end the session. :param response: The response ``_handle`` produced. :param identity: ROUTER identity of the requesting client, for the log. :returns: The encoded response, or an encoded error naming the type that could not be serialized. .. py:method:: _describe() :async: Introspect the device and list available methods. :returns: Available callable methods. :rtype: dict .. py:method:: _help() :async: .. py:method:: run() -> None :async: Run the server loop indefinitely. This method listens for incoming requests and processes them sequentially. Each request is handled asynchronously. .. py:method:: shutdown() -> None :async: Gracefully shutdown the server. .. py:class:: _AsyncWrapperSafe(obj) Bases: :py:obj:`AsyncWrapperBase` Asynchronous wrapper for synchronous device objects using thread offloading. This class converts all callable attributes of a synchronous object into asynchronous methods by executing them in a background thread via `asyncio.to_thread`. A per-instance asyncio lock ensures that only one operation is executed at a time, making it safe for thread-unsafe hardware interfaces. The wrapper is transparent: methods can be called using `await` without modifying the original device implementation. A global weak registry ensures that only one wrapper exists per device instance, preventing accidental concurrent access through multiple wrappers. .. rubric:: Example .. code-block:: python sync_device = Monochromator(...) device = AsyncWrapperSafe(sync_device) await device.goto(532) await device.set_grating(1) Internal State: Uses an internal asyncio lock to serialize access and a class-level weak registry to maintain one wrapper instance per wrapped object. .. rubric:: Notes - All wrapped methods are executed in a thread using `asyncio.to_thread`, which avoids blocking the event loop. - Calls are serialized using an asyncio lock, ensuring safe access to non-thread-safe hardware. - If the wrapped object already provides async methods, they are returned unchanged and not wrapped again. - Non-callable attributes are forwarded directly. .. warning:: - **Serialized is not the same as single-threaded.** The lock keeps calls from overlapping, but ``asyncio.to_thread`` hands each one to the *process-wide* default executor: the device owns no thread, its calls are not pinned to one, and they queue behind every unrelated ``to_thread`` job in the process. A driver that is merely thread-unsafe is protected; a driver with thread **affinity**, or one whose callbacks must run while a call is in flight, is not. - Do not mix synchronous and asynchronous access to the same device instance, as this can lead to race conditions or hardware conflicts. - Creating multiple wrappers for different instances controlling the same physical hardware (e.g., same COM port or TCP endpoint) is not prevented by this class. Use an external resource registry if needed. - Frequent high-rate calls may incur overhead due to thread creation; consider a dedicated worker thread model in such cases. :raises RuntimeError: May propagate exceptions raised by the underlying device methods during execution. When to Use: - When you need a quick, low-overhead way to integrate synchronous device APIs into an asyncio-based application. - When device calls are relatively infrequent and simplicity is preferred over maximum performance. When Not to Use: - For high-frequency polling or streaming applications. - When strict single-thread execution is required — which, measured against real instruments, is most of them. Use :class:`AsyncDeviceThread` (``build_server(..., fixed_threading=True)``, the default). .. note:: Thread affinity is not a theoretical concern. In the 2026-08 field round, migration between pool threads produced two distinct hardware failures: a LightField completion callback (Python, invoked from .NET) was starved by the very pool thread blocked waiting on it, and ZMQ sockets used across threads lost messages under real network latency. Both went away under fixed threading. This class remains for drivers with no affinity requirement and for comparison measurements. Initialize AsyncWrapperSafe, creating a per-instance asyncio lock. .. py:attribute:: _lock .. py:method:: _call(func, *args, **kwargs) :async: Execute a synchronous function asynchronously. Must be implemented by subclasses. .. py:class:: _AsyncDeviceThread(obj) Bases: :py:obj:`AsyncWrapperBase` Asynchronous wrapper for synchronous device objects using a dedicated worker thread. This class provides a transparent async interface for a synchronous device by executing all method calls in a single background thread. Calls are queued and processed sequentially (FIFO), ensuring strict ordering and eliminating concurrent access to the underlying device. Unlike thread-per-call approaches, this design avoids repeated thread creation and is well-suited for high-frequency or continuous device interactions. The wrapper is transparent: callable attributes of the underlying device can be accessed as async methods using `await`. .. rubric:: Example .. code-block:: python sync_device = Wavemeter(...) device = AsyncDeviceThread(sync_device) wavelength = await device.get_wavelength() await device.set_mode("fast") Internal State: Uses a per-instance call queue plus a dedicated worker thread and a class-level weak registry to maintain one wrapper per object. .. rubric:: Notes - All device operations are executed in a single dedicated thread, guaranteeing thread safety for non-thread-safe hardware APIs. - Calls are processed in FIFO order, ensuring deterministic execution. - Results and exceptions are safely communicated back to the asyncio event loop using Futures. - If the wrapped object already provides async methods, they are returned unchanged and not wrapped again. - Non-callable attributes are forwarded directly. .. warning:: - Do not mix synchronous and asynchronous access to the same device instance, as this may lead to race conditions or undefined behavior. - Creating multiple device instances for the same physical hardware (e.g., same COM port or TCP endpoint) is not prevented by this class. Use a resource registry or device server for distributed coordination. - Long-running or blocking device calls will block the worker thread and delay subsequent queued operations. :raises RuntimeError: May propagate exceptions raised by the underlying device methods during execution. :raises asyncio.TimeoutError: If a timeout is specified and exceeded. When to Use: - When working with thread-unsafe hardware (serial, TCP, DLL). - For high-frequency polling or continuous control loops. - When strict ordering and single-thread execution are required. - When performance matters and thread creation overhead must be avoided. When Not to Use: - For simple or infrequent device interactions where a lightweight wrapper is sufficient. - When the underlying API is already fully asynchronous. Initialize AsyncDeviceThread, starting the background worker thread. .. py:attribute:: _registry :type: weakref.WeakKeyDictionary[Any, Any] .. py:attribute:: _queue .. py:attribute:: _thread .. py:method:: _worker() Worker thread: executes device calls sequentially. .. py:method:: _call(func, *args, **kwargs) :async: Schedule a function call in the worker thread and await result. .. py:function:: build_server(device: Any, fixed_threading: bool = True, address: str = 'tcp://*:5555') -> tcp_ip_server.DeviceTCPIPServer Factory function to create a DeviceTCPIPServer instance. :param device: Synchronous device instance. :param fixed_threading: Pin every device call to one dedicated worker thread (:class:`AsyncDeviceThread`). The default since the 2026-08 field round: instrument drivers turned out to have thread *affinity*, not merely thread-unsafety, and migrating between pool threads starved a .NET completion callback and lost ZMQ messages under real latency. Pass ``False`` for :class:`AsyncWrapperSafe`, which serializes calls but runs them on rotating executor threads. :param address: ZMQ bind address. The server reads lockable resources from ``device._resources``, given as a flat collection or a nested dictionary tree:: ["Dev1/port0/line0", "Dev1/port0/line1"] { "Dev1": { "port0": ["line0", "line1"], "port1": ["line0", "line1"], } } :returns: DeviceTCPIPServer instance. .. py:function:: build_client(address: str = 'tcp://localhost:5555', timeout: int = 5000, resources: Any = None, client_id: str | None = None, connect_kwargs: dict | None = None) -> tcp_ip_client.DeviceTCPIPClient Factory function to create a DeviceTCPIPClient instance. :param address: Server address. :param timeout: Receive timeout in milliseconds. :param resources: Optional resources to reserve on connect — a single string (``"Dev1/port0"`` requests every resource under that subtree), a flat collection (``["Dev1/port0/line0", "Dev1/port1/line1"]``), or a nested dictionary tree (``{"Dev1": {"port0": ["line0", "line1"]}}``). :param client_id: Optional stable client identity. :param connect_kwargs: Optional extra metadata sent during connect handshake. :returns: DeviceTCPIPClient instance.