plesty.lib.experiment.base_experiment

Asynchronous Experiment ABC with plan execution and checkpoint/resume.

An Experiment orchestrates one or more Devices to execute a measurement routine (issue plesty-lib#3). Its public lifecycle is setup() → run() → teardown(), all asynchronous; run() is a template method that executes the experiment’s Plan of atomic steps with crash-safe status logging and resume — analogous to checkpointing in deep learning: the plan is the training schedule, the journal is the training log, and the completed-step set plus persisted step results form the checkpoint.

Every run owns a directory (the convention is spelled out once, in plesty.lib.experiment.runs, which also reads it back):

<run_root>/<run_id>/            # run_id = <name>_<YYYYmmdd-HHMMSS>
├── plan.json                   # frozen schedule + config + content hash
├── journal.jsonl               # append-only event log (status, resume source)
├── records.jsonl               # one line per completed step: its result document
└── data/                       # blobs the records reference (arrays, images)
    └── step_0003.npy

Blobs written by devices themselves (frames, images) never cross the network: they go to the Plesty share, into <run_id>/raw under each device’s bench folder (Experiment.raw_dir, Experiment.route()), and the device reports its folder relative to the share. Every process is told where the share is on its own machine — PLESTY_DATA — and joins a reported path to it, so any machine reads the run back. With PLESTY_DATA set, run_root defaults to it: runs land on the share too.

Minimal runnable example (mock, no hardware):

import asyncio
from plesty.lib.data import PlestyArray
from plesty.lib.experiment import Experiment, Plan, Step

class LineScan(Experiment):
    def build_plan(self) -> Plan:
        steps = [
            Step(id=f"scan[x={x}]", op="scan_point", params={"x": x})
            for x in range(5)
        ]
        return Plan(steps, config={"points": 5})

    def scan_point(self, x: int) -> PlestyArray:
        return PlestyArray([float(x)], name="signal", unit="a.u.")

run_id = asyncio.run(LineScan(run_root="runs").run())
# ... interrupted? Resume, skipping completed steps:
asyncio.run(LineScan(run_root="runs").run(resume=run_id))

Attributes

_RESERVED_OPS

Exceptions

RunStopped

Raised inside a run when the driver asked it to stop.

PlanMismatchError

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

InvalidOperationError

Raised when a plan step references an invalid experiment operation.

Classes

Experiment

Base class for PLESTY experiments with atomic scheduling and resume.

Module Contents

exception plesty.lib.experiment.base_experiment.RunStopped(kind: str, reason: str = '')

Bases: RuntimeError

Raised inside a run when the driver asked it to stop.

Variables:

  • kind – CANCEL when the step in flight was allowed to finish, ABORT when it was interrupted.

  • reason – What the driver gave as the reason, if any.

Parameters:

  • kind (str)

  • reason (str)

Record which kind of stop was asked for, and why.

kind
reason = ''
exception plesty.lib.experiment.base_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.

exception plesty.lib.experiment.base_experiment.InvalidOperationError

Bases: RuntimeError

Raised when a plan step references an invalid experiment operation.

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

plesty.lib.experiment.base_experiment._RESERVED_OPS
class plesty.lib.experiment.base_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:
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:
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:
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:
Return type:

None