plesty.server.cli.main

The plesty-server command group.

Every command opens the Bench for the --home/-f given (or found) and calls one presenter method. Long work (install, update, field test) is started as a job and the job log is streamed unless --no-wait is given; the hidden _job group is what those jobs execute in the foreground.

Attributes

`pass_context`
`EXPECTED`

Classes

Context

What every command receives: how to open the bench, and how loud to be.

Functions

`app`(→ None)	Install, launch and supervise PLESTY device servers on this bench.
`_run`(→ T)	Call action, turning every expected failure into a one-line CLI error.
`_follow_job`(→ int)	Stream a job's log until it ends; returns its exit code.
`_finish_job`(→ None)
`declare`(→ None)	Add device NAME running PACKAGE to the fleet file, or to the fleet NAME.
`serve`(→ None)	Have NAME running for an experiment's run; say the port it listens on.
`share_check`(→ None)	Read PROBE under ROOT, and write REPLY there: is ROOT the master's shared disk?
`_parse_env`(→ dict[str, str])
`install`(→ None)	Install the environment NAME declares (as a job).
`update`(→ None)	Pull and re-sync (git) or reinstall (PyPI) the environment of NAME.
`uninstall`(→ None)	Stop NAME and remove its environment (unless another device uses it).
`purge`(→ None)	Remove parts of this bench's home — or all of it.
`_write_purge_list`(→ None)	Write found for the uninstaller's list: one tab-separated line per row.
`_bytes`(→ str)	Render size in units a person reads at a glance.
`clear_log`(→ None)	Empty the server log of NAME (and with it the traffic history).
`module`(→ None)	The device modules on this bench — what the fleet declares instances of.
`module_install`(→ None)	Put PACKAGE on this bench (as a job); no device is declared.
`module_load`(→ None)	Load the module repository at FOLDER where it lives (as a job).
`module_list`(→ None)	List the modules installed on this bench.
`list_devices`(→ None)	Show every declared device: source, installed version, state.
`start`(→ None)	Start the device server NAME (installing first when needed).
`stop`(→ None)	Stop the device server NAME.
`restart`(→ None)	Stop (if running) and start the device server NAME.
`status`(→ None)	Show the state of NAMES (default: every declared device).
`log`(→ None)	Print the tail of the server log of NAME.
`field_test`(→ None)	Run the field test of NAME on this bench (as a job). EXTRA goes to the script.
`_field_test_run`(→ plesty.server.model.FieldTestRun)
`reports`(→ None)	List the published field-test reports of NAME.
`jobs`(→ None)	Long-running work on the bench: installs, updates, field tests.
`jobs_list`(→ None)	List jobs, newest first.
`jobs_log`(→ None)	Print the log of JOB_ID (from --offset; --json for a tailing client).
`jobs_cancel`(→ None)	Kill a running job.
`fleet`(→ None)	Operate every device declared in fleet.yaml at once.
`fleet_new`(→ None)	Create the empty fleet NAME on this bench (<home>/fleets/NAME.yaml).
`fleet_list`(→ None)	Every fleet of this bench: its name, file and device count; * marks the current.
`fleet_init`(→ None)	Write an annotated example fleet.yaml to PATH.
`fleet_up`(→ None)	Install and start the autostart devices (or NAMES).
`fleet_down`(→ None)	Stop the declared devices (or NAMES).
`fleet_run`(→ None)	Bring the fleet up and keep it up until interrupted (headless bench mode).
`fleet_status`(→ None)	Show the state of every declared device.
`agent`(→ None)	Serve plesty-bench clients: run plesty-server commands for them on this bench.
`data_path`(→ None)	Show this bench's Plesty share and folder, or set them.
`_echo_data_location`(→ None)	Print a bench's share, folder and the root they make, or that none is set.
`gui`(→ None)	Open the Qt console (needs the 'gui' extra).
`job_group`(→ None)	Foreground twins of the job commands (run by the launcher, not by people).
`job_install`(→ None)	Install NAME in the foreground.
`job_module_install`(→ None)	Install the module PACKAGE in the foreground.
`job_module_load`(→ None)	Load the module repository at FOLDER in the foreground.
`job_experiment_run`(→ None)	Run experiment NAME's module COMMAND, then stop the devices the run started.
`_keep_log_with_run`(→ None)	Leave a copy of this job's log in the run's directory; say why when it cannot.
`_report_on_run`(→ None)	Write the run's report into its own directory; say why when it cannot.
`job_update`(→ None)	Update NAME in the foreground.
`job_field_test`(→ None)	Run the field test of NAME in the foreground.
`launch_waiter`(→ None)	Run the detached launcher (spec as JSON on stdin); used by the PyInstaller bundle.
`_device_dict`(→ dict[str, object])
`_report_dict`(→ dict[str, object])
`_table`(→ list[str])
`_status_table`(→ list[str])
`_echo_status`(→ None)
`main`(→ None)	Console-script entry point.
`gui_main`(→ None)	GUI-script entry point (`plesty-server-gui`): the console window, no terminal.

Module Contents

class plesty.server.cli.main.Context(home: str | None, fleet: str | None, quiet: bool)

What every command receives: how to open the bench, and how loud to be.

Remember the global options.

Parameters:

home (str | None)
fleet (str | None)
quiet (bool)

home_arg

fleet_arg

quiet

_bench: plesty.server.presenter.Bench | None = None

bench() → plesty.server.presenter.Bench

The bench (opened once per invocation).

Return type:: plesty.server.presenter.Bench

say(line: str) → None

Progress to stderr unless --quiet.

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

plesty.server.cli.main.pass_context

plesty.server.cli.main.app(ctx: click.Context, home_arg: str | None, fleet_arg: str | None, quiet: bool) → None

Install, launch and supervise PLESTY device servers on this bench.

Parameters:

ctx (click.Context)
home_arg (str | None)
fleet_arg (str | None)
quiet (bool)

Return type:

None

plesty.server.cli.main.EXPECTED

plesty.server.cli.main._run[T](ctx: Context, action: collections.abc.Callable[[], T]) → T

Call action, turning every expected failure into a one-line CLI error.

Parameters:

ctx (Context)
action (collections.abc.Callable[[], T])

Return type:

T

plesty.server.cli.main._follow_job(ctx: Context, job: plesty.server.model.Job) → int

Stream a job’s log until it ends; returns its exit code.

Parameters:

ctx (Context)
job (plesty.server.model.Job)

Return type:

int

plesty.server.cli.main._finish_job(ctx: Context, job: plesty.server.model.Job, wait: bool) → None

Parameters:

ctx (Context)
job (plesty.server.model.Job)
wait (bool)

Return type:

None

plesty.server.cli.main.declare(ctx: Context, name: str, package: str, version: str | None, git_url: str | None, ref: str | None, args: tuple[str, Ellipsis], env: tuple[str, Ellipsis], no_autostart: bool, fleet_name: str | None) → None

Add device NAME running PACKAGE to the fleet file, or to the fleet NAME.

Parameters:

ctx (Context)
name (str)
package (str)
version (str | None)
git_url (str | None)
ref (str | None)
args (tuple[str, Ellipsis])
env (tuple[str, Ellipsis])
no_autostart (bool)
fleet_name (str | None)

Return type:

None

plesty.server.cli.main.serve(ctx: Context, name: str, entry_json: str, as_json: bool) → None

Have NAME running for an experiment’s run; say the port it listens on.

Declares it from –entry when the fleet lacks it (a free port when the entry names none), then starts it unless it is running already. What a master bench asks of this one when a run needs a device here.

Parameters:

ctx (Context)
name (str)
entry_json (str)
as_json (bool)

Return type:

None

plesty.server.cli.main.share_check(root: str, probe: str, reply: str, as_json: bool) → None

Read PROBE under ROOT, and write REPLY there: is ROOT the master’s shared disk?

A master bench writes PROBE under the disk as it sees it, and reads REPLY back: the same content at both ends shows both see one folder, and that this bench can write to it. Both paths stay under ROOT.

Parameters:

root (str)
probe (str)
reply (str)
as_json (bool)

Return type:

None

plesty.server.cli.main._parse_env(pairs: tuple[str, Ellipsis]) → dict[str, str]

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

plesty.server.cli.main.install(ctx: Context, name: str, force: bool, no_wait: bool) → None

Install the environment NAME declares (as a job).

Parameters:

ctx (Context)
name (str)
force (bool)
no_wait (bool)

Return type:

None

plesty.server.cli.main.update(ctx: Context, name: str, ref: str | None, no_wait: bool) → None

Pull and re-sync (git) or reinstall (PyPI) the environment of NAME.

Parameters:

ctx (Context)
name (str)
ref (str | None)
no_wait (bool)

Return type:

None

plesty.server.cli.main.uninstall(ctx: Context, name: str) → None

Stop NAME and remove its environment (unless another device uses it).

Parameters:

ctx (Context)
name (str)

Return type:

None

plesty.server.cli.main.purge(ctx: Context, parts: tuple[str, Ellipsis], listing: bool, everything: bool, as_json: bool, tsv: str | None, yes: bool) → None

Remove parts of this bench’s home — or all of it.

Uninstalling the application and wiping the bench are different acts, and only the first had a command. PARTS names what to remove (purge –list prints the names): a whole part, such as packages, or one item of it, such as module:plesty-pm100d. –all removes the home directory itself.

Stop the fleet first: a running server holds its environment open.

Parameters:

ctx (Context)
parts (tuple[str, Ellipsis])
listing (bool)
everything (bool)
as_json (bool)
tsv (str | None)
yes (bool)

Return type:

None

plesty.server.cli.main._write_purge_list(path: pathlib.Path, found: collections.abc.Sequence[plesty.server.model.purge.Removable]) → None

Write found for the uninstaller’s list: one tab-separated line per row.

A row is level, key, label, size: a group’s heading at level 0 and its items at level 1 under it, or a single level-0 row for a part that is one item (process state, tools). The uninstaller ticks a heading’s items with it and passes back only item keys. UTF-8 with a byte-order mark, which is how Inno Setup’s LoadStringsFromFile tells it from the ANSI codepage.

Parameters:

path (pathlib.Path)
found (collections.abc.Sequence[plesty.server.model.purge.Removable])

Return type:

None

plesty.server.cli.main._bytes(size: int) → str

Render size in units a person reads at a glance.

Binary divisor, binary names. It divided by 1024 and wrote MB, so the printed column and the --json byte count disagreed by 5% — on the one screen whose whole purpose is telling somebody what removing a part costs them, which is exactly where a reader stops and checks twice. MiB is ugly and unambiguous; the alternative, dividing by 1000 to match the name, would have disagreed with what Windows shows for the same directory instead.

Parameters:: size (int) – A number of bytes.
Returns:: The size with a unit that means what it says.
Return type:: str

plesty.server.cli.main.clear_log(ctx: Context, name: str) → None

Empty the server log of NAME (and with it the traffic history).

Parameters:

ctx (Context)
name (str)

Return type:

None

plesty.server.cli.main.module() → None

The device modules on this bench — what the fleet declares instances of.

Return type:: None

plesty.server.cli.main.module_install(ctx: Context, package: str, version: str | None, git_url: str | None, ref: str | None, force: bool, no_wait: bool) → None

Put PACKAGE on this bench (as a job); no device is declared.

Parameters:

ctx (Context)
package (str)
version (str | None)
git_url (str | None)
ref (str | None)
force (bool)
no_wait (bool)

Return type:

None

plesty.server.cli.main.module_load(ctx: Context, folder: str, ref: str | None, no_wait: bool) → None

Load the module repository at FOLDER where it lives (as a job).

Nothing is copied: the bench records the folder, what the module is, its branch, version and commit, and runs it with the folder’s own .venv.

Parameters:

ctx (Context)
folder (str)
ref (str | None)
no_wait (bool)

Return type:

None

plesty.server.cli.main.module_list(ctx: Context, as_json: bool) → None

List the modules installed on this bench.

Parameters:

ctx (Context)
as_json (bool)

Return type:

None

plesty.server.cli.main.list_devices(ctx: Context, no_probe: bool, as_json: bool) → None

Show every declared device: source, installed version, state.

Parameters:

ctx (Context)
no_probe (bool)
as_json (bool)

Return type:

None

plesty.server.cli.main.start(ctx: Context, name: str, no_wait: bool, timeout: float | None) → None

Start the device server NAME (installing first when needed).

Parameters:

ctx (Context)
name (str)
no_wait (bool)
timeout (float | None)

Return type:

None

plesty.server.cli.main.stop(ctx: Context, name: str, timeout: float | None) → None

Stop the device server NAME.

Parameters:

ctx (Context)
name (str)
timeout (float | None)

Return type:

None

plesty.server.cli.main.restart(ctx: Context, name: str, no_wait: bool) → None

Stop (if running) and start the device server NAME.

Parameters:

ctx (Context)
name (str)
no_wait (bool)

Return type:

None

plesty.server.cli.main.status(ctx: Context, names: tuple[str, Ellipsis], no_probe: bool, as_json: bool) → None

Show the state of NAMES (default: every declared device).

Parameters:

ctx (Context)
names (tuple[str, Ellipsis])
no_probe (bool)
as_json (bool)

Return type:

None

plesty.server.cli.main.log(ctx: Context, name: str, lines: int) → None

Print the tail of the server log of NAME.

Parameters:

ctx (Context)
name (str)
lines (int)

Return type:

None

plesty.server.cli.main.field_test(ctx: Context, name: str, extra: tuple[str, Ellipsis], tier: str, address: str | None, gates: str | None, repetitions: int | None, keep_going: bool, keep_server: bool, no_wait: bool) → None

Run the field test of NAME on this bench (as a job). EXTRA goes to the script.

Parameters:

ctx (Context)
name (str)
extra (tuple[str, Ellipsis])
tier (str)
address (str | None)
gates (str | None)
repetitions (int | None)
keep_going (bool)
keep_server (bool)
no_wait (bool)

Return type:

None

plesty.server.cli.main._field_test_run(tier: str, address: str | None, gates: str | None, repetitions: int | None, keep_going: bool, keep_server: bool, extra: tuple[str, Ellipsis]) → plesty.server.model.FieldTestRun

Parameters:

tier (str)
address (str | None)
gates (str | None)
repetitions (int | None)
keep_going (bool)
keep_server (bool)
extra (tuple[str, Ellipsis])

Return type:

plesty.server.model.FieldTestRun

plesty.server.cli.main.reports(ctx: Context, name: str, as_json: bool) → None

List the published field-test reports of NAME.

Parameters:

ctx (Context)
name (str)
as_json (bool)

Return type:

None

plesty.server.cli.main.jobs() → None

Long-running work on the bench: installs, updates, field tests.

Return type:: None

plesty.server.cli.main.jobs_list(ctx: Context, name: str | None, kind: str | None, as_json: bool) → None

List jobs, newest first.

Parameters:

ctx (Context)
name (str | None)
kind (str | None)
as_json (bool)

Return type:

None

plesty.server.cli.main.jobs_log(ctx: Context, job_id: str, follow: bool, offset: int, as_json: bool) → None

Print the log of JOB_ID (from –offset; –json for a tailing client).

Parameters:

ctx (Context)
job_id (str)
follow (bool)
offset (int)
as_json (bool)

Return type:

None

plesty.server.cli.main.jobs_cancel(ctx: Context, job_id: str) → None

Kill a running job.

Parameters:

ctx (Context)
job_id (str)

Return type:

None

plesty.server.cli.main.fleet() → None

Operate every device declared in fleet.yaml at once.

Return type:: None

plesty.server.cli.main.fleet_new(ctx: Context, name: str) → None

Create the empty fleet NAME on this bench (<home>/fleets/NAME.yaml).

The current fleet stays current; declare into the new one with declare … --fleet NAME.

Parameters:

ctx (Context)
name (str)

Return type:

None

plesty.server.cli.main.fleet_list(ctx: Context, as_json: bool) → None

Every fleet of this bench: its name, file and device count; * marks the current.

Parameters:

ctx (Context)
as_json (bool)

Return type:

None

plesty.server.cli.main.fleet_init(path: str, force: bool) → None

Write an annotated example fleet.yaml to PATH.

Parameters:

path (str)
force (bool)

Return type:

None

plesty.server.cli.main.fleet_up(ctx: Context, names: tuple[str, Ellipsis], no_install: bool, no_wait: bool) → None

Install and start the autostart devices (or NAMES).

Parameters:

ctx (Context)
names (tuple[str, Ellipsis])
no_install (bool)
no_wait (bool)

Return type:

None

plesty.server.cli.main.fleet_down(ctx: Context, names: tuple[str, Ellipsis]) → None

Stop the declared devices (or NAMES).

Parameters:

ctx (Context)
names (tuple[str, Ellipsis])

Return type:

None

plesty.server.cli.main.fleet_run(ctx: Context, names: tuple[str, Ellipsis], interval: float, with_agent: bool, bind: str | None, token: str | None, leave_running: bool, sweeps: int | None) → None

Bring the fleet up and keep it up until interrupted (headless bench mode).

Parameters:

ctx (Context)
names (tuple[str, Ellipsis])
interval (float)
with_agent (bool)
bind (str | None)
token (str | None)
leave_running (bool)
sweeps (int | None)

Return type:

None

plesty.server.cli.main.fleet_status(ctx: Context, no_probe: bool) → None

Show the state of every declared device.

Parameters:

ctx (Context)
no_probe (bool)

Return type:

None

plesty.server.cli.main.agent(ctx: Context, bind: str | None, token: str | None, workers: int, max_replies: int | None) → None

Serve plesty-bench clients: run plesty-server commands for them on this bench.

Defaults come from the bench’s own agent settings (<home>/agent.yaml, written by the console’s Server ▸ Configure…), so a configured bench is served the same way whichever surface starts it. Every option overrides.

Parameters:

ctx (Context)
bind (str | None)
token (str | None)
workers (int)
max_replies (int | None)

Return type:

None

plesty.server.cli.main.data_path(ctx: Context, share: str | None, folder: str, clear: bool) → None

Show this bench’s Plesty share and folder, or set them.

SHARE is the lab’s shared disk as this machine mounts it, and FOLDER this bench’s folder in it, as Server ▸ Configure… sets them. Every device the bench starts writes under SHARE/FOLDER, to its data_dir or tmp/<device>, and is given SHARE as PLESTY_DATA. SHARE must be an existing folder; FOLDER is created when missing. Through plesty-bench -b BENCH data-path SHARE --folder FOLDER it sets another bench’s.

Parameters:

ctx (Context)
share (str | None)
folder (str)
clear (bool)

Return type:

None

plesty.server.cli.main._echo_data_location(settings: plesty.server.model.AgentSettings) → None

Print a bench’s share, folder and the root they make, or that none is set.

Parameters:: settings (plesty.server.model.AgentSettings)
Return type:: None

plesty.server.cli.main.gui(ctx: Context, title: str | None) → None

Open the Qt console (needs the ‘gui’ extra).

Parameters:

ctx (Context)
title (str | None)

Return type:

None

plesty.server.cli.main.job_group() → None

Foreground twins of the job commands (run by the launcher, not by people).

Return type:: None

plesty.server.cli.main.job_install(ctx: Context, name: str, force: bool) → None

Install NAME in the foreground.

Parameters:

ctx (Context)
name (str)
force (bool)

Return type:

None

plesty.server.cli.main.job_module_install(ctx: Context, package: str, version: str | None, git_url: str | None, ref: str | None, force: bool) → None

Install the module PACKAGE in the foreground.

Parameters:

ctx (Context)
package (str)
version (str | None)
git_url (str | None)
ref (str | None)
force (bool)

Return type:

None

plesty.server.cli.main.job_module_load(ctx: Context, folder: str, ref: str | None) → None

Load the module repository at FOLDER in the foreground.

Parameters:

ctx (Context)
folder (str)
ref (str | None)

Return type:

None

plesty.server.cli.main.job_experiment_run(ctx: Context, name: str, release: tuple[str, Ellipsis], run_root: str | None, run_id: str | None, command: tuple[str, Ellipsis]) → None

Run experiment NAME’s module COMMAND, then stop the devices the run started.

The module ends however it ends — completed, stopped, aborted, crashed — and its devices are stopped every time; the job exits with its code.

The run’s report is written into its directory afterwards, so a finished measurement carries its own reading of itself without anybody pressing anything. Devices are released first: a report is worth having and a device left holding the hardware is not.

Parameters:

ctx (Context)
name (str)
release (tuple[str, Ellipsis])
run_root (str | None)
run_id (str | None)
command (tuple[str, Ellipsis])

Return type:

None

plesty.server.cli.main._keep_log_with_run(name: str, run_root: str, run_id: str | None) → None

Leave a copy of this job’s log in the run’s directory; say why when it cannot.

The job’s log is in the bench home, which an uninstall can empty and another bench never had; the run is on the shared disk. Last, so the copy holds the report’s line too. Never raises, for the reason the report does not (#33).

Parameters:

name (str)
run_root (str)
run_id (str | None)

Return type:

None

plesty.server.cli.main._report_on_run(name: str, run_root: str, run_id: str | None) → None

Write the run’s report into its own directory; say why when it cannot.

Never raises. The measurement is already over and its data is on disk, so a report that cannot be compiled is worth a line in the job log and nothing more — failing the job here would report a successful run as a failed one.

Parameters:

name (str)
run_root (str)
run_id (str | None)

Return type:

None

plesty.server.cli.main.job_update(ctx: Context, name: str, ref: str | None) → None

Update NAME in the foreground.

Parameters:

ctx (Context)
name (str)
ref (str | None)

Return type:

None

plesty.server.cli.main.job_field_test(ctx: Context, name: str, extra: tuple[str, Ellipsis], tier: str, address: str | None, gates: str | None, repetitions: int | None, keep_going: bool, keep_server: bool) → None

Run the field test of NAME in the foreground.

Parameters:

ctx (Context)
name (str)
extra (tuple[str, Ellipsis])
tier (str)
address (str | None)
gates (str | None)
repetitions (int | None)
keep_going (bool)
keep_server (bool)

Return type:

None

plesty.server.cli.main.launch_waiter() → None

Run the detached launcher (spec as JSON on stdin); used by the PyInstaller bundle.

Return type:: None

plesty.server.cli.main._device_dict(view: plesty.server.presenter.DeviceView) → dict[str, object]

Parameters:: view (plesty.server.presenter.DeviceView)
Return type:: dict[str, object]

plesty.server.cli.main._report_dict(ctx: Context, report: plesty.server.model.FieldTestReport) → dict[str, object]

Parameters:

ctx (Context)
report (plesty.server.model.FieldTestReport)

Return type:

dict[str, object]

plesty.server.cli.main._table(header: collections.abc.Sequence[str], rows: collections.abc.Sequence[collections.abc.Sequence[str]]) → list[str]

Parameters:

header (collections.abc.Sequence[str])
rows (collections.abc.Sequence[collections.abc.Sequence[str]])

Return type:

list[str]

plesty.server.cli.main._status_table(statuses: collections.abc.Sequence[plesty.server.model.ProcessStatus]) → list[str]

Parameters:: statuses (collections.abc.Sequence[plesty.server.model.ProcessStatus])
Return type:: list[str]

plesty.server.cli.main._echo_status(status: plesty.server.model.ProcessStatus) → None

Parameters:: status (plesty.server.model.ProcessStatus)
Return type:: None

plesty.server.cli.main.main() → None

Console-script entry point.

Return type:: None

plesty.server.cli.main.gui_main() → None

GUI-script entry point (plesty-server-gui): the console window, no terminal.

Installed as a gui-scripts entry so Windows starts it with pythonw — no black console box behind the window. Arguments are the group’s (--home, -f); _launch/_job pass through for re-execution.

Return type:: None