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 a shared disk under <data_dir>/<run_id>/raw, in the acquiring host’s spelling of the share (Experiment.raw_dir). The share is named once, by environment — PLESTY_DATA_DIR as the host sees it, PLESTY_DATA_MOUNT as this machine sees it — and both spellings are journaled with run_started so any machine can read the run back. With the mount 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

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.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_dir: str | None = None, data_mount: 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; None takes PLESTY_DATA_MOUNT from the environment (runs land on the share) and falls back to runs.

  • 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_dir (Optional[str]) – The shared data root as the acquiring host sees it; None takes PLESTY_DATA_DIR. Device blobs go to raw_dir beneath it.

  • data_mount (Optional[str]) – The same root as this machine sees it; None takes PLESTY_DATA_MOUNT. Journaled for readers.

devices = None
data_dir
data_mount
run_root
name = ''
max_retries = 3
retry_sleep = 3.0
run_id: str | None = None
run_dir: pathlib.Path | None = None
logger
_plan_length = 0
property raw_dir: str | None

Where this run’s device-written blobs go, in the host’s spelling.

<data_dir>/<run_id>/raw — 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 without a data root or before a run has started.

Return type:

Optional[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) → 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.

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.

  • 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) → 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.

Parameters:
Return type:

tuple[str, set[str]]

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