plesty.bench

plesty.bench — talk to a bench running plesty-server from another machine.

This is the client half of plesty-server and the only part published to PyPI (distribution plesty-bench). It depends on pyzmq and click and imports nothing from plesty.server: the developer machine that runs the remote field-test loop, or a script on the experiment PC, installs this and nothing of the bench application.

The contract is deliberately thin — the CLI is the API: the bench agent accepts a plesty-server argument list, runs it, and returns the exit code and the output. Anything the command line can do is remote, with no second schema to keep in step.

from plesty.bench import BenchClient

with BenchClient("tcp://lab-bench-03:5550", token="…") as bench:
    print(bench.describe()["result"]["version"])
    result = bench.exec(["status", "--json"])
    job = bench.exec(["field-test", "pm100d", "--no-wait"]).stdout.strip()
    bench.follow(job, print)                         # tails the job log until it ends
    data = bench.fetch("repos/plesty-pm100d/exp/reports/pm100d-s120c/field-test.json")

Or from a shell: plesty-bench -b lab-bench-03 -- status.

Submodules

Attributes

DEFAULT_PORT

Exceptions

BenchClientError

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

Classes

`BenchClient`	A connection to one bench agent.
`ExecResult`	What one remote `plesty-server` invocation produced.
`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.

Package Contents

plesty.bench.DEFAULT_PORT = 5550

class plesty.bench.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

exception plesty.bench.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.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

class plesty.bench.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

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