plesty.server.model

Model layer: fleet configuration, device catalogue, installer and supervisor state.

Pure Python — no Qt imports. Everything the presenter needs to know about the bench lives here and is testable without a display.

Submodules

Exceptions

`DeviceManagerError`	An installed-version record is missing or inconsistent.
`FleetError`	A fleet file or device declaration is malformed.
`FieldTestError`	A field test cannot be set up as asked.
`JobError`	A job record does not exist or cannot be read.

Classes

`CatalogueEntry`	One module of the hub catalogue.
`DeviceSpec`	One device server as the fleet declares it.
`DeviceManager`	The installed-module inventory under a `Home`.
`InstalledPackage`	All installed versions of one device module.
`InstalledVersion`	One version of one device module present on the bench.
`PackageInfo`	What the bench knows about a module as a project — independent of any install.
`FieldTestReport`	What a field test published — the summary a bench needs, not the whole document.
`FieldTestRun`	One field-test request.
`GateResult`	One gate of a published report.
`Fleet`	A parsed fleet file.
`Home`	Layout of the plesty-server home directory.
`Job`	One unit of long-running work.
`JobStore`	Job records and logs under a `Home`.
`ProcessRecord`	What the supervisor persists about a started server.
`ProcessStatus`	The observed state of one device server.

Functions

`parse_catalogue`(→ list[CatalogueEntry])	Every entry of a `/api/modules/` document (a mapping with `modules` or a bare list).
`module_for`(→ str)	The import path served by `python -m` for a PLESTY distribution.
`parse_env`(→ dict[str, str])	Parse dotenv text into a mapping (later keys win).
`read_env`(→ dict[str, str])	Parse the file at path; empty when it does not exist.
`render_env`(→ str)	Render a mapping as dotenv text (one `KEY=VALUE` per line, quoted when needed).
`field_test_script`(→ pathlib.Path \| None)	The tier's script inside checkout, or `None` when the module has none.
`find_reports`(→ list[FieldTestReport])	Every published field-test report under `<checkout>/reports/`, newest first.
`gates_in_help`(→ tuple[str, Ellipsis])	The gate names a script's `--help` offers for `--gates`, in its order; empty if none.
`options_in_help`(→ set[str])	The long options an argparse `--help` text mentions (`--address`, `--mock` …).
`find_fleet`(→ pathlib.Path \| None)	Locate the fleet file a command should consult.
`load_fleet`(→ Fleet)	Read and validate a fleet file.
`save_fleet`(→ pathlib.Path)	Write a fleet file.
`default_home`(→ pathlib.Path)	The home directory the environment or the packaged config designates.
`tail_file`(→ str)	The last lines lines of path, read from the end in block steps.
`allocate_port`(→ int)	The lowest free port from base up, skipping taken and the agent's.
`declares_port`(→ bool)	Whether args or env name the port themselves.
`port_from`(→ int)	Resolve the TCP port a server binds from its args and environment.
`port_is_free`(→ bool)	Whether port can actually be bound right now.
`port_listening`(→ bool)	Whether something accepts TCP connections on port right now.

Package Contents

plesty.server.model.DEFAULT_INDEX_URL = 'https://hub.plesty.net/api/modules/'

plesty.server.model.REQUIRED_STANDARD = 'quantum'

class plesty.server.model.CatalogueEntry

One module of the hub catalogue.

Variables:

name – Distribution name (plesty-pm100d).
slug – Project slug.
namespace – GitLab path (plesty/hub/devices/thorlab/pm100d).
description – One line.
category – devices, analyzers or experiments.
vendor – Vendor group for devices (thorlab), else empty.
repo_url – Clone URL — what a git install uses.
web_url – Project page.
version – Latest release version, or empty.
latest_tag – Latest tag (v0.2.1), or empty.
standard – Compliance standard the module pins (quantum), or empty.
checked – Whether a plesty check report is published.
gates_ok – Whether that report passed.
gates_passed – Gates passed in it.
gates_failed – Gates failed in it.
field_tests – Number of published field-test reports.

name: str

slug: str = ''

namespace: str = ''

description: str = ''

category: str = ''

vendor: str = ''

repo_url: str = ''

web_url: str = ''

version: str = ''

latest_tag: str = ''

standard: str = ''

checked: bool = False

gates_ok: bool = False

gates_passed: int = 0

gates_failed: int = 0

field_tests: int = 0

property installable: bool

a device at the required standard with a repository.

Type:: Whether the bench offers it
Return type:: bool

property ref: str | None

the latest tag, else the default branch.

Type:: The git ref a fresh install checks out
Return type:: str | None

property quality: str

ok, failing, unchecked.

Type:: One word for listings
Return type:: str

matches(query: str) → bool

Case-insensitive substring match on name, slug, vendor and description.

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

classmethod from_registry(raw: collections.abc.Mapping[str, Any]) → CatalogueEntry

Build from one element of the registry’s modules list.

Parameters:: raw (collections.abc.Mapping[str, Any])
Return type:: CatalogueEntry

plesty.server.model.parse_catalogue(document: Any) → list[CatalogueEntry]

Every entry of a /api/modules/ document (a mapping with modules or a bare list).

Parameters:: document (Any)
Return type:: list[CatalogueEntry]

class plesty.server.model.DeviceSpec

One device server as the fleet declares it.

Variables:

name – Instance name — the venv, log and state key; unique per bench.
package – PyPI distribution of the device module.
version – Pinned release, or None for the newest one at install time.
git – Repository URL to install from instead of PyPI, or None.
ref – Branch, tag or commit for a git source; None = v<version> when a version is given, else the default branch.
module – Import path served with python -m; derived from package.
args – Arguments appended to python -m <module>.
env – Environment set for the server process on top of the bench’s own.
port – TCP port the server binds; probed for liveness.
autostart – Whether fleet up/fleet run start it.

name: str

package: str

version: str | None = None

git: str | None = None

ref: str | None = None

module: str = ''

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

env: collections.abc.Mapping[str, str]

port: int = 0

autostart: bool = True

__post_init__() → None

Fill the derived fields and validate the declaration.

Return type:: None

property port_is_explicit: bool

Whether this device asked for its port rather than taking the default.

True once the bench has allocated one and written it down, so the port a client connected to yesterday is the port it finds today.

Return type:: bool

property launch_args: tuple[str, Ellipsis]

The arguments python -m <module> is started with.

port is what the supervisor probes; unless the args already name a port, it is also handed to the server as --tcp-port, so what is probed and what is bound cannot drift. An environment value (*_TCP_PORT — often 5555 straight from a module’s .env.example) is deliberately not trusted here: an explicit flag beats the environment in every hub module, and the bench owns the ports.

Return type:: tuple[str, Ellipsis]

property git_ref: str | None

The ref a git install checks out (ref, else v<version>, else None).

Return type:: str | None

property requirement: str

The pip requirement string.

plesty-pm100d==0.2.1 for a PyPI source; plesty-pm100d @ git+<url>@v0.2.1 for a git source.

Return type:: str

classmethod from_mapping(name: str, raw: collections.abc.Mapping[str, Any]) → DeviceSpec

Build a spec from one devices: entry of a fleet file.

Parameters:

name (str) – The mapping key.
raw (collections.abc.Mapping[str, Any]) – The entry’s fields.

Raises:

FleetError – On unknown keys or wrong value types.

Return type:

DeviceSpec

to_mapping() → dict[str, Any]

The fleet-file form of this spec (derived defaults omitted).

Return type:: dict[str, Any]

plesty.server.model.module_for(package: str) → str

The import path served by python -m for a PLESTY distribution.

Parameters:: package (str) – PyPI distribution name, e.g. plesty-pm100d.
Returns:: plesty.pm100d — the platform’s flat namespace convention.
Return type:: str

plesty.server.model.SOURCE_GIT = 'git'

plesty.server.model.SOURCE_PYPI = 'pypi'

class plesty.server.model.DeviceManager(home: plesty.server.model.home.Home)

The installed-module inventory under a Home.

Bind to a home.

Parameters:: home (plesty.server.model.home.Home) – The bench home.

home

Write down that package version is installed; returns the record.

Parameters:

package (str) – Distribution name.
version (str) – Installed release.
source (str) – SOURCE_PYPI or SOURCE_GIT.
git (str | None) – Repository URL for a git source.
ref (str | None) – Requested ref for a git source.
commit (str | None) – Checked-out commit for a git source.
python (str) – Interpreter version of the environment.
job_id (str | None) – The install job.
path (str | os.PathLike[str] | None) – Environment location when not the default package dir.

Return type:

InstalledVersion

save(installed: InstalledVersion) → pathlib.Path

Write installed to its record file; returns the path.

Parameters:: installed (InstalledVersion)
Return type:: pathlib.Path

get(package: str, slot: str) → InstalledVersion | None

The record of package at slot (a version, or git-<ref>), or None.

Parameters:

package (str)
slot (str)

Return type:

InstalledVersion | None

require(package: str, slot: str) → InstalledVersion

Like get() but raises DeviceManagerError when absent.

Parameters:

package (str)
slot (str)

Return type:

InstalledVersion

package(package: str) → InstalledPackage

Every installed version of package (empty when none).

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

list() → list[InstalledPackage]

Every package with at least one installed version, by name.

Return type:: list[InstalledPackage]

info(package: str) → PackageInfo

The package-level record of package (empty fields when nothing is known).

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

remember(package: str, **fields: str) → PackageInfo

Merge non-empty fields into the package-level record; returns it.

Parameters:

package (str)
fields (str)

Return type:

PackageInfo

env_dir(installed: InstalledVersion) → pathlib.Path

The environment directory of installed (its path, else the package dir).

Parameters:: installed (InstalledVersion)
Return type:: pathlib.Path

resolve(spec: plesty.server.model.device.DeviceSpec) → InstalledVersion | None

The installed version that serves spec, or None when one must be installed.

The most recent install among those that InstalledVersion.satisfies() the spec wins — for an unpinned PyPI spec that is the newest install, for a git branch the last pull.

Parameters:: spec (plesty.server.model.device.DeviceSpec)
Return type:: InstalledVersion | None

forget(package: str, slot: str, remove_env: bool = False) → bool

Drop the record of package at slot; True when it existed.

Parameters:

package (str) – Distribution name.
slot (str) – Version (release) or git-<ref> (checkout).
remove_env (bool) – Also delete a release’s environment directory.

Return type:

bool

exception plesty.server.model.DeviceManagerError

Bases: ValueError

An installed-version record is missing or inconsistent.

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

class plesty.server.model.InstalledPackage

All installed versions of one device module.

Variables:

package – Distribution name.
versions – Records keyed by slot (version, or git-<ref>), oldest install first.

package: str

versions: collections.abc.Mapping[str, InstalledVersion]

__iter__() → collections.abc.Iterator[InstalledVersion]

Iterate over the versions, oldest install first.

Return type:: collections.abc.Iterator[InstalledVersion]

__len__() → int

Number of installed versions.

Return type:: int

__contains__(version: object) → bool

Whether version (a slot, or a release version) is installed.

Parameters:: version (object)
Return type:: bool

property latest: InstalledVersion | None

The most recently installed version, or None when empty.

Return type:: InstalledVersion | None

class plesty.server.model.InstalledVersion

One version of one device module present on the bench.

Variables:

package – PyPI distribution name, e.g. plesty-pm100d.
version – The installed release as reported by the environment.
source – SOURCE_PYPI or SOURCE_GIT.
git – Repository URL for a git source, else None.
ref – Branch, tag or commit that was requested for a git source.
commit – The commit actually checked out for a git source.
installed_utc – ISO timestamp of the install.
python – Version of the interpreter in the environment.
job_id – The install job that produced it, when known.
path – Where the environment lives when it is not the default <home>/packages/<package>/<version>/ — a git checkout, whose .venv is the environment.

package: str

version: str

source: str = 'pypi'

git: str | None = None

ref: str | None = None

commit: str | None = None

installed_utc: str = ''

python: str = ''

job_id: str | None = None

path: str = ''

__post_init__() → None

Validate the record.

Return type:: None

property slot: str

the version for a release, git-<ref> for a checkout.

A checkout moves — its version changes with every pull — so it is kept by the ref it tracks, and two checkouts at the same commit (two refs) never share a record.

Type:: What identifies the environment within the package
Return type:: str

property key: tuple[str, str]

(package, slot) — what identifies the environment.

Return type:: tuple[str, str]

satisfies(spec: plesty.server.model.device.DeviceSpec) → bool

Whether this install is what spec asks for.

A PyPI spec is satisfied by the pinned version (or any PyPI version when unpinned); a git spec by an install from the same repository at the same ref (or v<version>), a commit hash matching by prefix.

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

to_dict() → dict[str, Any]

JSON form.

Return type:: dict[str, Any]

classmethod from_dict(data: collections.abc.Mapping[str, Any]) → InstalledVersion

Rebuild a record from to_dict() output (extra keys ignored).

Parameters:: data (collections.abc.Mapping[str, Any])
Return type:: InstalledVersion

class plesty.server.model.PackageInfo

What the bench knows about a module as a project — independent of any install.

Learned from the catalogue (adopt) or from a git install, kept in <home>/packages/<package>/package.json and never asked again: the repository a module comes from, its project page, its latest release.

Variables:

package – Distribution name.
repo_url – Clone URL of the module’s repository.
web_url – Project page (GitLab), "" when unknown.
latest_tag – Latest release tag seen (v0.2.1), "" when none.
version – Latest release version seen, "" when none.
vendor – Hub vendor group, "" when unknown.
description – One line, "" when unknown.

package: str

repo_url: str = ''

web_url: str = ''

latest_tag: str = ''

version: str = ''

vendor: str = ''

description: str = ''

to_dict() → dict[str, Any]

JSON form.

Return type:: dict[str, Any]

classmethod from_dict(data: collections.abc.Mapping[str, Any]) → PackageInfo

Rebuild from to_dict() output (extra keys ignored).

Parameters:: data (collections.abc.Mapping[str, Any])
Return type:: PackageInfo

merged(**fields: str) → PackageInfo

A copy with the non-empty fields replacing the current values.

Parameters:: fields (str)
Return type:: PackageInfo

plesty.server.model.ENV_FILENAME = '.env'

plesty.server.model.parse_env(text: str) → dict[str, str]

Parse dotenv text into a mapping (later keys win).

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

plesty.server.model.read_env(path: str | os.PathLike[str]) → dict[str, str]

Parse the file at path; empty when it does not exist.

Parameters:: path (str | os.PathLike[str])
Return type:: dict[str, str]

plesty.server.model.render_env(values: collections.abc.Mapping[str, str]) → str

Render a mapping as dotenv text (one KEY=VALUE per line, quoted when needed).

Parameters:: values (collections.abc.Mapping[str, str])
Return type:: str

exception plesty.server.model.FleetError

Bases: ValueError

A fleet file or device declaration is malformed.

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

plesty.server.model.JOB_KIND = 'field-test'

plesty.server.model.REPORT_BASENAME = 'reports/field-test-run'

plesty.server.model.STANDARD_GATES: tuple[str, Ellipsis] = ('discovery', 'connect_lifecycle', 'param_roundtrip', 'param_constraints', 'functions',...

plesty.server.model.TIER_ARTIFACTS: collections.abc.Mapping[str, str]

plesty.server.model.TIER_SCRIPTS: collections.abc.Mapping[str, str]

exception plesty.server.model.FieldTestError

Bases: plesty.server.model.errors.FleetError

A field test cannot be set up as asked.

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

class plesty.server.model.FieldTestReport

What a field test published — the summary a bench needs, not the whole document.

Variables:

path – The field-test*.json file.
tier – host or client, from the file name.
instrument – The reports/<instrument>/ directory name.
module – Module name from the document.
version – Module version the run tested.
ok – Overall outcome.
passed – Gates passed.
failed – Gates failed.
skipped – Gates skipped.
gates – Per-gate results, in run order.
findings – Count of findings per severity (error/warning/info).
modified_utc – ISO timestamp of the file — when the run ended.

path: pathlib.Path

tier: str

instrument: str

module: str

version: str

ok: bool

passed: int

failed: int

skipped: int

gates: tuple[GateResult, Ellipsis] = ()

findings: collections.abc.Mapping[str, int]

modified_utc: str = ''

property summary: str

pm100d 0.2.1 [host] ok 9/9 style.

Type:: One line
Return type:: str

classmethod from_path(path: pathlib.Path) → FieldTestReport

Read a published report.

Raises:: FieldTestError – When the file is not a field-test document.
Parameters:: path (pathlib.Path)
Return type:: FieldTestReport

class plesty.server.model.FieldTestRun

One field-test request.

Variables:

tier – host (instrument, server must be stopped) or client (running server).
address – --address — host tier: instrument address or mock; client tier: tcp://…; None lets the script use .env.
gates – Run only these gates (None = all).
repetitions – --repetitions for the timing gates (None = script default).
keep_going – Carry on after a failed gate.
keep_server – Host tier: leave a running server alone (only right for mock or another instrument).
extra – Further arguments passed to the script unchanged.

tier: str = 'host'

address: str | None = None

gates: tuple[str, Ellipsis] | None = None

repetitions: int | None = None

keep_going: bool = False

keep_server: bool = False

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

__post_init__() → None

Validate the request.

Return type:: None

property script: str

The script this request runs, relative to the checkout.

Return type:: str

property artifact: str

The published file name this tier writes.

Return type:: str

property needs_instrument: bool

Whether the run opens the instrument itself (host tier, not mock).

Return type:: bool

argv(report_base: str | None = REPORT_BASENAME, supports: collections.abc.Collection[str] | None = None) → list[str]

The script’s arguments for this request.

Parameters:

report_base (str | None) – --report for the host tier’s internal .md/.jsonl pair; None leaves it to the script.
supports (collections.abc.Collection[str] | None) – The option names the script accepts (from its --help), or None to assume the current scaffold. Older scaffolds take --port PORT and --mock where the current one takes --address; the request is spelled in whichever the script understands, and options it lacks are left out.

Return type:

list[str]

command(python: pathlib.Path, checkout: pathlib.Path, supports: collections.abc.Collection[str] | None = None) → list[str]

The full command line: python running the tier’s script in checkout.

Parameters:

python (pathlib.Path) – The checkout’s interpreter.
checkout (pathlib.Path) – The module checkout.
supports (collections.abc.Collection[str] | None) – The script’s option names (see argv()).

Raises:

FieldTestError – When the checkout has no script for this tier.

Return type:

list[str]

to_dict() → dict[str, Any]

Record form (stored on the job’s detail).

Return type:: dict[str, Any]

classmethod from_dict(data: collections.abc.Mapping[str, Any]) → FieldTestRun

Rebuild a request from to_dict() output.

Parameters:: data (collections.abc.Mapping[str, Any])
Return type:: FieldTestRun

class plesty.server.model.GateResult

One gate of a published report.

name: str

status: str

duration_s: float

reason: str

plesty.server.model.field_test_script(checkout: pathlib.Path, tier: str = 'host') → pathlib.Path | None

The tier’s script inside checkout, or None when the module has none.

Raises:

FieldTestError – On an unknown tier.

Parameters:

checkout (pathlib.Path)
tier (str)

Return type:

pathlib.Path | None

plesty.server.model.find_reports(checkout: pathlib.Path, since_utc: str | None = None) → list[FieldTestReport]

Every published field-test report under <checkout>/reports/, newest first.

Parameters:

checkout (pathlib.Path) – The module checkout.
since_utc (str | None) – Only reports modified at or after this ISO timestamp — what a run that started then produced.

Return type:

list[FieldTestReport]

plesty.server.model.gates_in_help(text: str) → tuple[str, Ellipsis]

The gate names a script’s --help offers for --gates, in its order; empty if none.

Parameters:: text (str)
Return type:: tuple[str, Ellipsis]

plesty.server.model.options_in_help(text: str) → set[str]

The long options an argparse --help text mentions (--address, --mock …).

Parameters:: text (str)
Return type:: set[str]

plesty.server.model.FLEET_FILENAME = 'fleet.yaml'

plesty.server.model.FLEET_VERSION = 1

class plesty.server.model.Fleet

A parsed fleet file.

Variables:

devices – Specs keyed by device name, in file order.
home – home: from the file, or None.
index_url – index_url: from the file, or None.
path – Where it was read from, or None for an in-memory fleet.

devices: collections.abc.Mapping[str, plesty.server.model.device.DeviceSpec]

home: pathlib.Path | None = None

index_url: str | None = None

path: pathlib.Path | None = None

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

The spec of device name; KeyError when undeclared.

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

__contains__(name: object) → bool

Whether name is a declared device.

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

__iter__() → Any

Iterate over the specs in file order.

Return type:: Any

__len__() → int

Number of declared devices.

Return type:: int

property autostart: list[plesty.server.model.device.DeviceSpec]

The specs fleet up starts.

Return type:: list[plesty.server.model.device.DeviceSpec]

classmethod from_mapping(raw: collections.abc.Mapping[str, Any], path: pathlib.Path | None = None) → Fleet

Build a fleet from a parsed YAML document.

Parameters:

raw (collections.abc.Mapping[str, Any]) – The document.
path (pathlib.Path | None) – Where it came from, for error messages and path.

Raises:

FleetError – On a wrong schema version or malformed entries.

Return type:

Fleet

to_mapping() → dict[str, Any]

The YAML document form of this fleet.

Return type:: dict[str, Any]

with_device(spec: plesty.server.model.device.DeviceSpec) → Fleet

A copy with spec added or replaced under its name.

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

without_device(name: str) → Fleet

A copy without device name (no error when absent).

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

plesty.server.model.find_fleet(explicit: str | os.PathLike[str] | None, home: pathlib.Path) → pathlib.Path | None

Locate the fleet file a command should consult.

Parameters:

explicit (str | os.PathLike[str] | None) – A path the user gave (-f); returned as-is when set.
home (pathlib.Path) – The bench home, whose fleet.yaml is the last resort after the working directory’s.

Returns:

The first existing candidate, or None when no fleet file exists.

Return type:

pathlib.Path | None

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

Read and validate a fleet file.

Parameters:: path (str | os.PathLike[str]) – The YAML file.
Raises:: FleetError – When the file is missing or malformed.
Return type:: Fleet

plesty.server.model.save_fleet(fleet: Fleet, path: str | os.PathLike[str]) → pathlib.Path

Write a fleet file.

Parameters:

fleet (Fleet) – The fleet to serialize.
path (str | os.PathLike[str]) – Destination file; parents are created.

Returns:

The written path.

Return type:

pathlib.Path

plesty.server.model.HOME_ENV = 'PLESTY_SERVER_HOME'

class plesty.server.model.Home

Layout of the plesty-server home directory.

Variables:: root – The directory everything lives under.

root: pathlib.Path

classmethod resolve(root: str | os.PathLike[str] | None = None) → Home

Build a home from an explicit path or the environment default.

Parameters:: root (str | os.PathLike[str] | None) – Directory to use; None selects default_home().
Return type:: Home

property venvs: pathlib.Path

One virtual environment per installed device, keyed by device name.

Return type:: pathlib.Path

property logs: pathlib.Path

One append-only log per device server.

Return type:: pathlib.Path

property state: pathlib.Path

One JSON record per running server (pid, port, start time).

Return type:: pathlib.Path

property run: pathlib.Path

Working directories the servers run in; plesty-lib writes logs/ there.

Return type:: pathlib.Path

property repos: pathlib.Path

Git checkouts of device modules installed from a repository.

Return type:: pathlib.Path

property jobs: pathlib.Path

One JSON record per long-running job (install, update, field test, command).

Return type:: pathlib.Path

property fleets: pathlib.Path

Where the console keeps fleet files by default (<home>/fleets/).

Return type:: pathlib.Path

property tools: pathlib.Path

Tools the bench installed for itself (the portable git on Windows).

Return type:: pathlib.Path

property packages: pathlib.Path

one environment per (package, version), side by side.

Type:: Installed device modules
Return type:: pathlib.Path

package_dir(package: str, version: str) → pathlib.Path

Environment of package at version.

Parameters:

package (str)
version (str)

Return type:

pathlib.Path

package_record(package: str, version: str) → pathlib.Path

Record file of package at version (beside its environment).

Parameters:

package (str)
version (str)

Return type:

pathlib.Path

property last_fleet: pathlib.Path

Pointer file naming the fleet the console used last (<home>/last-fleet).

Return type:: pathlib.Path

package_info(package: str) → pathlib.Path

The package-level record of package: repository, project page, latest release.

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

checkout(package: str, ref: str | None) → pathlib.Path

Checkout directory of package at ref (git sources only).

Refs may contain / (feat/x); the directory name flattens them.

Parameters:

package (str)
ref (str | None)

Return type:

pathlib.Path

job_file(job_id: str) → pathlib.Path

Record file of job job_id.

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

job_log(job_id: str) → pathlib.Path

Log file of job job_id.

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

venv(name: str) → pathlib.Path

Venv directory of the device name.

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

log(name: str) → pathlib.Path

Log file of the device name.

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

state_file(name: str) → pathlib.Path

Process-state file of the device name.

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

run_dir(name: str) → pathlib.Path

Working directory of the device name’s server process.

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

ensure() → Home

Create the directory tree; returns self for chaining.

Return type:: Home

plesty.server.model.default_home() → pathlib.Path

The home directory the environment or the packaged config designates.

Returns:: $PLESTY_SERVER_HOME when set, else running.home from config.yaml with ~ expanded.
Return type:: pathlib.Path

plesty.server.model.CANCELLED_EXIT = -9

plesty.server.model.LOG_CHUNK = 65536

plesty.server.model.LOST_EXIT = -1

class plesty.server.model.Job

One unit of long-running work.

Variables:

id – <name>-<kind>-<timestamp>, unique per bench.
name – Device instance the job belongs to.
kind – install, update, field-test, command …
command – The command line.
cwd – Working directory.
pid – Process id while it runs.
started_utc – ISO timestamp of the launch.
finished_utc – ISO timestamp of the end, or None while running.
returncode – Exit code once finished (None while running; LOST_EXIT when the process vanished without one).
log – Path of the log file.
detail – Kind-specific data (e.g. the reports a field test wrote).

id: str

name: str

kind: str

command: list[str]

cwd: str

pid: int | None

started_utc: str

finished_utc: str | None = None

returncode: int | None = None

log: str = ''

detail: dict[str, Any]

property running: bool

Whether the job has not finished yet.

Return type:: bool

property state: str

running, done (exit 0), failed (any other exit).

Return type:: str

to_dict() → dict[str, Any]

JSON form, with the derived state included.

Return type:: dict[str, Any]

classmethod from_dict(data: collections.abc.Mapping[str, Any]) → Job

Rebuild a record from to_dict() output (extra keys ignored).

Parameters:: data (collections.abc.Mapping[str, Any])
Return type:: Job

exception plesty.server.model.JobError

Bases: RuntimeError

A job record does not exist or cannot be read.

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

class plesty.server.model.JobStore(home: plesty.server.model.home.Home)

Job records and logs under a Home.

Bind to a home.

Parameters:: home (plesty.server.model.home.Home) – The bench home.

home

create(name: str, kind: str, command: list[str], cwd: str | os.PathLike[str], pid: int | None = None, detail: collections.abc.Mapping[str, Any] | None = None) → Job

Open a new record and return it (state running).

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.
pid (int | None) – Process id when already known.
detail (collections.abc.Mapping[str, Any] | None) – Kind-specific data stored on the record.

Return type:

Job

save(job: Job) → pathlib.Path

Write job to its record file; returns the path.

Parameters:: job (Job)
Return type:: pathlib.Path

get(job_id: str) → Job | None

The record of job_id, or None when unknown or unreadable.

Parameters:: job_id (str)
Return type:: Job | None

require(job_id: str) → Job

Like get() but raises JobError when unknown.

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

list(name: str | None = None, kind: str | None = None) → list[Job]

Jobs on record, newest first, optionally filtered by device and kind.

Parameters:

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

Return type:

list[Job]

finish(job_id: str, returncode: int) → Job

Close the record of job_id with returncode; returns it.

Parameters:

job_id (str)
returncode (int)

Return type:

Job

refresh(job: Job, alive: collections.abc.Callable[[int], bool], exit_code: collections.abc.Callable[[str], int | None]) → Job

Settle a running record against what the process layer knows.

Parameters:

job (Job) – The record to bring up to date (mutated and saved when it ends).
alive (collections.abc.Callable[[int], bool]) – Whether a pid is still running.
exit_code (collections.abc.Callable[[str], int | None]) – The exit code left for a job id, or None when none.

Returns:

job, finished when an exit code exists or the pid is gone (LOST_EXIT then), unchanged while it still runs.

Return type:

Job

log(job_id: str, offset: int = 0, limit: int = LOG_CHUNK) → tuple[str, int]

Read the job log from byte offset.

Parameters:

job_id (str) – The job.
offset (int) – Where the previous read stopped (0 for the start).
limit (int) – Maximum bytes to return.

Returns:

The text read and the offset to continue from. Decoding is lenient — a chunk may cut a multi-byte character.

Return type:

tuple[str, int]

tail(job_id: str, lines: int = 50) → str

The last lines lines of the job log (empty when there is none yet).

Parameters:

job_id (str)
lines (int)

Return type:

str

plesty.server.model.tail_file(path: pathlib.Path, lines: int, block: int = 8192) → str

The last lines lines of path, read from the end in block steps.

Parameters:

path (pathlib.Path)
lines (int)
block (int)

Return type:

str

plesty.server.model.AGENT_PORT = 5550

plesty.server.model.DEFAULT_TCP_PORT = 5555

plesty.server.model.DEVICE_PORT_BASE = 5551

plesty.server.model.allocate_port(taken: collections.abc.Iterable[int], base: int = DEVICE_PORT_BASE, is_free: collections.abc.Callable[[int], bool] = port_is_free) → int

The lowest free port from base up, skipping taken and the agent’s.

Parameters:

taken (collections.abc.Iterable[int]) – Ports already spoken for on this bench.
base (int) – Where counting starts.
is_free (collections.abc.Callable[[int], bool]) – Asks the host whether a port can be bound; injected by tests.

Returns:

A port no other device on the bench holds and nothing on the host is listening on.

Raises:

FleetError – When counting runs past the end of the port range.

Return type:

int

plesty.server.model.declares_port(args: tuple[str, Ellipsis], env: collections.abc.Mapping[str, str]) → bool

Whether args or env name the port themselves.

A declaration that names its own port is left alone — the bench allocates only for devices that would otherwise take the default and collide.

Parameters:

args (tuple[str, Ellipsis])
env (collections.abc.Mapping[str, str])

Return type:

bool

plesty.server.model.port_from(args: tuple[str, Ellipsis], env: collections.abc.Mapping[str, str]) → int

Resolve the TCP port a server binds from its args and environment.

Parameters:

args (tuple[str, Ellipsis]) – Command-line arguments passed to python -m <module>.
env (collections.abc.Mapping[str, str]) – Environment values the fleet sets for the process.

Returns:

The port to probe, falling back to DEFAULT_TCP_PORT.

Return type:

int

plesty.server.model.port_is_free(port: int, host: str = '127.0.0.1') → bool

Whether port can actually be bound right now.

The fleet file and the process records say what this bench believes it is using. A server that outlived its record, or any other program on the machine, believes nothing — and the socket is the only thing that knows.

The socket option differs because the platforms disagree about what SO_REUSEADDR means. On POSIX it says “a port left in TIME_WAIT is still free”, which is the question. On Windows it says “bind even if another socket is listening here” — so the probe succeeded against a live server and every busy port reported free, which silently disabled everything built on this: the port a start moves away from, and the port allocate_port() hands out. SO_EXCLUSIVEADDRUSE is the Windows way to ask the question POSIX answers with SO_REUSEADDR.

Parameters:

port (int)
host (str)

Return type:

bool

plesty.server.model.port_listening(port: int, host: str = '127.0.0.1', timeout: float = 0.3) → bool

Whether something accepts TCP connections on port right now.

The complement of port_is_free() for a port a server holds: a device server binds tcp://*, and on some hosts a specific-address bind still succeeds beside a wildcard listener — connecting is the one test that answers the question everywhere.

Parameters:

port (int)
host (str)
timeout (float)

Return type:

bool

plesty.server.model.EXITED = 'exited'

plesty.server.model.RUNNING = 'running'

plesty.server.model.STARTING = 'starting'

plesty.server.model.STATES

plesty.server.model.STOPPED = 'stopped'

class plesty.server.model.ProcessRecord

What the supervisor persists about a started server.

Variables:

name – Device instance name.
pid – Process id of the launcher that owns the server.
port – TCP port it was told to bind.
started_utc – ISO timestamp of the launch.
command – The full command line.
cwd – Working directory the server runs in.
log – Path of the log file its output goes to.
python – The interpreter — what a relaunch has to repeat.
module – The python -m target — likewise.
args – The spec’s arguments — likewise.
env – The spec’s environment — likewise.
package – Distribution the server runs, for listings.
version – Its installed version, for listings.

name: str

pid: int

port: int

started_utc: str

command: list[str]

cwd: str

log: str

python: str = ''

module: str = ''

args: list[str] = []

env: dict[str, str]

package: str = ''

version: str = ''

to_dict() → dict[str, Any]

JSON form.

Return type:: dict[str, Any]

classmethod from_dict(data: collections.abc.Mapping[str, Any]) → ProcessRecord

Rebuild a record from to_dict() output (extra keys ignored).

Parameters:: data (collections.abc.Mapping[str, Any])
Return type:: ProcessRecord

classmethod load(path: pathlib.Path) → ProcessRecord | None

Read a record file; None when absent or unreadable.

Parameters:: path (pathlib.Path)
Return type:: ProcessRecord | None

save(path: pathlib.Path) → pathlib.Path

Write the record to path; returns it.

Parameters:: path (pathlib.Path)
Return type:: pathlib.Path

class plesty.server.model.ProcessStatus

The observed state of one device server.

Variables:

name – Device instance name.
state – One of STATES.
pid – Process id when a record exists.
port – Probed port when a record exists.
started_utc – Launch time when a record exists.
log – Log path when a record exists.
returncode – Exit code for an exited server, when the launcher recorded one (-1 when it vanished without).
detail – For an exited server, the tail of its log — why it died.

name: str

state: str

pid: int | None = None

port: int | None = None

started_utc: str | None = None

log: str | None = None

returncode: int | None = None

detail: str = ''

property alive: bool

Whether the process exists (running or still starting).

Return type:: bool

to_dict() → dict[str, Any]

JSON form (what status prints and the agent returns).

Return type:: dict[str, Any]

`DEFAULT_INDEX_URL`
`REQUIRED_STANDARD`
`SOURCE_GIT`
`SOURCE_PYPI`
`ENV_FILENAME`
`JOB_KIND`
`REPORT_BASENAME`
`STANDARD_GATES`
`TIER_ARTIFACTS`
`TIER_SCRIPTS`
`FLEET_FILENAME`
`FLEET_VERSION`
`HOME_ENV`
`CANCELLED_EXIT`
`LOG_CHUNK`
`LOST_EXIT`
`AGENT_PORT`
`DEFAULT_TCP_PORT`
`DEVICE_PORT_BASE`
`EXITED`
`RUNNING`
`STARTING`
`STATES`
`STOPPED`

plesty.server.model

Submodules

Attributes

Exceptions

Classes

Functions

Package Contents