plesty.server.presenter.bench

The local bench: one object that composes the model and the services.

Bench is what the CLI and the GUI talk to. It owns a Home, a Fleet, and the four services, and answers the operator’s questions — what is declared, what is installed, what is running — and carries out the operator’s intents — install, start, stop, field-test, bring the fleet up. Anything that takes longer than a request/reply (an install, a field test) it starts as a job by re-invoking plesty-server _job … through the detached launcher (Bench.install(), Bench.field_test()); the synchronous twins (install_now(), field_test_now()) are what those jobs run.

Attributes

Say

Exceptions

BenchError

An operation could not be carried out as asked.

Classes

`DeviceView`	What the operator sees about one device instance.
`SourceOptions`	What a declaration form can offer for a package's source.
`Bench`	The local bench.

Functions

`_example_blocks`(→ dict[str, str])	Each `KEY=` line of an example with the comment lines directly above it, keyed by KEY.
`release_tag`(→ bool)	Whether ref names a release (`v0.3.1`) rather than a branch.
`module_spec`(→ plesty.server.model.DeviceSpec)	A declaration standing in for a bare module install.
`_stderr`(→ None)
`_now`(→ str)

Module Contents

plesty.server.presenter.bench.Say

exception plesty.server.presenter.bench.BenchError

Bases: RuntimeError

An operation could not be carried out as asked.

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

class plesty.server.presenter.bench.DeviceView

What the operator sees about one device instance.

Variables:

spec – The fleet declaration.
installed – The installed version that serves it, or None.
status – The server’s observed state.
field_test_blocked – Why a field test cannot run, or None when it can.
info – The package-level record (repository, project page, latest release).

spec: plesty.server.model.DeviceSpec

installed: plesty.server.model.InstalledVersion | None

status: plesty.server.model.ProcessStatus

field_test_blocked: str | None = None

info: plesty.server.model.PackageInfo | None = None

property name: str

The instance name.

Return type:: str

property source: str

the requirement plus the commit for a checkout.

Type:: One line
Return type:: str

class plesty.server.presenter.bench.SourceOptions

What a declaration form can offer for a package’s source.

Variables:

versions – Release versions known for the package (installed on this bench, then the catalogue’s), newest first.
refs – Git refs known for it (installed checkouts, the catalogue’s latest tag, the usual branches).
git_urls – Repository URLs known for it.

versions: tuple[str, Ellipsis] = ()

refs: tuple[str, Ellipsis] = ()

git_urls: tuple[str, Ellipsis] = ()

class plesty.server.presenter.bench.Bench(home: plesty.server.model.Home, fleet: plesty.server.model.Fleet | None = None, manager: plesty.server.model.DeviceManager | None = None, installer: plesty.server.services.Installer | None = None, jobs: plesty.server.services.JobRunner | None = None, supervisor: plesty.server.services.Supervisor | None = None, say: Say | None = None)

The local bench.

Compose a bench.

Parameters:

home (plesty.server.model.Home) – The bench home.
fleet (plesty.server.model.Fleet | None) – The fleet file, or None for a bench with no declarations (instances are then declared through declare()).
manager (plesty.server.model.DeviceManager | None) – Installed-version inventory; default on home.
installer (plesty.server.services.Installer | None) – Environment creator; default on home and manager.
jobs (plesty.server.services.JobRunner | None) – Job runner; default on home.
supervisor (plesty.server.services.Supervisor | None) – Server supervisor; default on home.
say (Say | None) – Where progress lines go; default: stderr.

home

fleet

manager

installer

jobs

supervisor

_say: Say

Build a bench from the environment: the home, and the fleet file it finds.

Parameters:

home (str | os.PathLike[str] | None) – Explicit home directory (--home); else the default.
fleet_path (str | os.PathLike[str] | None) – Explicit fleet file (-f); else ./fleet.yaml, then <home>/fleet.yaml, then none.
say (Say | None) – Progress sink.

Return type:

Bench

say(line: str) → None

Emit a progress line.

Parameters:: line (str)
Return type:: None

say_to(say: Say) → None

Send progress lines to say from now on.

The console does this at start-up. Without it these lines go to stderr, which a windowed process on Windows discards outright — so “port 5551 is in use — moving to 5553” was written to nowhere while the fleet file was quietly edited.

Parameters:: say (Say)
Return type:: None

spec(name: str) → plesty.server.model.DeviceSpec

The declaration of name.

Raises:: BenchError – When the fleet does not declare it.
Parameters:: name (str)
Return type:: plesty.server.model.DeviceSpec

declare(spec: plesty.server.model.DeviceSpec, save: bool = True) → plesty.server.model.DeviceSpec

Add or replace spec in the fleet, allocating a port when it names none.

Parameters:

spec (plesty.server.model.DeviceSpec) – The declaration.
save (bool) – Write the fleet file back (creating one when there is none).

Return type:

plesty.server.model.DeviceSpec

forget(name: str, save: bool = True) → None

Remove name from the fleet (no error when absent).

Parameters:

name (str)
save (bool)

Return type:

None

_save_fleet() → pathlib.Path

Write the fleet back, giving it <home>/fleet.yaml when it has no file yet.

A declaration that is not on disk is invisible to the jobs the console starts — they are separate processes that read the fleet file. So a bench that declares a device gets a fleet file, whether or not anyone asked for one.

Return type:: pathlib.Path

remove(name: str) → bool

Drop name from the fleet, stopping its server first; True when one was stopped.

The environment stays installed (another instance may use it); the .env stays too, in case the device is declared again.

Parameters:: name (str)
Return type:: bool

set_autostart(name: str, flag: bool) → plesty.server.model.DeviceSpec

Include name in (or drop it from) the working fleet; persists to the fleet file.

Parameters:

name (str)
flag (bool)

Return type:

plesty.server.model.DeviceSpec

property working: list[plesty.server.model.DeviceSpec]

the declared devices marked autostart.

Type:: The working fleet
Return type:: list[plesty.server.model.DeviceSpec]

save_fleet_as(path: str | os.PathLike[str], names: collections.abc.Sequence[str] | None = None) → pathlib.Path

Write the working fleet (or names) to path as a new fleet file; returns it.

The file is portable: it carries the declarations, not this bench’s home. Every device written is marked autostart — it is the fleet of that file.

Parameters:

path (str | os.PathLike[str])
names (collections.abc.Sequence[str] | None)

Return type:

pathlib.Path

create_fleet(path: str | os.PathLike[str]) → plesty.server.model.Fleet

Write an empty fleet file at path and make it the current one.

Raises:: BenchError – When path already exists.
Parameters:: path (str | os.PathLike[str])
Return type:: plesty.server.model.Fleet

load_fleet(path: str | os.PathLike[str]) → plesty.server.model.Fleet

Replace the declarations with the fleet file at path (the home stays).

Parameters:: path (str | os.PathLike[str])
Return type:: plesty.server.model.Fleet

remember_fleet() → None

Note the current fleet file as the one to reopen next time (console start-up).

Return type:: None

fleet_text() → str

The current fleet as YAML text (what the fleet-file editor shows).

Return type:: str

replace_fleet_text(text: str) → plesty.server.model.Fleet

Validate text as a fleet file, write it to the fleet path and adopt it.

Raises:: FleetError – When the text is not a valid fleet document.
Parameters:: text (str)
Return type:: plesty.server.model.Fleet

env_file(name: str) → pathlib.Path

The device’s .env file: <home>/run/<name>/.env.

Parameters:: name (str)
Return type:: pathlib.Path

read_env_text(name: str) → str

The .env text of name (empty when there is none).

Parameters:: name (str)
Return type:: str

write_env_text(name: str, text: str) → pathlib.Path

Write the .env of name; returns the path.

Parameters:

name (str)
text (str)

Return type:

pathlib.Path

process_env(spec: plesty.server.model.DeviceSpec) → dict[str, str]

The environment a server or field test of spec runs with.

The bench’s own environment, then the device’s .env file, then the fleet’s env: (explicit in the declaration, so it wins).

Parameters:: spec (plesty.server.model.DeviceSpec)
Return type:: dict[str, str]

catalogue(index_url: str | None = None) → plesty.server.services.Catalogue

The hub catalogue (the fleet’s index_url when it names one).

Parameters:: index_url (str | None)
Return type:: plesty.server.services.Catalogue

install_entry(entry: plesty.server.model.CatalogueEntry, version: str | None = None, ref: str | None = None, repository: bool = False) → plesty.server.model.Job

Install a catalogue module onto this bench as a job; nothing is declared.

The catalogue is a market place: it puts modules on the bench, and the fleet then declares device instances of what is on it. Which is why this touches no fleet file and needs none.

Parameters:

entry (plesty.server.model.CatalogueEntry) – The catalogue module.
version (str | None) – Release to install; None takes the newest release.
ref (str | None) – Ref for a repository install; None takes the entry’s latest tag, and no tag means the default branch.
repository (bool) – Install the repository instead of a release.

Raises:

BenchError – When the entry is not an installable device.

Return type:

plesty.server.model.Job

module_release(package: str) → str | None

The newest release of package on PyPI, or None — one network call.

What the console asks before it offers a source: a release install needs nothing but uv, a repository install needs git.

Parameters:: package (str)
Return type:: str | None

source_options(package: str, fetch: bool = False) → SourceOptions

Versions, refs and repository URLs the bench knows for package.

Reads the inventory and the fleet; the catalogue only when it has already been fetched — unless fetch is set, which asks the hub (a network call: callers on a GUI thread do that on a worker first, so later calls find the cache).

Parameters:

package (str)
fetch (bool)

Return type:

SourceOptions

seed_env(name: str, installed: plesty.server.model.InstalledVersion | None = None) → pathlib.Path | None

Bring the device’s .env in line with the module’s .env.example.

On the first install the example is copied verbatim — comments and all — as the .env (an empty or missing file counts as “first”). On a later install of another version or ref, keys the new example has and the .env lacks are appended together with the example’s own comment lines above them; the operator’s values are never touched. A checkout’s own .env.example is used; a PyPI install (no repository on disk) gets the generic example shipped with plesty-server.

Returns:

The written path, or None when nothing had to change.

Parameters:

name (str)
installed (plesty.server.model.InstalledVersion | None)

Return type:

pathlib.Path | None

env_example(name: str, installed: plesty.server.model.InstalledVersion | None = None) → str

The .env.example text that applies to device name.

A checkout’s own file first; for a release install (no repository on disk) the file is fetched from the module’s repository as the catalogue names it — the wheel does not carry it; and the generic example shipped with plesty-server when neither is reachable.

Parameters:

name (str)
installed (plesty.server.model.InstalledVersion | None)

Return type:

str

_remote_env_example(package: str) → str | None

.env.example from the module’s repository (GitLab raw URL), or None.

The repository comes from the package record; the catalogue is asked only when the bench has never learned it.

Parameters:: package (str)
Return type:: str | None

_startable_port(spec: plesty.server.model.DeviceSpec) → plesty.server.model.DeviceSpec

spec with a port nothing else is holding; declares the choice.

A port the bench allocated is not a promise it can keep against another process on the machine — a second bench allocates from the same range, and a server left over from an earlier start holds what it holds. When it is taken the device cannot serve a client there whatever we do, so it moves and the fleet file records where it went.

A port the declaration names — --tcp-port in args, or an env value the device itself reads — is a different thing: that number is passed to the device, an operator chose it, and moving it silently would start the server somewhere its own arguments deny. That clash is reported instead.

Parameters:: spec (plesty.server.model.DeviceSpec)
Return type:: plesty.server.model.DeviceSpec

_with_port(spec: plesty.server.model.DeviceSpec) → plesty.server.model.DeviceSpec

Give spec a port of its own unless it asked for one.

Parameters:: spec (plesty.server.model.DeviceSpec)
Return type:: plesty.server.model.DeviceSpec

installed(name: str) → plesty.server.model.InstalledVersion | None

The installed version serving name, or None.

Parameters:: name (str)
Return type:: plesty.server.model.InstalledVersion | None

require_installed(name: str) → plesty.server.model.InstalledVersion

Like installed() but raises BenchError when absent.

Parameters:: name (str)
Return type:: plesty.server.model.InstalledVersion

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

The server state of name.

Parameters:

name (str)
probe (bool)

Return type:

plesty.server.model.ProcessStatus

device(name: str, probe: bool = True) → DeviceView

Declaration, installation and state of name in one view.

Parameters:

name (str)
probe (bool)

Return type:

DeviceView

devices(probe: bool = True) → list[DeviceView]

Every declared device, in fleet order.

Parameters:: probe (bool)
Return type:: list[DeviceView]

log(name: str, lines: int = 50) → str

The tail of the server log of name.

Parameters:

name (str)
lines (int)

Return type:

str

reports(name: str) → list[plesty.server.model.FieldTestReport]

Published field-test reports of name (git installs only).

Parameters:: name (str)
Return type:: list[plesty.server.model.FieldTestReport]

install_now(name: str, force: bool = False, job_id: str | None = None) → plesty.server.model.InstalledVersion

Install what name declares, in the foreground; returns the record.

Parameters:

name (str)
force (bool)
job_id (str | None)

Return type:

plesty.server.model.InstalledVersion

update_now(name: str, ref: str | None = None, job_id: str | None = None) → plesty.server.model.InstalledVersion

Update the environment of name in the foreground.

A git checkout moves to ref (default: its recorded ref) and re-syncs; a PyPI install is reinstalled. The declaration follows a new ref.

Parameters:

name (str)
ref (str | None)
job_id (str | None)

Return type:

plesty.server.model.InstalledVersion

install(name: str, force: bool = False) → plesty.server.model.Job

Install what the device name declares, as a job; returns it at once.

Parameters:

name (str)
force (bool)

Return type:

plesty.server.model.Job

install_module(package: str, version: str | None = None, git: str | None = None, ref: str | None = None, force: bool = False) → plesty.server.model.Job

Install a module onto the bench as a job — no device, no fleet.

The environment is keyed by the package, never by a device name, so installing one is a market-place act: it stocks the bench, and a declaration later picks from what is in stock.

Parameters:

package (str) – Distribution name (plesty-pm100d).
version (str | None) – Release to install; None takes the newest.
git (str | None) – Repository to install from instead of a release.
ref (str | None) – Branch, tag or commit for git.
force (bool) – Reinstall even when it is already on the bench.

Return type:

plesty.server.model.Job

install_module_now(package: str, version: str | None = None, git: str | None = None, ref: str | None = None, force: bool = False, job_id: str | None = None) → plesty.server.model.InstalledVersion

Install a module in the foreground; returns the record.

Parameters:

package (str)
version (str | None)
git (str | None)
ref (str | None)
force (bool)
job_id (str | None)

Return type:

plesty.server.model.InstalledVersion

learn_repository(package: str) → str

Make sure the bench knows where package lives; returns the URL or "".

The release→tag fallback can only reach a repository the package record names, and a bench that has never seen the module names none — which is exactly the first install of a young module, the case the fallback exists for. The hub is what knows: one lookup, best effort, remembered so it is not asked twice. A bench that cannot reach it simply keeps the release failure it would have had anyway.

Parameters:: package (str)
Return type:: str

update(name: str, ref: str | None = None) → plesty.server.model.Job

Update name as a job; returns it at once.

Parameters:

name (str)
ref (str | None)

Return type:

plesty.server.model.Job

uninstall(name: str) → bool

Stop name if it runs and remove its environment; True when one was removed.

Parameters:: name (str)
Return type:: bool

start(name: str, wait: bool = True, timeout: float | None = None) → plesty.server.model.ProcessStatus

Start the server of name (installing first when nothing serves it).

Raises:

BenchError – When it is not installed and cannot be, or the launch fails.

Parameters:

name (str)
wait (bool)
timeout (float | None)

Return type:

plesty.server.model.ProcessStatus

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

Stop the server of name; True when a process was stopped.

Parameters:

name (str)
timeout (float | None)

Return type:

bool

restart(name: str, wait: bool = True) → plesty.server.model.ProcessStatus

Stop (if running) and start the server of name.

Parameters:

name (str)
wait (bool)

Return type:

plesty.server.model.ProcessStatus

up(names: collections.abc.Sequence[str] | None = None, install: bool = True, wait: bool = True) → list[plesty.server.model.ProcessStatus]

Install (unless install is off) and start the autostart devices, or names.

A device that is already running is left alone. Returns one status per device, in fleet order.

Parameters:

names (collections.abc.Sequence[str] | None)
install (bool)
wait (bool)

Return type:

list[plesty.server.model.ProcessStatus]

down(names: collections.abc.Sequence[str] | None = None) → list[str]

Stop everything running (or names); returns the names stopped.

fleet up starts the working fleet, so down starts there — but a device declared autostart: false and started by hand is running all the same, and stopping only the working fleet left it up and called the bench down. Anything with a launch record is included.

Parameters:: names (collections.abc.Sequence[str] | None)
Return type:: list[str]

run(stop: threading.Event, names: collections.abc.Sequence[str] | None = None, interval: float = 5.0, max_sweeps: int | None = None, leave_running: bool = False, max_backoff: float = 60.0) → int

Bring the fleet up and keep it up until stop is set (headless bench mode).

Every interval seconds each device is inspected; one found exited is started again, after a backoff that doubles per consecutive crash (capped at max_backoff) and resets once it has answered a probe. Devices stopped on purpose (no record) are left alone: the loop keeps up what it started, it does not fight the operator.

Parameters:

stop (threading.Event) – Set to end the loop.
names (collections.abc.Sequence[str] | None) – Devices to keep up; default: the autostart ones.
interval (float) – Seconds between sweeps.
max_sweeps (int | None) – Stop after this many sweeps (tests).
leave_running (bool) – Do not bring the servers down when the loop ends.
max_backoff (float) – Longest wait before a restart.

Returns:

The number of restarts performed.

Return type:

int

field_test_now(name: str, run: plesty.server.model.FieldTestRun, say: Say | None = None) → int

Run the field test of name in the foreground; returns the script’s exit code.

Host tier: the server is stopped for the run and relaunched afterwards exactly as it was (unless run.keep_server). Client tier: the server is started when it is not up, and the address defaults to it.

Raises:

BenchError – When name is not a git install with that tier’s script.

Parameters:

name (str)
run (plesty.server.model.FieldTestRun)
say (Say | None)

Return type:

int

field_test_options(name: str, tier: str = 'host') → dict[str, list[str]]

What the device’s field-test script accepts, from its --help.

Returns:

{"options": [...], "gates": [...]} — the long options and the gate choices (the standard gates when the script does not list them); both empty when the script cannot be asked.

Parameters:

name (str)
tier (str)

Return type:

dict[str, list[str]]

field_test_availability(name: str, tier: str = 'host') → str | None

Why the field test of name cannot run, or None when it can.

The reason is what a tooltip shows: not installed, a release install (no repository on disk), or a checkout without the tier’s script.

Parameters:

name (str)
tier (str)

Return type:

str | None

field_test(name: str, run: plesty.server.model.FieldTestRun) → plesty.server.model.Job

Run the field test of name as a job; returns it at once.

Parameters:

name (str)
run (plesty.server.model.FieldTestRun)

Return type:

plesty.server.model.Job

_field_test_command(run: plesty.server.model.FieldTestRun, spec: plesty.server.model.DeviceSpec, python: pathlib.Path, checkout: pathlib.Path) → list[str]

Parameters:

run (plesty.server.model.FieldTestRun)
spec (plesty.server.model.DeviceSpec)
python (pathlib.Path)
checkout (pathlib.Path)

Return type:

list[str]

static _script_help(python: pathlib.Path, script: pathlib.Path) → str

The --help text of a field-test script ("" when it cannot be asked).

Parameters:

python (pathlib.Path)
script (pathlib.Path)

Return type:

str

classmethod _script_options(python: pathlib.Path, script: pathlib.Path) → set[str] | None

The options a field-test script accepts, read from its --help.

Scaffolds differ between SDK releases (--address now, --port/ --mock before); asking the script is cheaper than guessing wrong. None when it cannot be asked (the current scaffold is assumed).

Parameters:

python (pathlib.Path)
script (pathlib.Path)

Return type:

set[str] | None

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

The job job_id, up to date, or None.

Parameters:: job_id (str)
Return type:: plesty.server.model.Job | None

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

Jobs on record, newest first.

Parameters:

name (str | None)
kind (str | None)

Return type:

list[plesty.server.model.Job]

job_log(job_id: str, offset: int = 0) → tuple[str, int]

Job log from byte offset; returns text and the next offset.

Parameters:

job_id (str)
offset (int)

Return type:

tuple[str, int]

cancel(job_id: str) → bool

Cancel a running job.

Parameters:: job_id (str)
Return type:: bool

_job(name: str, kind: str, argv: list[str], detail: dict[str, object] | None = None, fleet_needed: bool = True) → plesty.server.model.Job

Start plesty-server _job <argv> against this bench as a job.

Raises:

BenchError – When the job reads declarations this bench has not written anywhere — a job is another process, and it opens the fleet file.

Parameters:

name (str)
kind (str)
argv (list[str])
detail (dict[str, object] | None)
fleet_needed (bool)

Return type:

plesty.server.model.Job

plesty.server.presenter.bench._example_blocks(template: str) → dict[str, str]

Each KEY= line of an example with the comment lines directly above it, keyed by KEY.

Parameters:: template (str)
Return type:: dict[str, str]

plesty.server.presenter.bench.release_tag(ref: str) → bool

Whether ref names a release (v0.3.1) rather than a branch.

Parameters:: ref (str) – A git ref as the inventory or the catalogue reports it.
Return type:: bool

plesty.server.presenter.bench.module_spec(package: str, version: str | None = None, git: str | None = None, ref: str | None = None) → plesty.server.model.DeviceSpec

A declaration standing in for a bare module install.

The installer works from a DeviceSpec because a device install is the common case; an environment is keyed by (package, version) or (package, ref) either way, so the name is only along for the ride and the package’s own is the honest one.

Parameters:

package (str)
version (str | None)
git (str | None)
ref (str | None)

Return type:

plesty.server.model.DeviceSpec

plesty.server.presenter.bench._stderr(line: str) → None

Parameters:: line (str)
Return type:: None

plesty.server.presenter.bench._now() → str

Return type:: str