The Experiment Class

plesty.lib.experiment.Experiment is an async abstract base class that turns a measurement into an atomic, journaled, resumable schedule. You describe what to measure; the base class owns how it executes — journaling, retries, checkpointing, and resume.

What a subclass implements

A subclass overrides exactly three lifecycle hooks, plus one public method per step operation:

Member	Required	Role
`build_plan()`	yes	Return the `Plan`: the atomic measurement schedule plus the frozen config. Must be deterministic — resume verifies the regenerated plan against the persisted one by content hash.
`setup()`	no (async)	Prepare the run. The default connects all devices; extend with `await super().setup()`.
`teardown()`	no (async)	Release resources. The default disconnects all devices. Always awaited — whether the run completes, fails, or is canceled.
step operations	one per `Step.op`	Public methods (sync or async) executed per step; the return value is persisted. See Plan & step operations.

Step operations are validated before the run starts: every Step.op must name an existing public method on the experiment, and the lifecycle names (build_plan, run, setup, teardown) are reserved.

Constructor

Experiment(
    devices=None,       # CompositeDevice with all instruments; connected in setup()
    run_root="runs",    # directory under which each run creates its checkpoint dir
    name=None,          # used in run ids; defaults to the subclass name lowercased
    max_retries=3,      # attempts per step before the run aborts
    retry_sleep=3.0,    # seconds between step retries
)

Device access goes exclusively through the CompositeDevice passed at construction — an experiment receives device clients, never raw hardware. Plan construction is device-free by contract: build_plan() must work without any devices attached.

Running

run_id = asyncio.run(experiment.run())            # new run
asyncio.run(experiment.run(resume=run_id))        # continue an interrupted run

run() executes the plan step by step, journaling every event. On resume, steps journaled as completed are skipped; a run whose regenerated plan no longer matches the stored one is refused with PlanMismatchError. Failed steps are retried up to max_retries times before the run aborts. See Runs, data & resume for what lands on disk.

Minimal runnable subclass

No hardware required:

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))

The demo experiment is the full reference version of this pattern — a simulated Gaussian line scan with async step operations, reproducible pseudo-noise, and a CLI with --resume.