plesty.bench.client

The bench client: one ZMQ DEALER socket, four request types.

The wire is the plesty-lib device-server wire — JSON frames, status: ok or status: error replies — so a plesty.lib.service.build_client() or a describe probe understands the agent too. Requests:

{"type": "describe"}: → {"status": "ok", "result": {"kind": "plesty-server-agent", "version": …, "home": …, "commands": [...], "requests": [...]}}
{"type": "exec", "argv": [...], "token": "…"}: → {"status": "ok", "result": {"code": 0, "stdout": "…", "stderr": "…"}}
{"type": "fetch", "path": "relative/to/home", "token": "…"}: → {"status": "ok", "result": {"path": …, "size": n, "data": "<base64>"}}
{"type": "device", "name": "pm100d", "request": {…}, "token": "…"}: → the device server’s own reply, forwarded unchanged

Everything long-running is already a job on the bench: ask for it with --no-wait, then BenchClient.follow() the job log.

device is the exception to that rule, and to “the CLI is the API”: a control panel is a long-lived session of many small round trips, which exec would answer by running the click application once per poll. BenchClient.device() wraps it as a RemoteDevice, the handle a plesty-lib DevicePanel drives.

Attributes

`DEFAULT_PORT`
`ADDRESS_ENV`
`TOKEN_ENV`

Exceptions

BenchClientError

The bench did not answer, refused the request, or the command failed to run.

Classes

`ExecResult`	What one remote `plesty-server` invocation produced.
`BenchClient`	A connection to one bench agent.
`RemoteDevice`	A device on the bench, driven through the agent.

Functions

parse_address(→ str)

Turn host, host:port or tcp://host:port into a ZMQ endpoint.

Module Contents

plesty.bench.client.DEFAULT_PORT = 5550

plesty.bench.client.ADDRESS_ENV = 'PLESTY_BENCH'

plesty.bench.client.TOKEN_ENV = 'PLESTY_BENCH_TOKEN'

exception plesty.bench.client.BenchClientError

Bases: RuntimeError

The bench did not answer, refused the request, or the command failed to run.

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

class plesty.bench.client.ExecResult

What one remote plesty-server invocation produced.

Variables:

argv – The arguments that were run.
code – Exit code.
stdout – Captured standard output.
stderr – Captured standard error (progress lines).

argv: tuple[str, Ellipsis]

code: int

stdout: str

stderr: str

property ok: bool

Whether the command exited 0.

Return type:: bool

json() → Any

Parse stdout as JSON (for commands run with --json).

Return type:: Any

plesty.bench.client.parse_address(value: str | None) → str

Turn host, host:port or tcp://host:port into a ZMQ endpoint.

Parameters:: value (str | None) – The address; None reads PLESTY_BENCH, else localhost.
Return type:: str

class plesty.bench.client.BenchClient(address: str | None = None, token: str | None = None, timeout_ms: int = 10000, context: zmq.Context[Any] | None = None)

A connection to one bench agent.

Connect lazily to address.

Parameters:

address (str | None) – host, host:port or tcp://host:port; None reads PLESTY_BENCH.
token (str | None) – Shared secret the agent was started with; None reads PLESTY_BENCH_TOKEN.
timeout_ms (int) – Default reply timeout for request().
context (zmq.Context[Any] | None) – ZMQ context to use; the global instance by default.

address

token

timeout_ms = 10000

_context = None

_socket: zmq.Socket[Any] | None = None

__enter__() → BenchClient

Open the socket.

Return type:: BenchClient

__exit__(*_exc: object) → None

Close the socket.

Parameters:: _exc (object)
Return type:: None

close() → None

Close the socket (a later request reconnects).

Return type:: None

_connect() → zmq.Socket[Any]

Return type:: zmq.Socket[Any]

request(message: dict[str, Any], timeout_ms: int | None = None) → dict[str, Any]

Send one request and return the result of an ok reply.

Raises:

BenchClientError – On no reply in time, a malformed reply, or an error reply (its message is the exception text).

Parameters:

message (dict[str, Any])
timeout_ms (int | None)

Return type:

dict[str, Any]

describe(timeout_ms: int | None = None) → dict[str, Any]

The agent’s self-description (kind, version, home, commands).

Parameters:: timeout_ms (int | None)
Return type:: dict[str, Any]

ping(timeout_ms: int = 2000) → bool

Whether a bench agent answers at the address.

Parameters:: timeout_ms (int)
Return type:: bool

exec(argv: collections.abc.Sequence[str], timeout_ms: int | None = None) → ExecResult

Run plesty-server <argv> on the bench and return what it produced.

The result carries the exit code; it does not raise on a non-zero exit, only when the agent could not run the command at all.

Parameters:

argv (collections.abc.Sequence[str])
timeout_ms (int | None)

Return type:

ExecResult

fetch(path: str, timeout_ms: int | None = None) → bytes

Download a file from the bench home (path relative to it).

Parameters:

path (str)
timeout_ms (int | None)

Return type:

bytes

device(name: str, timeout_ms: int | None = None) → RemoteDevice

Return a handle on one running device of the bench.

Parameters:

name (str) – The declared device.
timeout_ms (int | None) – Reply timeout for its calls; the client’s by default.

Returns:

A RemoteDevice — the four calls a control panel makes.

Return type:

RemoteDevice

follow(job_id: str, sink: collections.abc.Callable[[str], None], poll_s: float = 0.5, timeout_s: float | None = None) → int

Stream the log of job_id to sink until the job ends; returns its exit code.

Uses jobs log <id> --offset N --json on the bench, so nothing is held open between polls and a lost connection just resumes.

Raises:

BenchClientError – When the job is unknown or the bench stops answering.

Parameters:

job_id (str)
sink (collections.abc.Callable[[str], None])
poll_s (float)
timeout_s (float | None)

Return type:

int

class plesty.bench.client.RemoteDevice(client: BenchClient, name: str, timeout_ms: int | None = None)

A device on the bench, driven through the agent.

The four calls plesty-lib’s DeviceHandle protocol asks for, so a DevicePanel built on this is the same panel the console opens beside the instrument — only the transport differs.

Every call is one device request: the agent forwards it to that device’s own server and hands the reply straight back. It goes through the agent rather than to the device’s port because device servers check no token and the agent does; the bench keeps one authenticated door.

Bind to name on the bench client is connected to.

Parameters:

client (BenchClient) – The bench connection.
name (str) – The declared device.
timeout_ms (int | None) – Reply timeout for its calls; the client’s by default.

client

name

timeout_ms = None

_send(request: dict[str, Any]) → Any

Relay one device-server request and return its result.

Parameters:: request (dict[str, Any])
Return type:: Any

describe() → dict[str, Any]

Return the device’s methods and its serialized doc model.

Return type:: dict[str, Any]

query(key: str) → Any

Read configuration parameter key.

Parameters:: key (str)
Return type:: Any

write(key: str, value: Any) → Any

Write value to configuration parameter key.

Parameters:

key (str)
value (Any)

Return type:

Any

call(op: str, **kwargs: Any) → Any

Run operation op.

Parameters:

op (str)
kwargs (Any)

Return type:

Any

close() → None

Release nothing: the connection belongs to the bench client.

Return type:: None