plesty.lib.service.tcp_ip_server

ZeroMQ-based asynchronous TCP/IP device server for Plesty.

Classes

DeviceTCPIPServer

ZeroMQ-based asynchronous device server.

Module Contents

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