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.

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

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

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