The Experiment Framework

The device → analyzer → experiment pipeline finally has its top tier. Commit 40d68e6 adds an asynchronous Experiment base class to plesty.lib.experiment: a run is a frozen plan of atomic steps recorded in a crash-safe journal, so any measurement can stop and resume exactly where it left off.

If someone hands you an experiment, running it is two lines. A long scan that crashes, is canceled, or loses its instrument connection resumes without repeating a single completed step:

import asyncio
from my_experiments import LineScan

# start a fresh run — returns the run id
run_id = asyncio.run(LineScan(run_root="runs").run())

# interrupted? resume by id — only the unfinished steps execute
asyncio.run(LineScan(run_root="runs").run(resume=run_id))

Each run owns a self-describing directory — reproducible and inspectable long after it finishes:

plan.json      # the frozen schedule + config + content hash
journal.jsonl  # append-only event log — the source of truth for status
data/          # per-step results: raw blob + JSON metadata
  step_0000.json
  step_0000.npy

• run() Starts the plan, journalling each step and retrying failures up to max_retries.
• run(resume=id) Replays the journal and executes only the steps that never completed.
• status Always derived by replaying journal.jsonl — never a mutable file a crash could corrupt.

To add an experiment to the hub, subclass Experiment, implement build_plan() (it must be deterministic — its content hash gates resume), and add one method per step op. Device access always goes through a CompositeDevice; an experiment receives Device instances, never raw 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.")

• build_plan() The atomic schedule. Deterministic for a given config — its hash gates resume.
• step methods One per Step.op; return a result that save_result persists as blob + metadata.
• setup / teardown Async hooks — connect the CompositeDevice before the plan, disconnect after.

Before it ships, an experiment must pass the enforced baseline: plan-step validation before a run starts (27a16c2) and the ExperimentPipeline standard contract test suite (lib#14). Verify it locally the same way as any module:

uv run plesty check   # includes gate E1 — experiment contract validation