plesty.server.services

Services: the model’s side effects on the host.

Everything that touches the operating system — uv and git for installing, subprocesses for running servers and jobs — lives here. Services take model objects in and hand model objects back; presenters drive them and never call the tools directly, so a service is the one place a tool’s quirks are handled and the one seam tests replace.

Submodules

Attributes

CHECKOUT_VENV

LOADABLE_TYPES

PYTHON_SPEC

Echo

Runner

DEVICE_TIMEOUT_MS

ABORT

CANCEL

CONTROL_FILE

EXPERIMENT_KIND

RUN_CONFIG_FILE

RUN_LOGS_DIR

RUN_REPORT_FILE

VIEWER_KIND

LOCAL_HOST_NAME

ACTION_TIMEOUT_MS

POLL_TIMEOUT_MS

PROBE_HOST

Exceptions

GitUnavailable

No git on the bench, and none could be installed.

InstallError

uv/git failed, or the environment is not in the shape expected.

CatalogueError

The catalogue could not be fetched or read.

ExperimentError

Raised when an experiment cannot be read, started or driven.

NodeError

A node could not be reached, or refused what it was asked.

AlreadyRunning

start was asked for a device whose server is already up.

NotRunning

stop was asked for a device that has no live server.

PortBusy

start found the device's port already answering — another server holds it.

SupervisorError

A start or stop request could not be carried out.

Classes

Installer

Install, update and remove device environments; records them in the manager.

LocalModule

What a repository on disk says it is, in its pyproject.toml.

Agent

Serve plesty.bench clients for one bench.

Catalogue

The hub catalogue at one index URL.

DeviceLink

A lazy DeviceHandle over one device server.

JobRunner

Start, observe and cancel jobs under a Home.

NodeDevice

One device of a node, on a connection this handle owns.

NodeLink

This master's link to one node's agent.

Nodes

Every node a master drives, one NodeLink each.

Supervisor

Launch and control device servers under a Home.

Functions

bench_git(→ pathlib.Path)

Where the bench keeps its own git (whether or not one is there yet).

find_git(→ str | None)

A usable git: PATH first, then the usual installs, then the bench's own.

install_git(→ str)

Unpack a portable git into <home>/tools/git; returns the executable.

default_runner(→ subprocess.CompletedProcess[str])

Run cmd, streaming its merged output to echo as it arrives.

local_module(→ LocalModule)

Read what the repository at root is, and refuse what a bench cannot run.

serve_in_thread(→ tuple[threading.Thread, threading.Event])

Run agent on a daemon thread; returns the thread and its stop event.

http_fetch(→ bytes)

GET url with the standard library (the bench may have no requests).

pypi_release(→ str | None)

The newest release of package on PyPI, or None when it has none there.

pypi_releases(→ tuple[str, Ellipsis])

Every release of package on PyPI, newest first; none when it has none or PyPI is out.

release_order(→ tuple[tuple[int, Ellipsis], int, int])

A sort key for release versions: a snapshot or pre-release before its release.

remote_tags(→ tuple[str, Ellipsis])

The release tags (v0.3.1) of the repository at url, newest first.

ask(→ dict[str, Any])

Amend a run's control file and return what it now asks for.

asked(→ dict[str, Any])

Return what a run is currently being asked for.

clear(→ None)

Remove a run's control file, so a resumed run is asked for nothing.

check_run_id(→ str)

Return text if it is usable as a run directory name, else raise.

add_note(→ None)

Append a note to the run's NOTES_FILE, while it runs or after it ended.

create_run(→ dict[str, Any])

Name a run before it is started, recording what it will be run with.

created_of(→ dict[str, Any])

Return what the console recorded when the run was created, if anything.

find_run(→ pathlib.Path | None)

Return the run directory run_id names, or the newest one.

notes_of(→ list[dict[str, str]])

The run's notes, oldest first: each {"at", "by", "text"}; none without a file.

has_started(→ bool)

Whether the run has started: its configuration is frozen into its plan then.

keep_log(→ pathlib.Path)

Copy a job's log into the run it ran, under RUN_LOGS_DIR.

kept_log(→ pathlib.Path | None)

The newest job log run_dir keeps, or None when it keeps none.

latest(→ pathlib.Path | None)

Return the newest run of run_root, or None when there is none.

progress_of(→ dict[str, Any] | None)

Return what the step in flight last said about itself.

read_devices(→ list[str] | None)

Ask the module which devices of its setup.yaml a run of config uses.

read_manifest(...)

Ask the module what it takes, and return what it said.

reconfigure_run(→ pathlib.Path)

Replace the configuration of a run that has not started.

records_of(→ list[dict[str, Any]])

Return the newest committed step records.

rename_run(→ pathlib.Path)

Give a run that has not started another name, which is its directory's.

report_of(→ dict[str, Any] | None)

The run's own report, or None when it has none.

resumable(→ str | None)

Return the id of the newest run that could be continued, if any.

run_environment(→ dict[str, str])

Build the environment a run gets.

runs_of(→ list[pathlib.Path])

Return this experiment's run directories, newest first.

status_of(→ dict[str, Any])

Return what a run's journal says about it, by replaying the journal here.

steps_of(→ list[dict[str, Any]])

Return the plan's steps, so a status can name the one in flight.

probe(→ bool)

Ask the server on port to describe itself.

request(→ dict[str, Any])

Send one request to the server on port and return its reply.

Package Contents

exception plesty.server.services.GitUnavailable

Bases: RuntimeError

No git on the bench, and none could be installed.

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

plesty.server.services.bench_git(home: plesty.server.model.home.Home) → pathlib.Path

Where the bench keeps its own git (whether or not one is there yet).

Parameters:

home (plesty.server.model.home.Home)

Return type:

pathlib.Path

plesty.server.services.find_git(home: plesty.server.model.home.Home) → str | None

A usable git: PATH first, then the usual installs, then the bench’s own.

Parameters:

home (plesty.server.model.home.Home) – The bench home, which may hold a git of its own.

Returns:

The executable, or None when the bench has none.

Return type:

str | None

plesty.server.services.install_git(home: plesty.server.model.home.Home, echo: Echo | None = None, fetch: Fetcher | None = None) → str

Unpack a portable git into <home>/tools/git; returns the executable.

Windows only: it is the platform where a bench routinely has no git and no way to install one. The archive is verified against the pinned digest before anything is unpacked — a mismatch is an error, not a warning.

Parameters:
Raises:

GitUnavailable – Not Windows, unknown architecture, download failed, digest mismatch, or the archive held no git.

Return type:

str

plesty.server.services.CHECKOUT_VENV = '.venv'
plesty.server.services.LOADABLE_TYPES
plesty.server.services.PYTHON_SPEC = '>=3.12'
plesty.server.services.Echo
class plesty.server.services.Installer(home: plesty.server.model.home.Home, manager: plesty.server.model.device_manager.DeviceManager | None = None, uv: str | None = None, git: str | None = None, runner: Runner | None = None, echo: Echo | None = None)

Install, update and remove device environments; records them in the manager.

Bind the installer to a home.

Parameters:

  • home (plesty.server.model.home.Home) – The bench home.

  • manager (plesty.server.model.device_manager.DeviceManager | None) – The inventory to record into; a fresh one on home by default.

  • uv (str | None) – Path of the uv executable; looked up on PATH by default.

  • git (str | None) – Path of the git executable; found (or installed) by default.

  • runner (Runner | None) – Executes a command line — injected by tests.

  • echo (Echo | None) – Where each line uv and git write goes while they run. Defaults to stdout, which is the job log when run as a job.

home
manager
_uv = None
_git = None
_run
_echo: Echo
property uv: str

The uv executable; raises InstallError when the bench has none.

Return type:

str

_find_uv() → str

PATH first, then where uv installs itself (find_uv()).

Return type:

str

property git: str

The git executable, installing one for the bench when it has none.

Raises:

InstallError – When the bench has no git and cannot be given one.

Return type:

str

_on_path(git: str) → str

Put git’s directory on PATH for this process and its children.

Calling git by absolute path is enough for the installer’s own clone and checkout — and not enough for the install. A module’s dependencies may be git+https URLs, and the uv this spawns to resolve them looks for git on PATH and nowhere else: on the very bench this feature exists for, one with no git at all, the install would fetch a portable git, clone with it, and then fail in uv sync with “Git executable not found”. The same holds for a git found under %ProgramFiles% but absent from PATH.

Parameters:

git (str)

Return type:

str

_find_or_install_git() → str

Locate git, or fetch a portable one (Windows); explains itself either way.

Return type:

str

local_branches(root: pathlib.Path) → list[str]

The local branches of the repository at root; empty when it has none or no git.

Parameters:

root (pathlib.Path)

Return type:

list[str]

local_branch(root: pathlib.Path) → str

The branch the repository at root is on; "" when detached or not git.

Parameters:

root (pathlib.Path)

Return type:

str

python(installed: plesty.server.model.device_manager.InstalledVersion) → pathlib.Path

The interpreter that runs installed.

Parameters:

installed (plesty.server.model.device_manager.InstalledVersion)

Return type:

pathlib.Path

load_local(folder: str | os.PathLike[str], ref: str | None = None) → plesty.server.model.device_manager.InstalledVersion

Put a repository already on disk onto the bench, where it lives.

Nothing is copied or moved: the bench records the folder, what the module is, the version its environment holds and the commit it is on, under <home>/packages/<package>/, and runs it with the folder’s own .venv — the one the developer’s uv run uses. When there is none yet, uv sync makes it; one that is there is never re-synced, so the developer’s tools in it stay.

Loading the same folder again refreshes the record: a new commit, a bumped version.

Parameters:

  • folder (str | os.PathLike[str]) – The repository’s root, holding its pyproject.toml.

  • ref (str | None) – A branch to check out in the folder first; the one it is on when None. Refused over uncommitted changes, which a switch would carry onto the other branch.

Returns:

The record.

Raises:

InstallError – When the folder is not a module this bench can run, the branch cannot be checked out, or the environment cannot be made.

Return type:

plesty.server.model.device_manager.InstalledVersion

_seeded(installed: plesty.server.model.device_manager.InstalledVersion) → plesty.server.model.device_manager.InstalledVersion

Give the checkout of installed a .env line for each variable its setup names.

An experiment’s setup.yaml names the variables holding where each bench is, its token and where it sees the shared disk; the .env beside it holds their values, set in the console. A value already there is kept. A module with no setup is left as it is.

Parameters:

installed (plesty.server.model.device_manager.InstalledVersion)

Return type:

plesty.server.model.device_manager.InstalledVersion

install(spec: plesty.server.model.device.DeviceSpec, force: bool = False, job_id: str | None = None) → plesty.server.model.device_manager.InstalledVersion

Make sure an environment serving spec exists; returns its record.

Parameters:

  • spec (plesty.server.model.device.DeviceSpec) – What to install: package plus a pinned version (PyPI) or a repository and ref (git checkout).

  • force (bool) – Reinstall even when a satisfying environment is on record.

  • job_id (str | None) – The job this runs as, stored on the record.

Raises:

InstallError – When uv/git fail or the package did not end up installed.

Return type:

plesty.server.model.device_manager.InstalledVersion

_install_the_tag_instead(spec: plesty.server.model.device.DeviceSpec, job_id: str | None, failure: InstallError) → plesty.server.model.device_manager.InstalledVersion

After a failed release install, install the same version from the repository.

A module is on the hub before it is on PyPI: the registry knows the release from its tag, and the package reaches the index only once the release job has run — so “that release is not on the index” is a routine state of a young module, not a broken bench. The repository holds the same version under v<version>, which is the ref a spec with a version already resolves to.

Raises:

InstallError – When no repository is known for the package, or the repository has no such tag either — carrying both reasons.

Parameters:
Return type:

plesty.server.model.device_manager.InstalledVersion

update(installed: plesty.server.model.device_manager.InstalledVersion, ref: str | None = None, job_id: str | None = None) → plesty.server.model.device_manager.InstalledVersion

Bring installed up to date; returns the (possibly new) record.

Git checkout: fetch, check out ref (default: the recorded ref), fast-forward, uv sync. PyPI: reinstall the same version. Local repository: read it again — it is the developer’s to pull and sync.

Parameters:
Return type:

plesty.server.model.device_manager.InstalledVersion

uninstall(installed: plesty.server.model.device_manager.InstalledVersion) → bool

Delete the environment of installed and forget it.

Returns:

True when a directory was removed.

Parameters:

installed (plesty.server.model.device_manager.InstalledVersion)

Return type:

bool

_install_release(spec: plesty.server.model.device.DeviceSpec, job_id: str | None) → plesty.server.model.device_manager.InstalledVersion
Parameters:
Return type:

plesty.server.model.device_manager.InstalledVersion

_install_checkout(spec: plesty.server.model.device.DeviceSpec, job_id: str | None) → plesty.server.model.device_manager.InstalledVersion
Parameters:
Return type:

plesty.server.model.device_manager.InstalledVersion

_uv_call(args: list[str], what: str, quiet: bool = False, venv: pathlib.Path | None = None, creates: pathlib.Path | None = None) → subprocess.CompletedProcess[str]

Run uv, waiting out a scanner that is holding a file it just wrote.

uv writes every console script as an .exe and reopens it at once to patch its resources. An on-access scanner opening that new executable first makes the reopen fail, and uv reports the whole wheel as uninstallable — so an install dies on whichever dependency ships the most scripts, which is why the wheel named in the error moves around (PySide6 on one bench, Sphinx on another) and why it is nothing to do with the module being installed.

The wait is the fix: nothing here can make the scanner let go, and the file is fine once it does. Only this failure is retried, and only a few times — a scanner that never releases is a bench problem, and saying so beats retrying until the job looks hung.

A command that builds venv and was interrupted can leave it without VENV_CONFIG; the retry then failed “not a valid Python environment” every time (#45). Such a .venv is removed before each retry, as _sync_checkout() removes one before its first attempt.

Parameters:

  • args (list[str])

  • what (str)

  • quiet (bool)

  • venv (pathlib.Path | None)

  • creates (pathlib.Path | None)

Return type:

subprocess.CompletedProcess[str]

_clear_bench_reports(checkout: pathlib.Path) → None

Move the bench’s own reports/ files out of the way of the commits a pull brings.

A field test writes its report into the checkout, and the procedure then commits that same file to the module. The next pull refused to overwrite the bench’s copy — untracked, or changed by a later run — so the device kept its old code until someone deleted the file by hand on the bench PC (#37). A copy identical to the incoming one is dropped; any other is kept beside it as <name>.bench-<time><suffix>, said in the log, so no result of a run is lost.

Only reports/ is touched — the folder the bench writes. Anything else in the way is the developer’s and still stops the pull.

Parameters:

checkout (pathlib.Path)

Return type:

None

_sync_checkout(checkout: pathlib.Path, package: str) → None

uv sync --no-dev checkout, rebuilding its .venv if that stops it.

The module itself is rebuilt every time (--reinstall-package). It is installed editable, so its code follows the pull anyway, but uv keeps the version metadata of the first build: after an update the bench, list and a field-test report all named the commit it was first installed from (#36).

The environment holds what the device needs to run and nothing else. tests/field_test.py is unaffected — it imports plesty.lib and the module, never pytest. What no longer works in a fresh checkout is the authoring toolchain: plesty check, plesty init and pytest in the Terminal tab have nothing to run, because plesty-sdk lives in the dev group. Someone who wants them can uv sync in the checkout by hand and get them back.

uv refuses a .venv that exists without an interpreter in it rather than replacing it, and nothing removes it, so every later sync fails the same way: one interrupted install wedges the module until someone deletes the directory by hand — which is what a bench operator has no console for, and what the message does not tell them to do. The interruption can be almost anything (a stop mid-install, a scanner taking the interpreter, an uv-managed Python that went away), so this recovers from the state rather than from any one cause.

Deleting it is safe: uv owns the directory and rebuilds it from the lock. Only this failure is retried — any other sync error is the module’s own and is raised unchanged.

A .venv without VENV_CONFIG is removed before the sync, and one still without it afterwards fails the install: its interpreter is the shared one, and an install that “succeeds” into it is worse than one that stops.

Parameters:

  • checkout (pathlib.Path)

  • package (str)

Return type:

None

_uv_resolve(args: list[str], what: str, venv: pathlib.Path | None = None) → subprocess.CompletedProcess[str]

Run a resolving uv command, retrying once with pre-releases allowed.

A hub module under development depends on a .dev release of plesty-lib, and uv considers only stable versions unless it is told otherwise — it fails with a hint naming the pre-release. Allowing them from the start would pull a pre-release into every install that has a stable answer, so it is the retry, not the rule.

Parameters:

  • args (list[str])

  • what (str)

  • venv (pathlib.Path | None)

Return type:

subprocess.CompletedProcess[str]

_git_call(args: list[str], what: str, quiet: bool = False) → subprocess.CompletedProcess[str]
Parameters:

  • args (list[str])

  • what (str)

  • quiet (bool)

Return type:

subprocess.CompletedProcess[str]

_call(cmd: list[str], what: str, quiet: bool = False) → subprocess.CompletedProcess[str]

Run cmd; quiet keeps a query’s output out of the log (it is read, not watched).

Parameters:

  • cmd (list[str])

  • what (str)

  • quiet (bool)

Return type:

subprocess.CompletedProcess[str]

_installed_version(python: pathlib.Path, package: str) → str | None
Parameters:

  • python (pathlib.Path)

  • package (str)

Return type:

str | None

_python_version(python: pathlib.Path) → str
Parameters:

python (pathlib.Path)

Return type:

str

exception plesty.server.services.InstallError

Bases: RuntimeError

uv/git failed, or the environment is not in the shape expected.

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

class plesty.server.services.LocalModule

What a repository on disk says it is, in its pyproject.toml.

Variables:

  • package – [project] name.

  • module_type – [tool.plesty] module_type — device, experiment, …

  • description – [project] description.

package: str
module_type: str
description: str = ''
property category: str

devices or experiments.

Type:

The catalogue category it belongs to

Return type:

str

plesty.server.services.Runner
plesty.server.services.default_runner(cmd: list[str], echo: Echo | None = None) → subprocess.CompletedProcess[str]

Run cmd, streaming its merged output to echo as it arrives.

stderr is folded into stdout so the order the tool wrote them in survives; the text is returned as well, so a failure still carries its own detail.

Parameters:

  • cmd (list[str])

  • echo (Echo | None)

Return type:

subprocess.CompletedProcess[str]

plesty.server.services.local_module(root: pathlib.Path) → LocalModule

Read what the repository at root is, and refuse what a bench cannot run.

Raises:

InstallError – When there is no pyproject.toml, it names no project, or it is not a device or experiment module.

Parameters:

root (pathlib.Path)

Return type:

LocalModule

class plesty.server.services.Agent(home: plesty.server.model.home.Home, fleet_path: str | os.PathLike[str] | None = None, bind: str = DEFAULT_BIND, token: str | None = None, workers: int = 4)

Serve plesty.bench clients for one bench.

Configure the agent.

Parameters:

  • home (plesty.server.model.home.Home) – The bench home every command runs against.

  • fleet_path (str | os.PathLike[str] | None) – Fleet file passed to every command (-f), or None.

  • bind (str) – ZMQ endpoint to bind.

  • token (str | None) – Shared secret; None reads PLESTY_BENCH_TOKEN; empty means no check (a bench on a trusted network).

  • workers (int) – Concurrent requests.

home
fleet_path
bind = 'tcp://*:5550'
token
workers = 4
bound: str | None = None
served = 0
device_timeout_ms = 15000
handle(message: Any) → dict[str, Any]

Answer one decoded request (pure; no socket involved).

Parameters:

message (Any)

Return type:

dict[str, Any]

description() → dict[str, Any]

What describe returns.

Return type:

dict[str, Any]

exec(argv: collections.abc.Sequence[Any]) → dict[str, Any]

Run plesty-server <argv> in-process against this bench; capture everything.

Parameters:

argv (collections.abc.Sequence[Any])

Return type:

dict[str, Any]

device(name: str, request: Any) → dict[str, Any]

Relay one device-server request to the server of name, and its reply back.

Why a request type of its own, when the rule is “the CLI is the API”: a control panel is a long-lived session of many small round trips — a poll per parameter per second — and exec would spawn the click application for each one. This carries the device wire itself, so the panel a console opens locally is the same panel driven from a laptop.

Why relayed at all, when device servers bind tcp://* and could be reached directly: they check no token. The agent does, so the bench keeps one authenticated door instead of one per instrument.

The reply is passed through as the device server framed it, so an error from the instrument reaches the caller as its own error, not as a relay failure.

Parameters:

  • name (str) – The declared device whose server to reach.

  • request (Any) – The device-server request (query, write, call, describe, help) to forward verbatim.

Returns:

The reply envelope to send back to the client.

Raises:

ValueError – When the name or the request is missing or malformed.

Return type:

dict[str, Any]

fetch(path: str) → dict[str, Any]

Read a file under the bench home; path is relative to it.

Parameters:

path (str)

Return type:

dict[str, Any]

serve(stop: threading.Event, poll_ms: int = 100, max_replies: int | None = None) → int

Bind and answer requests until stop is set; returns the number answered.

Parameters:

  • stop (threading.Event) – Set to end the loop.

  • poll_ms (int) – Socket poll interval.

  • max_replies (int | None) – Stop once this many replies went out (tests).

Return type:

int

plesty.server.services.serve_in_thread(agent: Agent, stop: threading.Event | None = None) → tuple[threading.Thread, threading.Event]

Run agent on a daemon thread; returns the thread and its stop event.

Parameters:

  • agent (Agent)

  • stop (threading.Event | None)

Return type:

tuple[threading.Thread, threading.Event]

class plesty.server.services.Catalogue(index_url: str | None = None, fetch: Fetcher | None = None)

The hub catalogue at one index URL.

Bind to an index.

Parameters:

  • index_url (str | None) – /api/modules/ endpoint; the public hub by default.

  • fetch (Fetcher | None) – Reads a URL; injected by tests.

index_url = 'https://hub.plesty.net/api/modules/'
_fetch
_cache: list[plesty.server.model.catalogue.CatalogueEntry] | None = None
entries(refresh: bool = False) → list[plesty.server.model.catalogue.CatalogueEntry]

Every module the index lists (fetched once, then cached).

Raises:

CatalogueError – When the index cannot be fetched or is not JSON.

Parameters:

refresh (bool)

Return type:

list[plesty.server.model.catalogue.CatalogueEntry]

search(query: str = '', vendor: str = '', category: str = 'devices', standard: str = REQUIRED_STANDARD, refresh: bool = False) → list[plesty.server.model.catalogue.CatalogueEntry]

Entries matching query, filtered by vendor, category and standard ("" = any).

refresh asks the index again instead of answering from the cache — what a module published while the bench runs needs to be seen.

Parameters:

  • query (str)

  • vendor (str)

  • category (str)

  • standard (str)

  • refresh (bool)

Return type:

list[plesty.server.model.catalogue.CatalogueEntry]

vendors(category: str = 'devices') → list[str]

The vendor groups present in category.

Parameters:

category (str)

Return type:

list[str]

lookup(package: str) → plesty.server.model.catalogue.CatalogueEntry | None

The entry named package (any category, any standard), or None.

Parameters:

package (str)

Return type:

plesty.server.model.catalogue.CatalogueEntry | None

exception plesty.server.services.CatalogueError

Bases: RuntimeError

The catalogue could not be fetched or read.

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

plesty.server.services.http_fetch(url: str, timeout: float = 15.0) → bytes

GET url with the standard library (the bench may have no requests).

Parameters:

  • url (str)

  • timeout (float)

Return type:

bytes

plesty.server.services.pypi_release(package: str, fetch: Fetcher | None = None) → str | None

The newest release of package on PyPI, or None when it has none there.

Parameters:

  • package (str) – Distribution name.

  • fetch (Fetcher | None) – Reads a URL; injected by tests.

Return type:

str | None

plesty.server.services.pypi_releases(package: str, fetch: Fetcher | None = None) → tuple[str, Ellipsis]

Every release of package on PyPI, newest first; none when it has none or PyPI is out.

A release whose files were all yanked is left out: uv would refuse to install it unpinned, and offering it pinned offers what its author withdrew.

Parameters:

  • package (str) – Distribution name.

  • fetch (Fetcher | None) – Reads a URL; injected by tests.

Return type:

tuple[str, Ellipsis]

plesty.server.services.release_order(version: str) → tuple[tuple[int, Ellipsis], int, int]

A sort key for release versions: a snapshot or pre-release before its release.

Plain string or chunk-wise order puts 0.5.0.dev6 above 0.5.0 and 0.10.0 below 0.9.0. Trailing zeros are dropped, so 1.0 and 1.0.0 are the same release. Anything unparseable sorts last.

Parameters:

version (str)

Return type:

tuple[tuple[int, Ellipsis], int, int]

plesty.server.services.remote_tags(git: str, url: str, timeout: float = 20.0) → tuple[str, Ellipsis]

The release tags (v0.3.1) of the repository at url, newest first.

One git ls-remote; none when it fails or takes longer than timeout. Git is told never to prompt: a repository that wants credentials is one whose tags are not listed, not a console that hangs on a hidden prompt.

Parameters:

  • git (str) – The git executable.

  • url (str) – Clone URL.

  • timeout (float) – Seconds before giving up.

Return type:

tuple[str, Ellipsis]

plesty.server.services.DEVICE_TIMEOUT_MS = 15000

A lazy DeviceHandle over one device server.

Prepare a link; nothing is connected yet.

Parameters:

  • name (str) – The device’s fleet name, for messages.

  • address (str) – ZMQ endpoint of its server, tcp://host:port.

  • timeout_ms (int) – Reply timeout for one call.

  • resolve (collections.abc.Callable[[], str | None] | None) – Returns where the server is now, or None when it is not running; raises when that is not known. A restarted server can come back on another port; a link that kept the first address drove a port nobody listened on, and the panel’s controls silently did nothing (#22).

name
address
timeout_ms = 15000
_resolve = None
_client: Any = None
_held = False
hold() → None

Refuse every call until release(): the server is being stopped.

A call sent to a server that is shutting down, or to the port it left, gets no reply and waits out the whole timeout — and every call after it waits behind it on the panel’s one worker. Restarting on a new port left the panel dead for up to fifteen seconds that way (#22). Held, a call fails at once and sends nothing.

Return type:

None

release() → None

Let calls through again after hold().

Return type:

None

_follow() → None

Move to the server’s current address, dropping a client on the old one.

Raises:

ConnectionError – When the server is not running, rather than send a call to an address nothing answers on.

Return type:

None

client() → Any

Return the connected client, building it on first use.

Returns:

The DeviceTCPIPClient.

Raises:

ConnectionError – When the link is held or the server is not running.

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 the connection; a later call reconnects.

Never raises: this runs when a panel closes and a window is going away, and a server that already died must not be what stops it.

Return type:

None

plesty.server.services.ABORT = 'abort'
plesty.server.services.CANCEL = 'cancel'
plesty.server.services.CONTROL_FILE = 'control.json'
plesty.server.services.EXPERIMENT_KIND = 'experiment'
plesty.server.services.RUN_CONFIG_FILE = 'config.yaml'
plesty.server.services.RUN_LOGS_DIR = 'job-logs'
plesty.server.services.RUN_REPORT_FILE = 'report.html'
plesty.server.services.VIEWER_KIND = 'viewer'
exception plesty.server.services.ExperimentError

Bases: RuntimeError

Raised when an experiment cannot be read, started or driven.

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

plesty.server.services.ask(run_dir: str | pathlib.Path, **changes: Any) → dict[str, Any]

Amend a run’s control file and return what it now asks for.

The file is replaced rather than rewritten, so a run reading it sees either the old document or the new one and never half of each.

Parameters:

  • run_dir (str | pathlib.Path) – The run directory.

  • **changes (Any) – Fields of the control document to set.

Returns:

The control document as it now stands.

Raises:

ExperimentError – If the file cannot be written.

Return type:

dict[str, Any]

plesty.server.services.asked(run_dir: str | pathlib.Path) → dict[str, Any]

Return what a run is currently being asked for.

Parameters:

run_dir (str | pathlib.Path) – The run directory.

Returns:

The control document; empty when nothing is being asked.

Return type:

dict[str, Any]

plesty.server.services.clear(run_dir: str | pathlib.Path) → None

Remove a run’s control file, so a resumed run is asked for nothing.

A run resumed while its control file still says cancel would stop at its first step, which reads as a resume that did nothing.

Parameters:

run_dir (str | pathlib.Path) – The run directory.

Return type:

None

plesty.server.services.check_run_id(text: str) → str

Return text if it is usable as a run directory name, else raise.

Deliberately the console’s own check rather than plesty-lib’s, for the same reason the journal is replayed here: the rule is about what a directory name may be, and the console must apply it on any plesty-lib — including one too old to carry the check at all. It is the same rule, and a short one.

Parameters:

text (str) – The proposed run id.

Returns:

The same string, trimmed.

Raises:

ExperimentError – If it is empty, a path, or holds a character that does not belong in a directory name.

Return type:

str

plesty.server.services.add_note(run_dir: str | pathlib.Path, text: str, by: str, when: Any = None) → None

Append a note to the run’s NOTES_FILE, while it runs or after it ended.

Parameters:

  • run_dir (str | pathlib.Path) – The run directory.

  • text (str) – The note; blank lines inside it are kept.

  • by (str) – Who wrote it.

  • when (Any) – A datetime; now by default (tests pass one).

Raises:

ExperimentError – When the note is empty or cannot be written.

Return type:

None

plesty.server.services.create_run(run_dir: str | pathlib.Path, config_text: str | None = None, **fields: Any) → dict[str, Any]

Name a run before it is started, recording what it will be run with.

Parameters:

  • run_dir (str | pathlib.Path) – The directory the run will use — its name is the run id.

  • config_text (str | None) – The run’s own configuration, written as its RUN_CONFIG_FILE; none is written when None.

  • **fields (Any) – What to record: the experiment, the configuration file the run’s copy came from, and whatever else the run needs to be started later.

Returns:

The document written.

Raises:

ExperimentError – If the directory already holds a run, or cannot be written. Naming a second run over a started one would interleave two journals into one directory.

Return type:

dict[str, Any]

plesty.server.services.created_of(run_dir: str | pathlib.Path) → dict[str, Any]

Return what the console recorded when the run was created, if anything.

Parameters:

run_dir (str | pathlib.Path) – The run directory.

Returns:

The document, or {} for a run that was started directly.

Return type:

dict[str, Any]

plesty.server.services.find_run(run_root: str | pathlib.Path, run_id: str | None) → pathlib.Path | None

Return the run directory run_id names, or the newest one.

Parameters:

  • run_root (str | pathlib.Path) – The experiment’s run root.

  • run_id (str | None) – The run to find, or None for the newest.

Returns:

The directory, or None when it does not exist. A run created but not yet started counts: it is the one an operator is about to run.

Return type:

pathlib.Path | None

plesty.server.services.notes_of(run_dir: str | pathlib.Path) → list[dict[str, str]]

The run’s notes, oldest first: each {"at", "by", "text"}; none without a file.

Parameters:

run_dir (str | pathlib.Path)

Return type:

list[dict[str, str]]

plesty.server.services.has_started(run_dir: str | pathlib.Path) → bool

Whether the run has started: its configuration is frozen into its plan then.

Parameters:

run_dir (str | pathlib.Path)

Return type:

bool

plesty.server.services.keep_log(run_dir: str | pathlib.Path, log: str | pathlib.Path) → pathlib.Path

Copy a job’s log into the run it ran, under RUN_LOGS_DIR.

Copied when the job ends, not written there from the start: a run the module names itself has no directory until it has begun, and the job’s log is already being written by then.

Parameters:

  • run_dir (str | pathlib.Path) – The run directory.

  • log (str | pathlib.Path) – The job’s log, named after the job.

Returns:

The copy.

Raises:

OSError – When the log cannot be read or the copy written.

Return type:

pathlib.Path

plesty.server.services.kept_log(run_dir: str | pathlib.Path) → pathlib.Path | None

The newest job log run_dir keeps, or None when it keeps none.

Job ids end in their start time, so the newest is the last by name — the one a resumed run was finished by.

Parameters:

run_dir (str | pathlib.Path)

Return type:

pathlib.Path | None

plesty.server.services.LOCAL_HOST_NAME = 'local'
plesty.server.services.latest(run_root: str | pathlib.Path) → pathlib.Path | None

Return the newest run of run_root, or None when there is none.

Parameters:

run_root (str | pathlib.Path)

Return type:

pathlib.Path | None

plesty.server.services.progress_of(run_dir: str | pathlib.Path) → dict[str, Any] | None

Return what the step in flight last said about itself.

Parameters:

run_dir (str | pathlib.Path) – The run directory.

Returns:

The reported progress, or None when the step reports none. A step that says nothing is not an error — most are short enough not to.

Return type:

dict[str, Any] | None

plesty.server.services.read_devices(spec: plesty.server.model.experiment.ExperimentSpec, python: str | pathlib.Path, cwd: str | pathlib.Path, config: str, env: Mapping[str, str] | None = None, timeout: float = MANIFEST_TIMEOUT) → list[str] | None

Ask the module which devices of its setup.yaml a run of config uses.

Only for a module whose manifest says chooses_devices (plesty-lib#62).

Parameters:

  • spec (plesty.server.model.experiment.ExperimentSpec) – The declared experiment.

  • python (str | pathlib.Path) – Interpreter of its installed environment.

  • cwd (str | pathlib.Path) – The checkout.

  • config (str) – The configuration file the run will read.

  • env (Mapping[str, str] | None) – Environment for the call.

  • timeout (float) – Seconds to wait.

Returns:

The device names, or None for all of them.

Raises:

ExperimentError – If the command fails, times out, or prints no list.

Return type:

list[str] | None

plesty.server.services.read_manifest(spec: plesty.server.model.experiment.ExperimentSpec, python: str | pathlib.Path, cwd: str | pathlib.Path, env: Mapping[str, str] | None = None, timeout: float = MANIFEST_TIMEOUT) → plesty.server.model.experiment.ExperimentManifest

Ask the module what it takes, and return what it said.

Parameters:

  • spec (plesty.server.model.experiment.ExperimentSpec) – The declared experiment.

  • python (str | pathlib.Path) – Interpreter of its installed environment.

  • cwd (str | pathlib.Path) – Where to run it — the checkout, so a relative configuration path in the answer means what it says.

  • env (Mapping[str, str] | None) – Environment for the call; the manifest needs none, but a module that reads its own package data may.

  • timeout (float) – Seconds to wait.

Returns:

The parsed manifest.

Raises:

ExperimentError – If the command fails, times out, or prints something that is not a manifest.

Return type:

plesty.server.model.experiment.ExperimentManifest

plesty.server.services.reconfigure_run(run_dir: str | pathlib.Path, config_text: str, config: str) → pathlib.Path

Replace the configuration of a run that has not started.

Parameters:

  • run_dir (str | pathlib.Path) – The run directory.

  • config_text (str) – The run’s configuration, as it is to be run.

  • config (str) – The module’s file it was taken from (config/demo.yaml), recorded in CREATED_FILE so the run table names it.

Returns:

The run’s configuration file.

Raises:

ExperimentError – If the run has started — its plan was built from the configuration it had, and resuming checks it against that — or the files cannot be written.

Return type:

pathlib.Path

plesty.server.services.records_of(run_dir: str | pathlib.Path, limit: int = 200) → list[dict[str, Any]]

Return the newest committed step records.

Parameters:

  • run_dir (str | pathlib.Path) – The run directory.

  • limit (int) – How many to return, newest last.

Returns:

The records in write order. A record is the durable result of its step, so this is the raw data a run has produced so far.

Return type:

list[dict[str, Any]]

plesty.server.services.rename_run(run_dir: str | pathlib.Path, new_name: str) → pathlib.Path

Give a run that has not started another name, which is its directory’s.

The directory moves with everything in it — the run’s configuration and its CREATED_FILE — and a duplicate recorded as a copy of it follows the new name, so it stays beside it in the table (#27).

Parameters:

  • run_dir (str | pathlib.Path) – The run directory.

  • new_name (str) – What to call it; checked as check_run_id() checks.

Returns:

The run’s directory under its new name.

Raises:

ExperimentError – If the name is unusable or taken, the run has started — its journal and data are written under the name it has — or the directory cannot be moved.

Return type:

pathlib.Path

plesty.server.services.report_of(run_dir: str | pathlib.Path) → dict[str, Any] | None

The run’s own report, or None when it has none.

Parameters:

run_dir (str | pathlib.Path) – The run directory.

Returns:

Its name, path, bytes and written (epoch seconds), or None when no report has been written — a run that has not ended, or one that ended before this bench wrote them.

Return type:

dict[str, Any] | None

plesty.server.services.resumable(run_root: str | pathlib.Path) → str | None

Return the id of the newest run that could be continued, if any.

A run is resumable when it stopped without finishing: canceled or aborted, or left running by a process that is no longer there. A completed run is not, and neither is one that never wrote a journal.

Parameters:

run_root (str | pathlib.Path) – The experiment’s run root.

Returns:

The run id, or None.

Return type:

str | None

plesty.server.services.run_environment(spec: plesty.server.model.experiment.ExperimentSpec, base: Mapping[str, str], dotenv: Mapping[str, str]) → dict[str, str]

Build the environment a run gets.

The bench’s own environment first, then the module’s .env, then what the fleet declares — so the declaration is what an operator edits to point one experiment at a different stage, and the .env stays the module’s own deployment file.

Parameters:
Returns:

The environment for the run.

Return type:

dict[str, str]

plesty.server.services.runs_of(run_root: str | pathlib.Path) → list[pathlib.Path]

Return this experiment’s run directories, newest first.

A run directory is one holding a journal.jsonl — or a created.json, for a run named and configured but not yet started. The run root may also hold whatever the module wrote beside its runs, and a directory that is neither is not a run anything can report on.

Parameters:

run_root (str | pathlib.Path) – The experiment’s run root.

Returns:

The run directories, newest first by name — run ids carry a date, so name order is close enough to time order to read as it.

Return type:

list[pathlib.Path]

plesty.server.services.status_of(run_dir: str | pathlib.Path) → dict[str, Any]

Return what a run’s journal says about it, by replaying the journal here.

Deliberately not plesty.lib.experiment.Journal.status(), though it computes the same thing. The run was written by an experiment module, whose plesty-lib has nothing to do with the one this console happens to have installed: delegating would make the same run read differently on two benches, and a run written by a newer module would lose its paused state on an older console. The journal is an append-only document with five event names, and reading it is cheaper than the coupling.

Parameters:

run_dir (str | pathlib.Path) – The run directory.

Returns:

state, completed, failed, reports, started, ended, provenance and the last event. A run that was named but not yet started is created, which is distinct from not_started: one is a run waiting to go, the other an experiment that has never had one.

Return type:

The status mapping

plesty.server.services.steps_of(run_dir: str | pathlib.Path) → list[dict[str, Any]]

Return the plan’s steps, so a status can name the one in flight.

Parameters:

run_dir (str | pathlib.Path) – The run directory.

Returns:

One {id, op, params} per step; empty when the plan is unreadable.

Return type:

list[dict[str, Any]]

class plesty.server.services.JobRunner(home: plesty.server.model.home.Home, store: plesty.server.model.job.JobStore | None = None)

Start, observe and cancel jobs under a Home.

Bind to a home.

Parameters:
home
store
exit_file(job_id: str) → pathlib.Path

Where the launcher records the exit code of job_id.

Parameters:

job_id (str)

Return type:

pathlib.Path

start(name: str, kind: str, command: list[str], cwd: str | os.PathLike[str], env: collections.abc.Mapping[str, str] | None = None, detail: collections.abc.Mapping[str, Any] | None = None) → plesty.server.model.job.Job

Launch command as a job and return its record.

Parameters:

  • name (str) – Device instance the job belongs to.

  • kind (str) – Job kind, part of the id.

  • command (list[str]) – The command line.

  • cwd (str | os.PathLike[str]) – Working directory.

  • env (collections.abc.Mapping[str, str] | None) – Environment on top of the bench’s own.

  • detail (collections.abc.Mapping[str, Any] | None) – Kind-specific data stored on the record.

Raises:

JobError – When the command cannot be launched.

Return type:

plesty.server.model.job.Job

get(job_id: str) → plesty.server.model.job.Job | None

The job job_id with its state brought up to date, or None.

Parameters:

job_id (str)

Return type:

plesty.server.model.job.Job | None

require(job_id: str) → plesty.server.model.job.Job

Like get() but raises JobError when unknown.

Parameters:

job_id (str)

Return type:

plesty.server.model.job.Job

list(name: str | None = None, kind: str | None = None) → list[plesty.server.model.job.Job]

Jobs on record, newest first, brought up to date.

Parameters:

  • name (str | None)

  • kind (str | None)

Return type:

list[plesty.server.model.job.Job]

wait(job_id: str, timeout: float | None = None, poll: float = 0.25) → plesty.server.model.job.Job

Block until the job finishes (or timeout seconds pass); returns it.

Parameters:

  • job_id (str)

  • timeout (float | None)

  • poll (float)

Return type:

plesty.server.model.job.Job

cancel(job_id: str) → bool

Kill a running job (its whole process tree); True when it was running.

Parameters:

job_id (str)

Return type:

bool

plesty.server.services.ACTION_TIMEOUT_MS = 120000
plesty.server.services.POLL_TIMEOUT_MS = 3000
class plesty.server.services.NodeDevice(peer: plesty.server.model.Peer, name: str, connect: Connect)

One device of a node, on a connection this handle owns.

What a control panel drives. It is the four calls of plesty-lib’s DeviceHandle — describe, query, write, call — delegated to a plesty.bench.RemoteDevice, plus the one thing that handle cannot offer: a close that actually releases something.

plesty.bench.RemoteDevice closes nothing by design, because its connection belongs to the BenchClient that made it. A panel needs the opposite — a connection of its own, released when the panel goes — so that the console’s poll and the panel never share a socket, and so that the console’s _release has something to close.

The client connects lazily, so the socket is made by whichever thread makes the first call. That is the panel’s worker, which is the thread that must own it.

Open a handle on name at peer.

Parameters:

  • peer (plesty.server.model.Peer) – The node the device runs on.

  • name (str) – The declared device.

  • connect (Connect) – Builds the client; injected by the tests.

name
peer
_client
_device: Any
describe() → dict[str, Any]

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 this handle’s own connection.

Return type:

None

exception plesty.server.services.NodeError

Bases: RuntimeError

A node could not be reached, or refused what it was asked.

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

This master’s link to one node’s agent.

Bind to peer; the socket opens on the first request.

Parameters:
peer
_connect
_client: plesty.bench.BenchClient | None = None
_lock
_trouble: str = ''
property name: str

The node’s name — what the Host column shows.

Return type:

str

property trouble: str

Why the last request failed, or empty after one that worked.

Return type:

str

property reachable: bool

Whether the last request got an answer.

Return type:

bool

close() → None

Drop the connection; a later request opens a new one.

Return type:

None

run(argv: collections.abc.Sequence[str], timeout_ms: int = ACTION_TIMEOUT_MS) → str

Run plesty-server <argv> on the node and return its output.

Parameters:

  • argv (collections.abc.Sequence[str]) – The command line, as the local CLI would take it.

  • timeout_ms (int) – Reply timeout.

Returns:

Standard output, stripped.

Raises:

NodeError – When the node did not answer, or the command failed — carrying what the node said, since it is the node that knows.

Return type:

str

devices() → list[dict[str, Any]]

Every device the node declares, as list --json reports them.

The same document the local list --json prints, so one call carries the declaration, the installed version, the source line and the state — everything a row needs, in one round trip per node per refresh.

Raises:

NodeError – When the node did not answer or gave no JSON.

Return type:

list[dict[str, Any]]

device(name: str) → NodeDevice

A handle on one running device of the node, for a control panel.

On a connection of its own, not the one this link polls with. A panel drives its handle from its own worker thread while the device table keeps polling on ours, and a ZMQ socket belongs to one thread at a time: sharing one would let a panel’s reply answer a poll, and every request after that would return the one before it.

Parameters:

name (str)

Return type:

NodeDevice

class plesty.server.services.Nodes(peers: collections.abc.Iterable[plesty.server.model.Peer] = (), connect: Connect | None = None)

Every node a master drives, one NodeLink each.

Hold a link per peer.

Parameters:

  • peers (collections.abc.Iterable[plesty.server.model.Peer]) – The declared nodes.

  • connect (Connect | None) – Builds each client; injected by the tests.

_connect = None
refresh(peers: collections.abc.Iterable[plesty.server.model.Peer]) → None

Make the links match peers, keeping the connections still wanted.

A peer whose address or secret changed is reconnected; one that is simply still there keeps the socket it has, so editing the list does not cost every other node its connection.

Parameters:

peers (collections.abc.Iterable[plesty.server.model.Peer])

Return type:

None

The links, in declaration order.

Return type:

list[NodeLink]

property names: list[str]

The node names.

Return type:

list[str]

The link to the node name.

Raises:

NodeError – When no such node is declared.

Parameters:

name (str)

Return type:

NodeLink

close() → None

Drop every connection.

Return type:

None

plesty.server.services.PROBE_HOST = '127.0.0.1'
exception plesty.server.services.AlreadyRunning

Bases: SupervisorError

start was asked for a device whose server is already up.

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

exception plesty.server.services.NotRunning

Bases: SupervisorError

stop was asked for a device that has no live server.

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

exception plesty.server.services.PortBusy

Bases: SupervisorError

start found the device’s port already answering — another server holds it.

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

class plesty.server.services.Supervisor(home: plesty.server.model.home.Home, ready_timeout: float | None = None, stop_timeout: float | None = None, probe_timeout_ms: int | None = None)

Launch and control device servers under a Home.

Bind the supervisor to a home.

Parameters:

  • home (plesty.server.model.home.Home) – The bench home.

  • ready_timeout (float | None) – Seconds start() waits for the first probe answer.

  • stop_timeout (float | None) – Seconds stop() waits before escalating.

  • probe_timeout_ms (int | None) – Milliseconds one probe waits for its reply.

  • running.) ((Defaults come from config.yaml)

home
ready_timeout
stop_timeout
probe_timeout_ms
record(name: str) → plesty.server.model.process.ProcessRecord | None

The persisted launch record of name, or None.

Parameters:

name (str)

Return type:

plesty.server.model.process.ProcessRecord | None

exit_file(name: str) → pathlib.Path

Where the launcher records the exit code of the server of name.

Parameters:

name (str)

Return type:

pathlib.Path

known() → list[str]

Names that have a launch record (alive or not).

Return type:

list[str]

_drop_record(name: str) → None
Parameters:

name (str)

Return type:

None

alive(name: str) → bool

Whether the recorded server of name still runs.

Parameters:

name (str)

Return type:

bool

poll(name: str) → int | None

Exit code of the server of name, None while it runs.

LOST_EXIT when the process is gone without a recorded exit (the bench rebooted, or someone killed the launcher) — and when nothing is known at all.

Parameters:

name (str)

Return type:

int | None

status(name: str, probe_port: bool = True) → plesty.server.model.process.ProcessStatus

Observe name: record, process existence and (optionally) a probe.

Parameters:

  • name (str) – Device instance name.

  • probe_port (bool) – Whether to send a describe to tell running from starting; skipping it is cheaper for bulk listings.

Return type:

plesty.server.model.process.ProcessStatus

wait_ready(name: str, timeout: float | None = None) → bool

Poll until the server of name answers a probe.

Returns:

True once it answers; False on timeout or when the process exits first.

Parameters:

  • name (str)

  • timeout (float | None)

Return type:

bool

start(spec: plesty.server.model.device.DeviceSpec, python: pathlib.Path, wait: bool = True, timeout: float | None = None, package: str = '', version: str = '') → plesty.server.model.process.ProcessStatus

Launch the server of spec with the interpreter python.

Parameters:

  • spec (plesty.server.model.device.DeviceSpec) – Module, arguments, environment and port to run with.

  • python (pathlib.Path) – Interpreter of the environment the module is installed in.

  • wait (bool) – Block until the server answers a probe (or fails).

  • timeout (float | None) – Readiness wait in seconds; defaults to ready_timeout.

  • package (str) – Distribution name, stored on the record for listings.

  • version (str) – Installed version, likewise.

Returns:

The status after launch — running when it came up, starting when wait is off or it has not answered yet, exited when it died before answering (detail carries the log tail).

Raises:
Return type:

plesty.server.model.process.ProcessStatus

_start_locked(spec: plesty.server.model.device.DeviceSpec, python: pathlib.Path, wait: bool, timeout: float | None, package: str, version: str) → plesty.server.model.process.ProcessStatus
Parameters:
Return type:

plesty.server.model.process.ProcessStatus

stop(name: str, timeout: float | None = None, note: collections.abc.Callable[[str], None] | None = None) → bool

Interrupt the server of name, escalating to a kill if it lingers.

Interrupt → wait timeout → terminate → wait 3 s → kill tree.

note, when given, receives one line about how the server ended, in the cases where that is worth saying — today, a server the stop killed before its shutdown ran, whose device is therefore still held. It has to be read here: _drop_record() removes the exit file and the record, and status() returns early once the record is gone, so after a stop there is nothing left for anyone to observe. That made the note unreachable on exactly the path an operator takes — press Stop, then ask why the device never disconnected.

Parameters:

  • name (str) – Device instance name.

  • timeout (float | None) – Seconds to wait before escalating.

  • note (collections.abc.Callable[[str], None] | None) – Receives a line about an ending worth reporting.

Returns:

True when a process was stopped, False when only a stale record was cleared.

Raises:

NotRunning – When nothing (not even a stale record) is known.

Return type:

bool

restart(spec: plesty.server.model.device.DeviceSpec, python: pathlib.Path, wait: bool = True, timeout: float | None = None) → plesty.server.model.process.ProcessStatus

Stop the server of spec if it runs, then start() it again.

Parameters:
Return type:

plesty.server.model.process.ProcessStatus

relaunch(record: plesty.server.model.process.ProcessRecord, wait: bool = True) → plesty.server.model.process.ProcessStatus

Start the server again exactly as record says it was started.

Used after a host-tier field test: the interpreter, arguments and environment of the interrupted launch are what the bench expects back.

Parameters:
Return type:

plesty.server.model.process.ProcessStatus

stop_all(names: list[str] | None = None) → list[str]

Stop every known (or the given) server; returns the names stopped.

Parameters:

names (list[str] | None)

Return type:

list[str]

tail(name: str, lines: int = 50) → str

The last lines lines of the log of name ("" when no log).

Parameters:

  • name (str)

  • lines (int)

Return type:

str

clear_log(name: str) → int

Start a new, empty log for name; returns the size of the one set aside.

Nothing is truncated or deleted. Truncating a file a running server writes left it writing at its old offset: the gap filled with NULs, and an 11.7 MB log with no line breaks hung the console, the agent and every reader (#23). The old file stays beside the new one.

A running server’s waiter is asked to write to the new file (request_switch()). One started by an older bench hands the server the file itself and cannot switch, so it is refused with what to do instead.

Parameters:

name (str) – The device whose log to clear.

Returns:

The size of the log set aside; 0 when there was none.

Raises:

SupervisorError – When the server is running and cannot switch, or did not take the new file.

Return type:

int

_wait_exit(name: str, timeout: float) → bool
Parameters:

  • name (str)

  • timeout (float)

Return type:

bool

exception plesty.server.services.SupervisorError

Bases: RuntimeError

A start or stop request could not be carried out.

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

plesty.server.services.probe(port: int, timeout_ms: int = 2000, host_addr: str = PROBE_HOST) → bool

Ask the server on port to describe itself.

Returns:

True when a plesty-lib server answered status: ok in time.

Parameters:

  • port (int)

  • timeout_ms (int)

  • host_addr (str)

Return type:

bool

plesty.server.services.request(port: int, message: dict[str, Any], timeout_ms: int = 5000, host_addr: str = PROBE_HOST) → dict[str, Any]

Send one request to the server on port and return its reply.

The same DEALER socket probe() uses, with the message left to the caller: this is how a developer asks a running device server anything its plesty-lib command solver understands, without a client library.

Raises:

SupervisorError – When nobody answers in time, or the answer is not JSON the wire format allows.

Parameters:

  • port (int)

  • message (dict[str, Any])

  • timeout_ms (int)

  • host_addr (str)

Return type:

dict[str, Any]