plesty.lib.service

Service package providing TCP/IP server and client factory functions.

Submodules

• plesty.lib.service.resource_manager
• plesty.lib.service.tcp_ip_client
• plesty.lib.service.tcp_ip_server

Classes

_DeviceTCPIPClient	ZeroMQ-based client for remote device control.
_DeviceTCPIPServer	ZeroMQ-based asynchronous device server.
_AsyncWrapperSafe	Asynchronous wrapper for synchronous device objects using thread offloading.
_AsyncDeviceThread	Asynchronous wrapper for synchronous device objects using a dedicated worker thread.

Functions

build_server(→ tcp_ip_server.DeviceTCPIPServer)	Factory function to create a DeviceTCPIPServer instance.
build_client(→ tcp_ip_client.DeviceTCPIPClient)	Factory function to create a DeviceTCPIPClient instance.

Package Contents

class plesty.lib.service._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 address.

Features:

• Standard query/write interface

• Arbitrary remote function calls

• Automatic method discovery and binding

• Timeout support

• Structured error handling

Initialize the device client.

Parameters:

• address – Server address.

• timeout – Receive timeout in milliseconds.

• resources – Resource request as a string, flat collection, or nested dict tree.

• client_id (str | None) – Optional stable client identity for server-side routing.

• connect_kwargs (dict | None) – Optional metadata sent during connect handshake.

ctx

socket

client_id = 'client-00000000000000000000000000000000'

address = 'tcp://localhost:5555'

resources

connect_kwargs

default_timeout = 5000

response_grace_ms = 1000

_connected = False

connection_info = None

set_default_timeout(timeout_ms: int) → None

Change the receive timeout used by requests without their own.

Parameters:

timeout_ms (int) – New default receive timeout in milliseconds.

Return type:

None

connect() → None

Connect to the server, discover methods, and bind them locally.

Return type:

None

_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.

Return type:

None

_send(payload)

Send a request to the server and wait for response.

Parameters:

payload – Dictionary payload.

Returns:

Result from server.

Return type:

Any

Raises:

• RuntimeError – If server returns an error.

• TimeoutError – If no response is received.

connect_to_server() → Any

Send a handshake so the server can register client metadata.

Return type:

Any

disconnect_from_server() → Any

Release server-side allocations for this client.

Return type:

Any

_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.

query(param: Any, timeout: Any = None) → Any

Query a device parameter.

Parameters:

• param (Any) – Parameter name.

• timeout (Any) – Optional timeout in seconds (server-side).

Returns:

Parameter value.

Return type:

Any

write(param: Any, value: Any, timeout: Any = None) → Any

Write a device parameter.

Parameters:

• param (Any) – Parameter name.

• value (Any) – Value to set.

• timeout (Any) – Optional timeout in seconds (server-side).

Returns:

Result from device.

Return type:

Any

call(func: Any, *args: Any, timeout: Any = None, **kwargs: Any) → Any

Call a remote device function.

Parameters:

• func (Any) – Function name.

• args (Any) – Positional arguments.

• timeout (Any) – Optional timeout in seconds (server-side).

• kwargs (Any) – Keyword arguments.

Returns:

Function result.

Return type:

Any

describe() → dict[str, Any]

Retrieve available device methods from server.

Returns:

Device description.

Return type:

dict

close() → None

Release server-side allocations and close local ZMQ resources.

Return type:

None

__enter__()

Enter the client context manager.

__exit__(exc_type, exc_value, traceback)

Exit the client context manager, closing the connection.

_build_methods()

Dynamically attach remote methods as local methods.

_make_proxy(name)

Create a proxy method for a remote function.

Parameters:

name – Function name.

Returns:

Proxy method.

Return type:

callable

class plesty.lib.service._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:

{
  "type": "query | write | call | help | describe",
  "timeout": 3.0
}

Response payload:

{"status": "ok", "result": "..."}

Error response payload:

{"status": "error", "error": "message", "type": "ExceptionType"}

Initialize the device server.

Parameters:

• device – Synchronous device instance.

• wrapper_cls – Async wrapper class (e.g., AsyncDeviceThread).

• address – ZMQ bind address.

• resources – Optional flat collection or nested dict tree of lockable resources.

resource_manager

client_metadata: dict[str, dict]

device

ctx

socket

host

port

property address: str

Get the server’s bind address.

Return type:

str

property is_running: bool

Check if the server is currently running.

Return type:

bool

async _execute(coro, timeout=None)

Execute a coroutine with optional timeout.

Parameters:

• coro – Coroutine to execute.

• timeout – Timeout in seconds.

Returns:

Result of coroutine.

Raises:

asyncio.TimeoutError – If timeout is exceeded.

static _client_id(identity: bytes) → str

Return a stable string form of a ZeroMQ ROUTER identity frame.

Parameters:

identity (bytes)

Return type:

str

_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.

Parameters:

• identity (bytes) – ROUTER identity frame of the vanished client.

• request_str (str) – The request being answered, as received.

• exc (Exception) – The routing error.

Return type:

None

static _requested_target(request_str: str) → str

Return what a request acted on (function or parameter), for the log.

Parameters:

request_str (str)

Return type:

str

async _handle(request_str, identity: bytes)

Handle a single client request.

Parameters:

• request_str – JSON-encoded request string.

• identity (bytes) – ZeroMQ ROUTER identity frame bytes.

Returns:

JSON-serializable response.

Return type:

dict

_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 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.

Parameters:

• response (dict) – The response _handle produced.

• identity (bytes) – 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.

Return type:

bytes

async _describe()

Introspect the device and list available methods.

Returns:

Available callable methods.

Return type:

dict

async _help()

async run() → None

Run the server loop indefinitely.

This method listens for incoming requests and processes them sequentially. Each request is handled asynchronously.

Return type:

None

async shutdown() → None

Gracefully shutdown the server.

Return type:

None

class plesty.lib.service._AsyncWrapperSafe(obj)

Bases: 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.

Example

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.

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 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.

_lock

async _call(func, *args, **kwargs)

Execute a synchronous function asynchronously. Must be implemented by subclasses.

class plesty.lib.service._AsyncDeviceThread(obj)

Bases: 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.

Example

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.

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.

• 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.

_registry: weakref.WeakKeyDictionary[Any, Any]

_queue

_thread

_worker()

Worker thread: executes device calls sequentially.

async _call(func, *args, **kwargs)

Schedule a function call in the worker thread and await result.

plesty.lib.service.build_server(device: Any, fixed_threading: bool = True, address: str = 'tcp://*:5555') → tcp_ip_server.DeviceTCPIPServer

Factory function to create a DeviceTCPIPServer instance.

Parameters:

• device (Any) – Synchronous device instance.

• fixed_threading (bool) – Pin every device call to one dedicated worker thread (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 AsyncWrapperSafe, which serializes calls but runs them on rotating executor threads.

• address (str) – ZMQ bind address.

Return type:

tcp_ip_server.DeviceTCPIPServer

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.

Parameters:

• device (Any)

• fixed_threading (bool)

• address (str)

Return type:

tcp_ip_server.DeviceTCPIPServer

plesty.lib.service.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.

Parameters:

• address (str) – Server address.

• timeout (int) – Receive timeout in milliseconds.

• resources (Any) – 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"]}}).

• client_id (str | None) – Optional stable client identity.

• connect_kwargs (dict | None) – Optional extra metadata sent during connect handshake.

Returns:

DeviceTCPIPClient instance.

Return type:

tcp_ip_client.DeviceTCPIPClient