plesty.server.presenter

Presenter layer: binds views to the model through the services.

Presenters receive intents from views (or the CLI), drive the model and the services (install, start, stop, field test) and hand back model objects the views render. They are the one place that composes several services into one operation — install then start, stop, field-test, relaunch — and they never import Qt, so the CLI and the GUI share them unchanged.

Submodules

plesty.server.presenter.bench

Exceptions

BenchError

An operation could not be carried out as asked.

Classes

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

Package Contents

class plesty.server.presenter.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

exception plesty.server.presenter.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.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.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] = ()