plesty.lib.service.tcp_ip_server

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

Attributes

`TRAFFIC_LOGGER`
`TRAFFIC_MARKER`
`LIVENESS_TYPES`

Exceptions

_BudgetExceeded

The server's own time budget for one request ran out.

Classes

DeviceTCPIPServer

ZeroMQ-based asynchronous device server.

Functions

stop_on_break(→ collections.abc.Iterator[None])

Deliver a Windows CTRL_BREAK_EVENT as KeyboardInterrupt, as Ctrl+C is.

Module Contents

plesty.lib.service.tcp_ip_server.TRAFFIC_LOGGER = 'plesty.server.traffic'

plesty.lib.service.tcp_ip_server.TRAFFIC_MARKER = '[TRAFFIC]'

plesty.lib.service.tcp_ip_server.LIVENESS_TYPES

exception plesty.lib.service.tcp_ip_server._BudgetExceeded

Bases: Exception

The server’s own time budget for one request ran out.

Kept apart from TimeoutError so a device that raises one of its own (“axis 2 did not reach target within 15 s”) is reported with its message, not as the server giving up.

Initialize self. See help(type(self)) for accurate signature.

plesty.lib.service.tcp_ip_server.stop_on_break() → collections.abc.Iterator[None]

Deliver a Windows CTRL_BREAK_EVENT as KeyboardInterrupt, as Ctrl+C is.

CTRL_BREAK_EVENT is the only console control event that can be aimed at a single process group, so it is what a supervisor sends to stop one server without touching its neighbours. Python delivers it as SIGBREAK, whose default action on Windows is to terminate the process outright (exit 0xC000013A): the finally that closes the device never runs, and whatever the instrument holds — a vendor subprocess, a camera, a serial handle — is leaked, blocking the next start.

Inside this context SIGBREAK is handled by the interpreter’s own SIGINT handler, so a stop raises KeyboardInterrupt in the main thread and takes the ordinary shutdown path. The previous handler is put back on the way out.

Does nothing off Windows, and nothing outside the main thread of the main interpreter, where signal handlers cannot be installed at all.

Return type:: collections.abc.Iterator[None]

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

_send_lock

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:

_BudgetExceeded – If timeout runs out. A TimeoutError the device raises itself propagates unchanged, message and all.

static _client_id(identity: bytes) → str

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

Parameters:: identity (bytes)
Return type:: str

_log_traffic(request_str: str, identity: bytes, response: dict, elapsed: float, delivered: bool) → None

Write one record for a served request, for a monitor to read.

What was asked, of what, by whom, how it ended and how long it took — the metadata of the exchange and none of its payload. A result can be a spectrum or a stack of frames; logging it would dwarf the log and put measurement data somewhere it was never meant to go. The error type is recorded, not its message, for the same reason: the message can carry a path, an address or an instrument’s own words.

delivered is false when the reply could not be routed — the client timed out and reconnected while the device was working. The operation happened and the outcome is real; only the answer was lost. That is the case a traffic pane most needs to show, because the caller saw a failure that did not occur.

A LIVENESS_TYPES request is recorded at DEBUG rather than INFO, so a supervisor polling the server does not drown the calls that actually moved the instrument. It is recorded, not dropped.

Never raises: a monitor that can break the thing it monitors is worse than no monitor. A malformed request, an unserializable field or a logging handler that throws costs the record, not the request.

Parameters:

request_str (str) – The request as received.
identity (bytes) – ROUTER identity frame of the client.
response (dict) – What _handle() answered.
elapsed (float) – Seconds from receipt to the end of the send attempt.
delivered (bool) – Whether the reply reached the client.

Return type:

None

_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: its callable methods and its documented schema.

methods is the original reply and stays first, so a client written against it keeps working. doc is the device’s DeviceDocModel serialized — units, bounds, options and descriptions — which is what a caller needs to offer a parameter without inventing its constraints. A device whose model cannot be built still answers with its method list.

Returns:: {"methods": [...], "budgets": {...}, "doc": {...} | None}.
Return type:: dict

static _doc_payload(device: Any) → dict[str, Any] | None

Return the device’s serialized doc model, or None when it has none.

A describe must answer. Building the model walks the parameter and function registries and every @expose_to_api signature, so a module with a malformed annotation would otherwise take down the one request a supervisor uses to decide the device is alive.

Parameters:: device (Any) – The unwrapped device.
Returns:: The doc model as JSON primitives, or None.
Return type:: dict[str, Any] | None

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.

A stop is delivered as KeyboardInterrupt on every platform (see stop_on_break()), so the finally below closes the device whichever signal the supervisor sent.

Return type:: None

async _answer(identity: bytes, request: str) → None

Handle one request and deliver its reply.

Runs as its own task so the accept loop keeps receiving. Sends are serialized on _send_lock — the tasks share one ROUTER socket, and a reply is two frames that must not be interleaved with another’s.

Parameters:

identity (bytes) – ROUTER identity frame of the requesting client.
request (str) – The request, as received.

Return type:

None

async _serve() → None

The accept loop, wrapped by run().

Return type:: None

async shutdown() → None

Gracefully shutdown the server.

Return type:: None