Experiment Framework

plesty.lib.experiment turns a measurement routine into a reproducible, crash-tolerant run. The mental model is checkpointed training 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.

An Experiment orchestrates devices through a CompositeDevice — it receives Device instances, never raw hardware.

The pieces

Piece	Role
Step	The smallest unit of work: a stable id, an operation name op, and its params. A step either completes (result persisted and journaled) or is re-run on resume.
Plan	The frozen, ordered schedule of steps plus the run configuration. Written to disk at run start, never mutated; a SHA-256 content_hash() identifies the schedule.
Journal	A crash-safe, append-only JSONL event log (run_started, step_started, step_completed, step_failed, run_completed/_canceled/_aborted). Replaying it derives the run state and the completed-step set.
records.jsonl	One line per completed step: its result document (plesty.lib.data.append_record / read_records). One file, appended and fsynced per step, tailed by offset — not one small file per step, which is slow to write and slow to follow on a network share.
Experiment	The async ABC tying it together: validate → journal → execute with retries → persist results → resume.
Run	Read access to a stored run — plan, config, journal status, records, and the translation of device-written paths onto this machine (plesty.lib.experiment.runs, which also spells out the naming below and finds runs: list_runs, latest_run).

Every run owns a directory:

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

Each record line is a typed ResultDocument with provenance (step_id, op, params) plus the step index; a JSON value result is inlined, an array or image stays a raw blob in data/ that the line references. The line is the commit — it is appended (and fsynced) only after the blob is on disk, and a resumed run treats a committed record as a completed step even if the crash came before its journal event. Runs written by plesty-lib ≤ 0.3.4 (one data/step_*.json per step) are still read by RunSource, records(), and Run.

Writing an experiment

Subclasses override exactly three lifecycle hooks — build_plan() (required), setup()/teardown() (optional, extend the defaults which connect/disconnect all devices) — plus one public method per step operation:

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:
        # drive devices via self.devices, return the measurement
        return PlestyArray([float(x)], name="signal", unit="a.u.")

run_id = asyncio.run(LineScan(run_root="runs").run())

Rules that keep runs reproducible:

• build_plan() must be deterministic for a given configuration — resume compares the regenerated plan against the persisted one by content hash.

• Step ids must be stable (e.g. "scan[x=3,y=5]") — never derived from run time or randomness; resume matches completed steps by id.

• Step operations are validated before anything runs: every Step.op must name an existing public method and may not be a reserved lifecycle method (build_plan, run, setup, teardown). Violations raise InvalidOperationError listing all offending steps — before any device is touched or journal written.

Step methods may be sync or async; results are awaited automatically. Each step is retried (max_retries, retry_sleep constructor arguments) before the run aborts.

Starting on a rig that is really there

An experiment’s __main__ builds the rig with connect_rig — the same start for every experiment: connect the servers, run the composite’s preflight() (each sub-device exposes the methods its configuration requires and answers identity()), and if anything is missing, tell the operator which device, at which address, from which variable — and ask:

from plesty.lib.experiment import connect_rig

rig, skipped = connect_rig(
    lambda skip: PolPlRig(with_powermeter="pm" not in skip),
    optional={"pm"},                       # the experiment can run without it
)

The rig is not ready — 1 device(s) failed the preflight:
  - pm: no answer from tcp://127.0.0.1:5557: TimeoutError('Server did not respond in time')
      address tcp://127.0.0.1:5557 (from PM_ADDRESS in .env / the environment)
Check that the pm server(s) are running and that PM_ADDRESS point at them
(a wrong port connects fine but is not the expected device).
[r] retry (after fixing .env or starting the server)  [a] abort  [s] continue without pm
>

r rebuilds the rig (an edited .env or a freshly started server is picked up), a raises RigNotReadyError, s — offered only when every failing sub-device is in optional — returns the rig without them and names them in skipped, so the experiment can drop the matching part of its plan. Without a terminal on stdin nothing waits: the diagnosis is logged and RigNotReadyError raised.

The check also runs inside Experiment.setup(): a run never starts on a composite whose preflight() reports a problem, whether or not connect_rig was used. Declare what each sub-device must be able to do in its configuration — that is what turns “the port answers” into “it is the stage”:

devices:
  hwp:  {timeout_ms: 5000, requires: [home_stage, move_absolute, get_position]}
  spec: {timeout_ms: 5000, requires: [acquire, get_recent_file, write, query]}
  pm:   {timeout_ms: 5000, requires: [measure_power, write]}

Data on a shared disk

Blobs a device writes itself (spectrometer frames, camera images) never cross the network: the acquiring host saves them to a shared disk and the step record carries only the path. Two machines spell that share differently, so it is named once, by environment:

Variable	Meaning
PLESTY_DATA_DIR	the share as the acquiring host sees it (e.g. G:\RAWDATA)
PLESTY_DATA_MOUNT	the same share as this machine sees it (e.g. /mnt/group/RAWDATA)

The experiment reads both (or takes data_dir= / data_mount= at construction). With the mount set, run_root defaults to it, so runs land on the share as well. Before setup() is awaited the run id is known and self.raw_dir names <data_dir>/<run_id>/raw in the host’s spelling — one folder per run, the same one on resume — which setup() routes the writing devices to:

async def setup(self) -> None:
    await super().setup()
    if self.raw_dir is not None:
        self.devices.set_data_path(self.raw_dir, devices=["spec"])

run_started journals data_dir, data_mount, and raw_dir, so a reader needs at most its own mount:

from plesty.lib.experiment import Run, latest_run

run = Run(latest_run("/mnt/group/RAWDATA", name="pol_pl"))
run.config["powermeter"]                 # the frozen configuration
run.status()["state"]                    # "running", "completed", ...
for record in run.records():
    run.local(record["data_file"])       # G:\RAWDATA\... → /mnt/group/RAWDATA/...

Run.local uses, in order, an explicit mount=, PLESTY_DATA_MOUNT, and the mount journaled by the machine that ran the experiment (only when that directory exists here); with neither a data root nor a mount the path is returned as recorded — the case of reading on the acquiring host itself. run.source(mapper) returns a RunSource (live) or, with Run(..., replay_rows=n), a ReplaySource — see Monitors.

Crash, cancel, resume

run() returns the run id. If the run is interrupted — Ctrl-C, a crash, an aborting step — pass the id back to continue:

run_id = asyncio.run(LineScan(run_root="runs").run())
# ... interrupted?
asyncio.run(LineScan(run_root="runs").run(resume=run_id))

On resume the framework:

1. regenerates the plan and refuses to continue if its content hash differs from the stored one (PlanMismatchError) — a silently changed schedule or configuration cannot corrupt a run;

2. replays the journal to find completed steps and skips them;

3. journals the resume (run_started with resumed: true) and continues.

teardown() is always awaited — on completion, failure, and cancellation — so devices are released no matter how the run ends.

Contract tests without hardware

plesty.lib.test.experiment_pipeline.ExperimentPipeline verifies the plan/checkpoint contract of an experiment module with no instrument attached (it never calls run()). Five gates:

Gate	Checks
1 test_experiment_subclass	public Experiment subclass, instantiable without devices
2 test_plan_deterministic	build_plan() twice → identical content_hash()
3 test_plan_ops_resolve	every Step.op resolves via the framework’s own plan validation
4 test_plan_serializable	Plan.save/load round-trip preserves the hash
5 test_lifecycle_hooks	setup/teardown overrides are async

Experiment HUB modules expose one pytest function per gate (SDK gate E1 verifies their presence and passage — the experiment-tier mirror of device gate d1); see Test Helpers for the pattern.