plesty.lib.experiment

Experiment orchestration: plans, journals, and the async Experiment ABC.

Submodules

• plesty.lib.experiment.base_experiment
• plesty.lib.experiment.cli
• plesty.lib.experiment.control
• plesty.lib.experiment.journal
• plesty.lib.experiment.manifest
• plesty.lib.experiment.preflight
• plesty.lib.experiment.runs
• plesty.lib.experiment.schedule
• plesty.lib.experiment.setup

Attributes

DEFAULT_GROUP
YAML_DTYPE
SETUP_FILE

Exceptions

InvalidOperationError	Raised when a plan step references an invalid experiment operation.
PlanMismatchError	Raised when resuming a run whose stored plan differs from the current one.
ManifestError	Raised when a manifest document is malformed or incomplete.
RigNotReadyError	The rig failed its preflight and the operator did not (or could not) recover it.
SetupError	Raised when a setup cannot be read, or writes a value where a variable belongs.
UnsetVariables	Raised when the setup names variables nothing gives a value.

Classes

Experiment	Base class for PLESTY experiments with atomic scheduling and resume.
ExperimentCLI	The standard python -m <module> command line over one experiment.
Journal	Crash-safe append-only event log for one experiment run.
EnvVar	One environment variable the run reads.
Manifest	What one experiment module declares about running it.
Option	One extra command-line option the module adds to the standard set.
Run	Read access to one run directory: plan, journal, records, data paths.
Plan	A frozen, hashable schedule of atomic steps plus the run configuration.
Step	One atomic measurement step in an experiment plan.
Bench	One bench of a setup.
Setup	The benches and devices of a module's setup.yaml.
SetupDevice	What a setup says about one device, as far as a run is concerned.

Functions

env_vars(→ tuple[EnvVar, Ellipsis])	Build required environment declarations.
group_of(→ tuple[str, str])	Split a configuration key into its group and parameter name.
iter_parameters(→ Iterable[tuple[str, ...)	Yield every (group, parameter) pair of a schema, groups in order.
yaml_parameter(→ plesty.lib.device.params.ConfigParameter)	Declare a configuration key whose value is a document, not a scalar.
connect_rig(, ask, str]] = None, env_prefix, out, ...)	Build the rig, run its preflight, and ask the operator on a failure.
check_run_id(→ str)	Return text if it is usable as a run directory name, else raise.
latest_run(→ pathlib.Path)	Return the most recently started run under run_root.
list_runs(→ list[pathlib.Path])	Return the run directories under run_root, oldest first.
parse_run_id(→ Optional[tuple[str, datetime.datetime]])	Split a run id into its experiment name and start time.
run_id(→ str)	Return the run id for experiment name started at when (now by default).
apply_setup(→ dict[str, str])	Put what the setup at root builds into os.environ, under what is set already.
seed_env(→ list[str])	Give the checkout at root a .env line for every variable its setup names.
split_benches(→ tuple[dict[str, dict[str, str]], ...)	Split benches given by value into what setup.yaml says and what .env holds.
variable_of(→ str)	The variable a setup's written field names — SPEC_PC_TOKEN for ${SPEC_PC_TOKEN}.

Package Contents

class plesty.lib.experiment.Experiment(devices: plesty.lib.device.composite_device.CompositeDevice | None = None, run_root: str | pathlib.Path | None = None, name: str | None = None, max_retries: int = 3, retry_sleep: float = 3.0, data_root: str | None = None)

Bases: abc.ABC

Base class for PLESTY experiments with atomic scheduling and resume.

Subclasses override exactly three lifecycle hooks:

• build_plan() — required: the atomic measurement schedule.

• setup() / teardown() — optional: extend device connection/release (call await super().setup() / ...teardown()).

plus one public method per step operation. Step operations are validated before a run starts — every Step.op must name an existing public callable on the experiment and must not be one of the reserved lifecycle methods above. Device access goes through the CompositeDevice passed at construction — an Experiment receives Device instances, never raw hardware.

Initialize the experiment.

Parameters:

• devices (Optional[plesty.lib.device.composite_device.CompositeDevice]) – Composite device holding all instruments this experiment orchestrates; connected in setup() and disconnected in teardown().

• run_root (str | pathlib.Path | None) – Directory under which each run creates its own checkpoint directory — the experiment’s root. A relative one is under the shared point (PLESTY_DATA, pol-pl → <PLESTY_DATA>/pol-pl); None is the shared point itself, and runs without one.

• name (Optional[str]) – Experiment name used in run ids; defaults to the subclass name in lowercase.

• max_retries (int) – Attempts per step before the run aborts.

• retry_sleep (float) – Seconds to wait between step retries.

• data_root (Optional[str]) – The shared data disk as this machine sees it; None takes PLESTY_DATA. Journaled for readers.

devices = None

data_root

run_root

name = ''

max_retries = 3

retry_sleep = 3.0

run_id: str | None = None

run_dir: pathlib.Path | None = None

control: plesty.lib.experiment.control.Control | None = None

control_poll = 0.5

_reported = 0

_step_id: str | None = None

_routes: dict[str, str]

logger

_plan_length = 0

property raw_dir: str | None

Where this run routes its devices’ blobs, relative to each one’s data root.

The run’s own path from the share, then raw — pol-pl/<run_id>/raw for a run under <PLESTY_DATA>/pol-pl, <run_id>/raw for one directly in it; <folder>/<run_id>/raw for a run kept off the share. A device may write only under its bench’s folder on the Plesty share, so this one relative path lands in that folder on every bench: route() sends each device there, into raw/<device>. One folder per run, so a directory the device lists on every acquisition never grows across runs, and a resumed run keeps writing where it started.

None before a run has started, or without a shared point — a run whose machine does not see the share has nowhere to read the files from.

Return type:

Optional[str]

route(devices: Any, names: collections.abc.Iterable[str] | None = None) → dict[str, str]

Send names of devices to write into raw_dir, each into raw/<name>.

What each device answers — its folder relative to the Plesty share, under its own bench’s folder — is kept, and raw_file() records its files there. Call it in setup(), which a resumed run repeats.

Parameters:

• devices (Any) – The composite holding them (rig, scanner), anything with set_data_path(path, subdirs=True, devices=...).

• names (Optional[collections.abc.Iterable[str]]) – The sub-devices that write files; None routes all.

Returns:

Sub-device name → its folder, relative to the Plesty share.

Raises:

RuntimeError – When the run has no raw folder (see raw_dir).

Return type:

dict[str, str]

raw_file(device: str, reported: str) → str

A file device wrote into this run’s raw folder, relative to the Plesty share.

A device reports a file as its own machine spells it; only its name is taken, since the run chose the folder — the one device answered when route() sent it there. The result is what a record keeps, and what plesty.lib.experiment.Run.local() joins to any machine’s view of the share.

Parameters:

• device (str) – The sub-device, as route() routed it.

• reported (str) – The file as the device reported it, in any spelling.

Raises:

• RuntimeError – When device was not routed this run.

• ValueError – When reported names no file.

Return type:

str

abstractmethod build_plan() → plesty.lib.experiment.schedule.Plan

Build the atomic measurement schedule for this experiment.

Must be deterministic for a given configuration: resume relies on the regenerated plan matching the persisted one by content hash.

Return type:

plesty.lib.experiment.schedule.Plan

async setup() → None

Prepare the experiment; the default connects and preflights all devices.

Raises:

RigNotReadyError – If a sub-device does not answer or is not the device its configuration requires (see CompositeDevice.preflight()). Nothing is measured on a rig that is not what the plan assumes.

Return type:

None

async teardown() → None

Release resources; the default disconnects all devices.

Always awaited by run(), whether the run completes, fails, or is canceled.

Return type:

None

async run(resume: str | None = None, run_id: str | None = None) → str

Execute the plan, journaling every step; optionally resume a run.

Parameters:

• resume (Optional[str]) – Run id of an earlier stopped/canceled/aborted run to continue. Steps journaled as completed are skipped; the stored plan must match the current one by content hash.

• run_id (Optional[str]) – What to call this run, instead of the generated <name>_<timestamp>. For a caller that names its own runs — a console where an operator types the name, or a schedule that numbers them — so the name a person chose is the directory the data lands in. Ignored when resume is given, which already names a run.

Returns:

The run id (directory name under run_root) — pass it back as resume to continue after an interruption.

Raises:

• InvalidOperationError – If a plan step references a missing, private, or reserved operation (checked before anything runs).

• PlanMismatchError – If resuming and the stored plan differs from the plan the experiment generates now.

• ValueError – If run_id is not usable as a directory name, or names a run that has already started.

• Exception – The final step error is re-raised after run_aborted is journaled and teardown() has completed.

Return type:

str

_validate_plan(plan: plesty.lib.experiment.schedule.Plan) → None

Check that every step operation is a valid public method.

Runs before any device is touched or any journal written, so a misconfigured plan fails fast instead of aborting mid-measurement.

Raises:

InvalidOperationError – Listing every offending step.

Parameters:

plan (plesty.lib.experiment.schedule.Plan)

Return type:

None

_prepare_run(plan: plesty.lib.experiment.schedule.Plan, resume: str | None, run_id: str | None = None) → tuple[str, set[str]]

Create a new run directory or validate and load a resumed one.

Returns:

The run id and the set of step ids already completed — those the journal marks completed plus those with a committed record.

Raises:

ValueError – If run_id is unusable as a directory name, or names a run that has already started — starting a second run into one directory would interleave two journals and two sets of records, and the pair could not be told apart afterwards.

Parameters:

• plan (plesty.lib.experiment.schedule.Plan)

• resume (Optional[str])

• run_id (Optional[str])

Return type:

tuple[str, set[str]]

provenance() → dict[str, Any]

What was running when this run started, journaled with it.

A run outlives the bench it was measured on and is read months later, often somewhere else. What it measured is in the records and what it was asked to measure is in the plan; this is the third thing a reader needs and the only one that cannot be reconstructed afterwards — which versions of what produced it. An analysis that changed between two releases is otherwise invisible in the data it produced.

Every distribution in the environment whose name begins plesty is recorded, not a chosen few: which of them mattered is a question for whoever reads the run, and a list curated now would be curated wrong.

Returns:

The experiment’s class and module, the host and the interpreter, the installed plesty distributions, and the UTC time it was read.

Return type:

dict[str, Any]

device_setup(root: str | os.PathLike[str] = '.') → dict[str, dict[str, str]]

Where each device was for this start of the run, journaled with it.

The addresses a run connects to are handed to it by the benches that served its devices, on a port each picked for this start; once the run has ended they are gone from everywhere but here. Read from the module’s setup.yaml and the run’s environment (record()).

Parameters:

root (str | os.PathLike[str]) – The module checkout the run was started in, where its setup.yaml and .env are.

Returns:

its bench, the address the run was given, and that bench’s data; empty for a module without a setup.

Return type:

By device

progress(done: int, total: int, detail: str = '') → None

Say how far the step in flight has got.

Call this from a step operation that takes long enough for silence to be indistinguishable from a hang — a stack of exposures, a magnet ramp, a wait for a file to appear on the share. The journal speaks only at step boundaries, so without this a watcher has nothing to show for minutes at a time.

The file is replaced on every call and holds only the present; it is not part of the run’s record and nothing resumes from it. Writing it must never cost the measurement, so a failure to write is ignored.

Parameters:

• done (int) – Units finished within this step.

• total (int) – Units the step will do in all.

• detail (str) – What it is doing, shown beside the count.

Return type:

None

_control_state() → plesty.lib.experiment.control.ControlState

Return what the driver is asking for, or nothing when none is attached.

Return type:

plesty.lib.experiment.control.ControlState

_journal_reports(journal: plesty.lib.experiment.journal.Journal, state: plesty.lib.experiment.control.ControlState) → None

Journal the operator’s reports that have not been recorded yet.

They are consumed by count, so re-reading the control file does not record the same line twice.

Parameters:

• journal (plesty.lib.experiment.journal.Journal) – The run’s journal.

• state (plesty.lib.experiment.control.ControlState) – The control state just read.

Return type:

None

async _await_control(journal: plesty.lib.experiment.journal.Journal) → None

Hold at a step boundary for as long as the driver asks, then go on.

This is where a pause takes effect: between two steps, where the journal has just committed and no instrument is mid-operation. The rig stays connected throughout — that is what a pause buys over stopping and resuming.

Parameters:

journal (plesty.lib.experiment.journal.Journal) – The run’s journal.

Raises:

RunStopped – If a cancel or an abort was asked for. At a boundary the two mean the same thing — there is no step in flight to interrupt — but each keeps its own name in the journal.

Return type:

None

async _guarded_step(step: plesty.lib.experiment.schedule.Step, index: int, run_dir: pathlib.Path, journal: plesty.lib.experiment.journal.Journal) → None

Run one step, watching for an abort while it is in flight.

A cancel is not watched for here: it means finish this step first, and so is read at the next boundary. An abort is the one request that interrupts the work, which is the reason it exists — an instrument that must be released now is not worth a spectrum.

Parameters:

• step (plesty.lib.experiment.schedule.Step) – The step to run.

• index (int) – Its position in the plan.

• run_dir (pathlib.Path) – The run’s checkpoint directory.

• journal (plesty.lib.experiment.journal.Journal) – The run’s journal.

Raises:

RunStopped – If an abort was asked for while the step ran.

Return type:

None

async _run_step(step: plesty.lib.experiment.schedule.Step, index: int, run_dir: pathlib.Path, journal: plesty.lib.experiment.journal.Journal) → None

Execute one atomic step with retries and persist its result.

Parameters:

• step (plesty.lib.experiment.schedule.Step)

• index (int)

• run_dir (pathlib.Path)

• journal (plesty.lib.experiment.journal.Journal)

Return type:

None

exception plesty.lib.experiment.InvalidOperationError

Bases: RuntimeError

Raised when a plan step references an invalid experiment operation.

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

exception plesty.lib.experiment.PlanMismatchError

Bases: RuntimeError

Raised when resuming a run whose stored plan differs from the current one.

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

class plesty.lib.experiment.ExperimentCLI(manifest: plesty.lib.experiment.manifest.Manifest)

The standard python -m <module> command line over one experiment.

Describe the experiment this command line runs.

Parameters:

manifest (plesty.lib.experiment.manifest.Manifest) – What the module declares — the configuration schema, the environment it reads, its default configuration file and run root. Options added with add_argument() are folded in, so the printed manifest and the parser cannot drift apart.

manifest

_rig: RigFunc | None = None

_build: BuildFunc | None = None

_configure: ConfigureFunc | None = None

_default_config: DefaultConfigFunc | None = None

_check: CheckFunc | None = None

_finished: FinishedFunc | None = None

_needs: NeedsFunc | None = None

_optional: frozenset[str]

_extra: list[tuple[tuple[Any, Ellipsis], dict[str, Any]]] = []

_options: list[plesty.lib.experiment.manifest.Option]

rig(func: RigFunc) → RigFunc

Register the function that builds the rig. Use as a decorator.

It is called again on every preflight retry, so an edited .env or a server started in the meantime is picked up. Returning None declares an experiment that drives no devices.

Parameters:

func (RigFunc)

Return type:

RigFunc

build(func: BuildFunc) → BuildFunc

Register the function that builds the experiment. Use as a decorator.

It receives the rig — None for --dry-run, where nothing is connected but the plan must still be built.

Parameters:

func (BuildFunc)

Return type:

BuildFunc

default_config(func: DefaultConfigFunc) → DefaultConfigFunc

Register which configuration file to read when none was named.

Use as a decorator, for a module whose own options imply a different configuration — a --demo that replays a recording is configured differently from a run against instruments. Returning None takes the declared file. --config still wins over both: whoever passed it said which file.

Parameters:

func (DefaultConfigFunc)

Return type:

DefaultConfigFunc

configure(func: ConfigureFunc) → ConfigureFunc

Register the configuration override hook. Use as a decorator.

Called once with the loaded configuration and the parsed arguments, before the plan is built; what it returns is the configuration the run freezes into its plan. Optional — without it the file is used as read.

Parameters:

func (ConfigureFunc)

Return type:

ConfigureFunc

finished(func: FinishedFunc) → FinishedFunc

Register what to say when a run completes. Use as a decorator.

Called with the run directory once the plan is through, for a module that can summarise what it measured — the fitted results, a count, a path worth naming. It runs only on a completed run: a canceled or aborted one has a summary that would be a half-truth.

Parameters:

func (FinishedFunc)

Return type:

FinishedFunc

check(func: CheckFunc) → CheckFunc

Register the pre-connection check. Use as a decorator.

Called before the rig is built, to refuse an incomplete configuration while nothing has been connected and nothing measured. Optional.

Parameters:

func (CheckFunc)

Return type:

CheckFunc

needs(func: NeedsFunc) → NeedsFunc

Register which devices of setup.yaml a configuration uses. Use as a decorator.

One module can measure in configurations that use different devices — in place with four, or navigating with four more on another bench. The bench serves only what this returns, and the run needs only those devices’ benches set up (#62). None means all of them, as without this function. Optional.

@cli.needs
def needs(config):
    return ["hwp", "spec"] if not config.get("dots") else None

Parameters:

func (NeedsFunc)

Return type:

NeedsFunc

devices(args: argparse.Namespace) → list[str] | None

The devices of setup.yaml this run uses; None for all of them.

Parameters:

args (argparse.Namespace) – The parsed arguments; their configuration is what decides.

Return type:

Optional[list[str]]

optional(*names: str) → None

Declare sub-devices the run can do without.

When only these fail the preflight, the operator may continue without them; anything else is a rig that is not what the plan assumes.

Parameters:

*names (str) – Sub-device names.

Return type:

None

add_argument(*args: Any, sets: str | None = None, **kwargs: Any) → None

Add one option beyond the standard set, and declare it.

The option reaches both the parser and the printed manifest, so a caller offering the experiment’s settings sees exactly what the command line accepts.

Parameters:

• *args (Any) – As argparse.ArgumentParser.add_argument().

• sets (Optional[str]) – The configuration key this option overrides, written as a renderer addresses it (analysis.fit_options.prominence); None when it changes something else.

• **kwargs (Any) – As argparse.ArgumentParser.add_argument().

Return type:

None

declared() → plesty.lib.experiment.manifest.Manifest

Return the manifest including the options added to the parser.

Return type:

plesty.lib.experiment.manifest.Manifest

build_parser() → argparse.ArgumentParser

Return the argument parser: the standard options plus the module’s.

Return type:

argparse.ArgumentParser

run_root(args: argparse.Namespace) → str | None

Return where this run’s checkpoint directory goes.

--run-root wins, as given; otherwise the module’s declared root — under the shared point when one is configured (pol-pl → <PLESTY_DATA>/pol-pl), as it is when not.

Parameters:

args (argparse.Namespace) – The parsed arguments.

Returns:

The run root, as a path string.

Return type:

Optional[str]

config_path(args: argparse.Namespace) → pathlib.Path

Return the configuration file this run reads.

--config always wins: whoever passed it said which file, and a module second-guessing that would make the option a suggestion. With nothing passed, a module that chooses its own — one whose --demo implies a different configuration — decides through default_config(), and otherwise it is the declared one.

Parameters:

args (argparse.Namespace) – The parsed arguments.

Returns:

The file to read.

Return type:

pathlib.Path

load(args: argparse.Namespace) → dict[str, Any]

Return the configuration this run uses, overrides applied.

A missing file is an empty configuration, so an experiment whose defaults are all internal runs without one.

Parameters:

args (argparse.Namespace) – The parsed arguments.

Returns:

The configuration to freeze into the plan.

Return type:

dict[str, Any]

main(argv: Sequence[str] | None = None) → int

Run the command line and return the exit code.

Parameters:

argv (Optional[Sequence[str]]) – Argument list; None uses sys.argv.

Returns:

0 when the run completed, the plan printed or the manifest was written; non-zero when the rig was not ready.

Raises:

RuntimeError – If no rig or build function was registered.

Return type:

int

async _execute(args: argparse.Namespace) → int

Load, plan, connect and run, printing what each outcome means.

Parameters:

args (argparse.Namespace) – The parsed arguments.

Returns:

The exit code.

Return type:

int

class plesty.lib.experiment.Journal(path: str | pathlib.Path)

Crash-safe append-only event log for one experiment run.

Attach the journal to path, creating parent directories.

Parameters:

path (str | pathlib.Path) – Location of the journal.jsonl file; appended to if it already exists (the resume case).

TERMINAL_EVENTS = ('run_completed', 'run_canceled', 'run_aborted')

PAUSE_EVENT = 'run_paused'

RESUME_EVENT = 'run_resumed'

path

append(event: str, **fields: Any) → dict[str, Any]

Append one event line and fsync it to disk.

Parameters:

• event (str) – Event name, e.g. "step_completed".

• fields (Any) – Additional payload stored on the event (step id, result path, error text, …).

Returns:

The full event record that was written.

Return type:

dict[str, Any]

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

Read all events back, tolerating a torn final line after a crash.

Returns:

The recorded events in write order; an empty list if the journal file does not exist yet.

Return type:

list[dict[str, Any]]

completed_steps() → set[str]

Return the ids of all steps journaled as completed.

Together with the steps that have a committed record line this is the resume set: a resumed run skips these and re-runs everything else.

Return type:

set[str]

status() → dict[str, Any]

Summarize the run status by replaying the journal.

Returns:

A dict with state ("not_started", "running", "paused", "completed", "canceled" or "aborted"), completed and failed step counts, the operator’s reports, and the last recorded event.

Return type:

dict[str, Any]

_paused(events: list[dict[str, Any]]) → bool

Whether the last pause of an unfinished run was never resumed.

Parameters:

events (list[dict[str, Any]]) – The replayed events, oldest first.

Returns:

True while the run is holding at a step boundary.

Return type:

bool

plesty.lib.experiment.DEFAULT_GROUP = 'default'

plesty.lib.experiment.YAML_DTYPE = 'yaml'

class plesty.lib.experiment.EnvVar

One environment variable the run reads.

Variables:

• name – The variable, as the process environment spells it.

• description – What it is for, shown beside the field that sets it.

• required – Whether a run without it fails. An optional variable has a working default inside the experiment; a required one does not, and a caller can refuse to start rather than abort mid-plan.

name: str

description: str = ''

required: bool = True

to_dict() → dict[str, Any]

Return the variable as JSON primitives.

Return type:

dict[str, Any]

classmethod from_dict(raw: Mapping[str, Any]) → EnvVar

Rebuild a variable from to_dict() output.

Parameters:

raw (Mapping[str, Any]) – The mapping to read.

Raises:

ManifestError – If name is missing or empty.

Return type:

EnvVar

class plesty.lib.experiment.Manifest

What one experiment module declares about running it.

Variables:

• module – Import path served with python -m, plesty.pol_pl.

• title – Human title, shown wherever the experiment is named.

• schema – Configuration parameters by group. The ungrouped ones belong to DEFAULT_GROUP, as they do on a device.

• environment – Variables the run reads from the environment or .env.

• options – Extra command-line options beyond the standard set.

• config – Default configuration file, relative to the checkout.

• run_root – Where runs land when nothing overrides it.

• viz – Import path of the module’s viewer (python -m <viz> monitor), or None when it ships none.

• chooses_devices – Whether python -m <module> --devices --config <file> names the devices of setup.yaml a run of that configuration uses; without it every device is served (#62).

module: str

title: str

schema: Mapping[str, Sequence[plesty.lib.device.params.ConfigParameter]]

environment: Sequence[EnvVar] = ()

options: Sequence[Option] = ()

config: str = 'config/default.yaml'

run_root: str = 'runs'

viz: str | None = None

chooses_devices: bool = False

__post_init__() → None

Validate what a caller cannot recover from.

Raises:

ManifestError – If the module is unnamed, or a parameter is declared twice in one group.

Return type:

None

to_dict() → dict[str, Any]

Return the whole manifest as JSON primitives.

This is what --manifest prints: everything a caller needs to offer the experiment’s settings and start it, and nothing about how the experiment reaches its instruments.

Returns:

A mapping with module, title, config, run_root, viz, schema (grouped, as a device’s describe holds its parameters), environment, options and chooses_devices.

Return type:

dict[str, Any]

classmethod from_dict(raw: Mapping[str, Any]) → Manifest

Rebuild a manifest from a printed document.

The parameters come back as ConfigParameter objects carrying the declared type name rather than the type — the name is what crossed the wire, and what a renderer works from.

Parameters:

raw (Mapping[str, Any]) – A to_dict() document, as read back from --manifest.

Returns:

The manifest.

Raises:

ManifestError – If a required field is missing or malformed.

Return type:

Manifest

keys() → set[str]

Return every configuration key, written as a caller addresses it.

A grouped key is group.parameter; an ungrouped one is its bare name. This is the set a checker compares against the keys a shipped configuration file sets — a schema nobody keeps current is worse than none, because a console renders it as fact.

Returns:

The declared keys.

Return type:

set[str]

defaults() → dict[str, Any]

Return the declared defaults, shaped like the configuration file.

Groups become nested mappings and ungrouped parameters stay at the top level, so the result can be written out as a starting configuration. A parameter that declares no default is left out rather than written as null — an absent key takes the experiment’s own default, while a null overrides it with nothing.

Returns:

The configuration the defaults describe.

Return type:

dict[str, Any]

unknown_keys(config: Mapping[str, Any]) → list[str]

Return the keys config sets that the schema does not declare.

Only the two levels the schema describes are examined — a top-level scalar and a group’s members. What lies inside a yaml parameter is the experiment’s business and is not walked into.

Parameters:

config (Mapping[str, Any]) – A loaded configuration document.

Returns:

The undeclared keys, sorted, addressed as keys() writes them.

Return type:

list[str]

exception plesty.lib.experiment.ManifestError

Bases: ValueError

Raised when a manifest document is malformed or incomplete.

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

class plesty.lib.experiment.Option

One extra command-line option the module adds to the standard set.

Variables:

• flag – The option as it is typed, --prominence.

• description – What it does — the option’s own help text.

• dtype – Declared type name of its value; "bool" for a flag that takes none.

• sets – The configuration key the option overrides, written as a renderer addresses it (analysis.fit_options.prominence), or None when it changes something other than the configuration.

flag: str

description: str = ''

dtype: str = 'str'

sets: str | None = None

to_dict() → dict[str, Any]

Return the option as JSON primitives.

Return type:

dict[str, Any]

classmethod from_dict(raw: Mapping[str, Any]) → Option

Rebuild an option from to_dict() output.

Parameters:

raw (Mapping[str, Any]) – The mapping to read.

Raises:

ManifestError – If flag is missing or empty.

Return type:

Option

plesty.lib.experiment.env_vars(*names: str, **described: str) → tuple[EnvVar, Ellipsis]

Build required environment declarations.

Parameters:

• *names (str) – Variables with no description of their own.

• **described (str) – Variable name to description.

Returns:

One EnvVar per name, in the order given.

Return type:

tuple[EnvVar, Ellipsis]

plesty.lib.experiment.group_of(key: str) → tuple[str, str]

Split a configuration key into its group and parameter name.

Parameters:

key (str) – group.parameter, or a bare name for an ungrouped parameter.

Returns:

The group (DEFAULT_GROUP when the key carries none) and the parameter name.

Return type:

tuple[str, str]

plesty.lib.experiment.iter_parameters(schema: Mapping[str, Sequence[plesty.lib.device.params.ConfigParameter]]) → Iterable[tuple[str, plesty.lib.device.params.ConfigParameter]]

Yield every (group, parameter) pair of a schema, groups in order.

Parameters:

schema (Mapping[str, Sequence[plesty.lib.device.params.ConfigParameter]]) – The configuration schema.

Yields:

The group name and one of its parameters.

Return type:

Iterable[tuple[str, plesty.lib.device.params.ConfigParameter]]

plesty.lib.experiment.yaml_parameter(name: str, description: str, default: Any = None) → plesty.lib.device.params.ConfigParameter

Declare a configuration key whose value is a document, not a scalar.

Parameters:

• name (str) – The key.

• description (str) – What it holds.

• default (Any) – The default document, if it has one.

Returns:

The parameter, typed YAML_DTYPE.

Return type:

plesty.lib.device.params.ConfigParameter

exception plesty.lib.experiment.RigNotReadyError(problems: dict[str, str], message: str)

Bases: RuntimeError

The rig failed its preflight and the operator did not (or could not) recover it.

Variables:

problems – Sub-device name → one-line problem.

Parameters:

• problems (dict[str, str])

• message (str)

Keep the per-device problems next to the message.

problems

plesty.lib.experiment.connect_rig(build: collections.abc.Callable[[frozenset[str]], plesty.lib.device.composite_device.CompositeDevice], *, optional: collections.abc.Iterable[str] = (), ask: collections.abc.Callable[[str], str] | None = None, env_prefix: str | None = None, out: collections.abc.Callable[[str], None] = print) → tuple[plesty.lib.device.composite_device.CompositeDevice, frozenset[str]]

Build the rig, run its preflight, and ask the operator on a failure.

Parameters:

• build (collections.abc.Callable[[frozenset[str]], plesty.lib.device.composite_device.CompositeDevice]) – Factory building the composite without the given (optional) sub-devices; called again on every retry, so an edited .env or a freshly started server is picked up.

• optional (collections.abc.Iterable[str]) – Sub-devices the experiment can run without; when only these fail, the operator may choose to continue without them.

• ask (Optional[collections.abc.Callable[[str], str]]) – Reads the operator’s answer to a prompt. Defaults to input when stdin is a terminal, else to no prompt at all (the diagnosis is logged and RigNotReadyError raised).

• env_prefix (Optional[str]) – Deployment prefix of the <NAME>_ADDRESS variables, for the diagnosis.

• out (collections.abc.Callable[[str], None]) – Where the diagnosis and the prompt go (print).

Returns:

The ready composite and the set of optional sub-devices skipped.

Raises:

RigNotReadyError – If the operator aborts, or nobody can be asked.

Return type:

tuple[plesty.lib.device.composite_device.CompositeDevice, frozenset[str]]

class plesty.lib.experiment.Run(run_dir: str | pathlib.Path, *, data_root: str | None = None, replay_rows: int = 0)

Read access to one run directory: plan, journal, records, data paths.

Nothing here writes. The plan is read once; the journal is replayed on each status() because a run under way keeps appending to it.

Attach to a run directory.

Parameters:

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

• data_root (Optional[str]) – This machine’s view of the shared data disk; None takes PLESTY_DATA from the environment, then the root journaled by the machine that ran the experiment (only when that directory exists here).

• replay_rows (int) – How source() hands records out — 0 follows the run live, n rations a stored run n records per poll.

path

replay_rows = 0

_data_root = None

_plan: plesty.lib.experiment.schedule.Plan | None = None

_started: dict[str, Any] | None = None

sources: list[Any] = []

property id: str

The run id (the directory name).

Return type:

str

property name: str | None

The experiment name the run id carries, if it parses as one.

Return type:

Optional[str]

property started: datetime.datetime | None

The start time the run id carries, if it parses as one.

Return type:

Optional[datetime.datetime]

property plan: plesty.lib.experiment.schedule.Plan

The frozen plan (loaded once).

Return type:

plesty.lib.experiment.schedule.Plan

property config: dict[str, Any]

The frozen configuration; {} when the plan is unreadable.

Return type:

dict[str, Any]

property journal: plesty.lib.experiment.journal.Journal

The run’s journal.

Return type:

plesty.lib.experiment.journal.Journal

status() → dict[str, Any]

The run status by journal replay (see Journal.status()).

Return type:

dict[str, Any]

progress() → dict[str, Any]

How far the run has got, and how much longer it is likely to take.

Everything here is derived from what the run already wrote — the plan’s length, the journal’s step timings, and whatever a step in flight last reported — so watching costs a directory read and never disturbs the measurement.

The estimate is a mean over completed steps, which is honest for a sweep of like steps and wrong for a plan whose steps differ in kind; it is offered as None until at least one step has completed rather than as a number derived from nothing.

Returns:

A mapping with state, completed and failed step counts, total steps in the plan, fraction of the plan done, elapsed_s since the run started, step_s (the mean completed step), eta_s (None while unknowable), step (what the step in flight last reported, or None) and reports.

Return type:

dict[str, Any]

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

What the step in flight last said about itself, if anything.

A step is atomic and the journal speaks only at its boundaries, so a long exposure or a magnet ramp is indistinguishable from a hung run without this. A step that reports nothing simply has no entry.

Returns:

The reported progress — step_id, done, total, detail and the time it was written — or None.

Return type:

Optional[dict[str, Any]]

_elapsed() → float | None

Seconds between the run starting and its last event.

Returns:

The span, or None when the journal carries no usable times.

Return type:

Optional[float]

records() → Iterable[dict[str, Any]]

Iterate the committed step records, oldest first.

Return type:

Iterable[dict[str, Any]]

_run_started() → dict[str, Any]

Return type:

dict[str, Any]

property provenance: dict[str, Any]

What was running when the run started; {} for a run without it.

Runs written before this was journaled have none, and say so by being empty rather than by inventing today’s versions — which would be the one answer guaranteed wrong.

Return type:

dict[str, Any]

property raw_dir: str | None

<run_id>/raw, under each’s data path.

Type:

Where the run asked its devices to write

Return type:

Optional[str]

property data_root: str | None

The shared data disk as this machine sees it (see the class docstring).

Return type:

Optional[str]

local(path: str) → pathlib.Path

Return path, as a record spells it, as a path on this machine.

A path relative to the shared disk is joined to this machine’s view of it; an absolute one is taken as it is.

Parameters:

path (str)

Return type:

pathlib.Path

source(mapper: plesty.lib.monitor.sources.FrameMapper | None = None) → plesty.lib.monitor.sources.RunSource | plesty.lib.monitor.sources.ReplaySource

Return a source over this run’s records — live or replayed.

A RunSource when replay_rows is 0, else a ReplaySource handing out that many records per poll. Every source made here is kept in sources.

Make one source per consumer: a source hands each record out once, so two views polling the same source would each see half the rows. The mapper is what they share.

Parameters:

mapper (Optional[plesty.lib.monitor.sources.FrameMapper])

Return type:

Union[plesty.lib.monitor.sources.RunSource, plesty.lib.monitor.sources.ReplaySource]

property exhausted: bool

Whether every source made here has handed out its last record.

Live sources never are (the next record may still be coming); a run without sources counts as exhausted so a watcher does not wait forever.

Return type:

bool

__repr__() → str

Return Run('<path>').

Return type:

str

plesty.lib.experiment.check_run_id(text: str) → str

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

A run id becomes a directory, and is written into journals and records that outlive the machine that made them. A name with a separator in it would put the run somewhere else entirely; one with a space or a colon travels badly between filesystems. Letters, digits, dot, dash and underscore are what every filesystem agrees on.

Parameters:

text (str) – The proposed run id.

Returns:

The same string.

Raises:

ValueError – If it is empty, a relative path, or carries a character that does not belong in a directory name.

Return type:

str

plesty.lib.experiment.latest_run(run_root: str | pathlib.Path, name: str | Sequence[str] | None = None) → pathlib.Path

Return the most recently started run under run_root.

Raises:

FileNotFoundError – If there is no such run yet — a viewer opened before its experiment is a mistake worth naming.

Parameters:

• run_root (str | pathlib.Path)

• name (Union[str, Sequence[str], None])

Return type:

pathlib.Path

plesty.lib.experiment.list_runs(run_root: str | pathlib.Path, name: str | Sequence[str] | None = None) → list[pathlib.Path]

Return the run directories under run_root, oldest first.

Parameters:

• run_root (str | pathlib.Path) – The directory runs are created in.

• name (Union[str, Sequence[str], None]) – Keep only runs of this experiment name (or any of several — an experiment that was renamed still owns its old runs). None keeps every run.

Return type:

list[pathlib.Path]

plesty.lib.experiment.parse_run_id(text: str) → tuple[str, datetime.datetime] | None

Split a run id into its experiment name and start time.

Parameters:

text (str) – A directory name.

Returns:

(name, started), or None when text is not a run id — the name may itself contain underscores, so the split is at the last one.

Return type:

Optional[tuple[str, datetime.datetime]]

plesty.lib.experiment.run_id(name: str, when: datetime.datetime | None = None) → str

Return the run id for experiment name started at when (now by default).

Parameters:

• name (str)

• when (Optional[datetime.datetime])

Return type:

str

class plesty.lib.experiment.Plan(steps: list[Step], config: dict[str, Any] | None = None)

A frozen, hashable schedule of atomic steps plus the run configuration.

Usage:

plan = Plan(
    steps=[Step(id=f"scan[{i}]", op="scan_point", params={"x": i}) for i in range(10)],
    config={"exposure_s": 0.1},
)
plan.save("runs/run_001/plan.json")

Initialize the plan with its steps and optional configuration.

Parameters:

• steps (list[Step]) – Ordered atomic steps; step ids must be unique.

• config (dict[str, Any] | None) – Experiment configuration recorded alongside the schedule.

Raises:

ValueError – If two steps share the same id.

steps: list[Step]

config: dict[str, Any]

to_dict() → dict[str, Any]

Return a JSON-serializable representation of the plan.

Return type:

dict[str, Any]

classmethod from_dict(data: dict[str, Any]) → Plan

Reconstruct a plan from to_dict() output.

Parameters:

data (dict[str, Any])

Return type:

Plan

content_hash() → str

Return a stable SHA-256 hash of the schedule and configuration.

Used on resume to verify that the persisted plan matches the plan the experiment would generate now — resuming under changed parameters is refused rather than silently mixing two schedules.

Return type:

str

save(path: str | pathlib.Path) → pathlib.Path

Write the plan as JSON to path, creating parent directories.

Parameters:

path (str | pathlib.Path)

Return type:

pathlib.Path

classmethod load(path: str | pathlib.Path) → Plan

Load a plan previously written by save().

Parameters:

path (str | pathlib.Path)

Return type:

Plan

__len__() → int

Return the number of steps in the plan.

Return type:

int

class plesty.lib.experiment.Step

One atomic measurement step in an experiment plan.

Variables:

• id – Stable, deterministic identifier unique within the plan, e.g. "scan[x=3,y=5]". Resume matches completed steps by this id, so it must not depend on run time or randomness.

• op – Name of the experiment method to call for this step.

• params – Keyword arguments passed to the method.

id: str

op: str

params: dict[str, Any]

to_dict() → dict[str, Any]

Return a JSON-serializable representation of the step.

Return type:

dict[str, Any]

classmethod from_dict(data: dict[str, Any]) → Step

Reconstruct a step from to_dict() output.

Parameters:

data (dict[str, Any])

Return type:

Step

plesty.lib.experiment.SETUP_FILE = 'setup.yaml'

class plesty.lib.experiment.Bench

One bench of a setup.

As a setup writes it, each field is the ${VARIABLE} holding the value; as Setup.resolve() gives it, the value itself.

Variables:

• name – What devices call it in host_name.

• address – Where its bench agent listens: host, host:port or a ZMQ endpoint. Empty for the machine the run is on.

• token – The secret that agent checks; empty when it checks none.

• data – Where it sees the lab’s shared data disk; empty when unset.

name: str

address: str = ''

token: str = ''

data: str = ''

property is_local: bool

the bench with no address.

Type:

Whether it is the machine the run is on

Return type:

bool

property host: str

The machine, without the agent’s port or scheme; 127.0.0.1 for the local one.

Return type:

str

to_dict() → dict[str, str]

The bench’s entry, without its name; empty fields left out.

Return type:

dict[str, str]

classmethod from_dict(name: str, raw: Any) → Bench

Read the bench name of a setup.

Raises:

SetupError – When the entry is not a mapping, has a key a bench does not take, or writes a value where a variable belongs.

Parameters:

• name (str)

• raw (Any)

Return type:

Bench

class plesty.lib.experiment.Setup

The benches and devices of a module’s setup.yaml.

Variables:

• benches – By name, as written — each field the variable holding it.

• devices – By name, in file order.

benches: collections.abc.Mapping[str, Bench]

devices: collections.abc.Mapping[str, SetupDevice]

classmethod from_dict(raw: Any) → Setup

Read a parsed setup.yaml.

Raises:

SetupError – When it is not a mapping of benches and devices, a bench or device entry is malformed, more than one bench has no address, or a device names a bench the setup does not declare.

Parameters:

raw (Any)

Return type:

Setup

classmethod load(root: str | os.PathLike[str] = '.') → Setup

Read the setup.yaml of the module checkout at root.

Raises:

SetupError – When there is none, or it does not parse.

Parameters:

root (str | os.PathLike[str])

Return type:

Setup

only(devices: collections.abc.Iterable[str]) → Setup

This setup reduced to devices and the benches they run on (#62).

A run that uses some of a module’s devices needs nothing of the benches serving only the others: their addresses, tokens and data need not be set. The bench without an address — this machine, where the run reads PLESTY_DATA — is always kept.

Raises:

SetupError – When devices names a device the setup does not have.

Parameters:

devices (collections.abc.Iterable[str])

Return type:

Setup

variables() → list[str]

Every variable the benches name, in file order.

Return type:

list[str]

unset(env: collections.abc.Mapping[str, str]) → list[str]

The variables the benches name that env leaves empty.

Parameters:

env (collections.abc.Mapping[str, str])

Return type:

list[str]

resolve(env: collections.abc.Mapping[str, str]) → dict[str, Bench]

The benches with each field’s value read from env; empty where unset.

Parameters:

env (collections.abc.Mapping[str, str])

Return type:

dict[str, Bench]

local_name() → str

The bench the file writes without an address; local when it declares none.

Read from the file, not from the values: a bench whose address variable is unset has an empty address once resolved, and is still another machine.

Return type:

str

bench_of(device: str, env: collections.abc.Mapping[str, str]) → Bench

The bench device runs on, with its values read from env.

A device naming no bench — or local when the setup declares no bench for this machine — runs on the machine the run is on: the setup’s bench without an address, or one called local when it declares none.

Raises:

• KeyError – When the setup has no such device.

• SetupError – When the device’s bench is another machine whose address env does not set.

Parameters:

• device (str)

• env (collections.abc.Mapping[str, str])

Return type:

Bench

environment(env: collections.abc.Mapping[str, str]) → dict[str, str]

What a run on this machine is given, built from the setup and the values in env.

A device’s address is not among it: its bench starts it, picks its port, and gives the run <DEVICE>_ADDRESS.

Parameters:

env (collections.abc.Mapping[str, str]) – The run’s environment, .env included — where the values the benches name are read.

Returns:

PLESTY_DATA — where this machine sees the shared data disk — when its bench says so; else nothing.

Return type:

dict[str, str]

record(env: collections.abc.Mapping[str, str]) → dict[str, dict[str, str]]

Where each device was for a run: what the run journals when it starts.

The setup says which bench a device runs on, and only the run’s environment says the rest — the address its bench served it on, which changes from start to start, and where that bench sees the shared disk. Nothing is refused here: a run that got this far connects with what it was given, and the record says what that was. A bench’s token is never recorded.

Parameters:

env (collections.abc.Mapping[str, str]) – The run’s environment, .env included.

Returns:

bench, the bench it runs on; address, the run’s <DEVICE>_ADDRESS; data, that bench’s shared point in its own spelling. A value nobody set is empty.

Return type:

By device, in file order

class plesty.lib.experiment.SetupDevice

What a setup says about one device, as far as a run is concerned.

Variables:

• name – The device’s name; the run reads its address from <NAME>_ADDRESS, upper-cased, which its bench provides.

• host_name – The bench it runs on; empty for the machine the run is on.

name: str

host_name: str = ''

classmethod from_dict(name: str, raw: Any) → SetupDevice

Read what a run needs of the device entry name; its other keys are the bench’s.

Raises:

SetupError – When the entry is not a mapping.

Parameters:

• name (str)

• raw (Any)

Return type:

SetupDevice

exception plesty.lib.experiment.SetupError

Bases: ValueError

Raised when a setup cannot be read, or writes a value where a variable belongs.

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

exception plesty.lib.experiment.UnsetVariables(names: collections.abc.Sequence[str], env: pathlib.Path)

Bases: SetupError

Raised when the setup names variables nothing gives a value.

Variables:

• names – The variables, in the order the setup names them.

• env – The .env file they are set in.

Parameters:

• names (collections.abc.Sequence[str])

• env (pathlib.Path)

Say which variables to set, and where.

names

env

plesty.lib.experiment.apply_setup(root: str | os.PathLike[str] = '.', require: bool = True, devices: collections.abc.Iterable[str] | None = None) → dict[str, str]

Put what the setup at root builds into os.environ, under what is set already.

The checkout’s .env is where the values the setup names are set: a fresh checkout has none, so it is made first (seed_env()), with a line for each. A variable already in the environment — set by hand, in .env, or by a bench that started the run — is left as it is.

Parameters:

• root (str | os.PathLike[str]) – The module checkout.

• require (bool) – Refuse when a variable the setup names has no value, rather than build what can be built. A run that connects to nothing — printing its plan — needs none of them.

• devices (Optional[collections.abc.Iterable[str]]) – The setup’s devices this run uses (Setup.only()); None for all of them.

Returns:

The variables this set.

Raises:

• UnsetVariables – When require and a variable the setup names is empty — before anything connects, naming them and the file.

• SetupError – When the setup does not parse.

Return type:

dict[str, str]

plesty.lib.experiment.seed_env(root: str | os.PathLike[str] = '.') → list[str]

Give the checkout at root a .env line for every variable its setup names.

A variable already there is left as it is; a new one takes the value the module’s .env.example suggests, else none — to be filled in. A module with no setup gets nothing.

Returns:

The variables added.

Raises:

SetupError – When the setup does not parse.

Parameters:

root (str | os.PathLike[str])

Return type:

list[str]

plesty.lib.experiment.split_benches(benches: collections.abc.Sequence[Bench], written: collections.abc.Mapping[str, Bench] | None = None) → tuple[dict[str, dict[str, str]], dict[str, str]]

Split benches given by value into what setup.yaml says and what .env holds.

A field keeps the variable the setup already names for it; a new one gets variable_for(). An empty field is left out of both.

Parameters:

• benches (collections.abc.Sequence[Bench]) – The benches, each field its value.

• written (Optional[collections.abc.Mapping[str, Bench]]) – The benches as the setup writes them now.

Returns:

benches: for the setup — name → field → ${VARIABLE} — and variable → value for .env.

Return type:

tuple[dict[str, dict[str, str]], dict[str, str]]

plesty.lib.experiment.variable_of(written: str) → str

The variable a setup’s written field names — SPEC_PC_TOKEN for ${SPEC_PC_TOKEN}.

Raises:

SetupError – When written is a value rather than a variable.

Parameters:

written (str)

Return type:

str