plesty.lib.experiment

Experiment orchestration: plans, journals, and the async Experiment ABC.

Submodules

Exceptions

InvalidOperationError

Raised when a plan step references an invalid experiment operation.

PlanMismatchError

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

RigNotReadyError

The rig failed its preflight and the operator did not (or could not) recover it.

Classes

Experiment

Base class for PLESTY experiments with atomic scheduling and resume.

Journal

Crash-safe append-only event log for one experiment run.

Run

Read access to one run directory: plan, journal, records, data paths.

Plan

A frozen, hashable schedule of atomic steps plus the run configuration.

Step

One atomic measurement step in an experiment plan.

Functions

connect_rig(, ask, str]] = None, env_prefix, out, ...)

Build the rig, run its preflight, and ask the operator on a failure.

latest_run(→ pathlib.Path)

Return the most recently started run under run_root.

list_runs(→ list[pathlib.Path])

Return the run directories under run_root, oldest first.

parse_run_id(→ Optional[tuple[str, datetime.datetime]])

Split a run id into its experiment name and start time.

run_id(→ str)

Return the run id for experiment name started at when (now by default).

Package Contents

class plesty.lib.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

exception plesty.lib.experiment.InvalidOperationError

Bases: RuntimeError

Raised when a plan step references an invalid experiment operation.

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

exception plesty.lib.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.

class plesty.lib.experiment.Journal(path: str | pathlib.Path)

Crash-safe append-only event log for one experiment run.

Attach the journal to path, creating parent directories.

Parameters:

path (str | pathlib.Path) – Location of the journal.jsonl file; appended to if it already exists (the resume case).

TERMINAL_EVENTS = ('run_completed', 'run_canceled', 'run_aborted')
path
append(event: str, **fields: Any) → dict[str, Any]

Append one event line and fsync it to disk.

Parameters:

  • event (str) – Event name, e.g. "step_completed".

  • fields (Any) – Additional payload stored on the event (step id, result path, error text, …).

Returns:

The full event record that was written.

Return type:

dict[str, Any]

replay() → list[dict[str, Any]]

Read all events back, tolerating a torn final line after a crash.

Returns:

The recorded events in write order; an empty list if the journal file does not exist yet.

Return type:

list[dict[str, Any]]

completed_steps() → set[str]

Return the ids of all steps journaled as completed.

Together with the steps that have a committed record line this is the resume set: a resumed run skips these and re-runs everything else.

Return type:

set[str]

status() → dict[str, Any]

Summarize the run status by replaying the journal.

Returns:

A dict with state ("not_started", "running", "completed", "canceled" or "aborted"), completed and failed step counts, and the last recorded event.

Return type:

dict[str, Any]

exception plesty.lib.experiment.RigNotReadyError(problems: dict[str, str], message: str)

Bases: RuntimeError

The rig failed its preflight and the operator did not (or could not) recover it.

Variables:

problems – Sub-device name → one-line problem.

Parameters:

  • problems (dict[str, str])

  • message (str)

Keep the per-device problems next to the message.

problems
plesty.lib.experiment.connect_rig(build: collections.abc.Callable[[frozenset[str]], plesty.lib.device.composite_device.CompositeDevice], *, optional: collections.abc.Iterable[str] = (), ask: collections.abc.Callable[[str], str] | None = None, env_prefix: str | None = None, out: collections.abc.Callable[[str], None] = print) → tuple[plesty.lib.device.composite_device.CompositeDevice, frozenset[str]]

Build the rig, run its preflight, and ask the operator on a failure.

Parameters:

  • build (collections.abc.Callable[[frozenset[str]], plesty.lib.device.composite_device.CompositeDevice]) – Factory building the composite without the given (optional) sub-devices; called again on every retry, so an edited .env or a freshly started server is picked up.

  • optional (collections.abc.Iterable[str]) – Sub-devices the experiment can run without; when only these fail, the operator may choose to continue without them.

  • ask (Optional[collections.abc.Callable[[str], str]]) – Reads the operator’s answer to a prompt. Defaults to input when stdin is a terminal, else to no prompt at all (the diagnosis is logged and RigNotReadyError raised).

  • env_prefix (Optional[str]) – Deployment prefix of the <NAME>_ADDRESS variables, for the diagnosis.

  • out (collections.abc.Callable[[str], None]) – Where the diagnosis and the prompt go (print).

Returns:

The ready composite and the set of optional sub-devices skipped.

Raises:

RigNotReadyError – If the operator aborts, or nobody can be asked.

Return type:

tuple[plesty.lib.device.composite_device.CompositeDevice, frozenset[str]]

class plesty.lib.experiment.Run(run_dir: str | pathlib.Path, *, mount: str | None = None, data_dir: str | None = None, replay_rows: int = 0)

Read access to one run directory: plan, journal, records, data paths.

Nothing here writes. The plan is read once; the journal is replayed on each status() because a run under way keeps appending to it.

Attach to a run directory.

Parameters:

  • run_dir (str | pathlib.Path) – The run directory.

  • mount (Optional[str]) – This machine’s view of the shared data root; None takes PLESTY_DATA_MOUNT from the environment, then the mount journaled by the machine that ran the experiment (only when that directory exists here).

  • data_dir (Optional[str]) – The shared root as the acquiring host sees it — only for runs that did not record it themselves.

  • replay_rows (int) – How source() hands records out — 0 follows the run live, n rations a stored run n records per poll.

path
replay_rows = 0
_mount = None
_data_dir = None
_plan: plesty.lib.experiment.schedule.Plan | None = None
_started: dict[str, Any] | None = None
sources: list[Any] = []
property id: str

The run id (the directory name).

Return type:

str

property name: str | None

The experiment name the run id carries, if it parses as one.

Return type:

Optional[str]

property started: datetime.datetime | None

The start time the run id carries, if it parses as one.

Return type:

Optional[datetime.datetime]

property plan: plesty.lib.experiment.schedule.Plan

The frozen plan (loaded once).

Return type:

plesty.lib.experiment.schedule.Plan

property config: dict[str, Any]

The frozen configuration; {} when the plan is unreadable.

Return type:

dict[str, Any]

property journal: plesty.lib.experiment.journal.Journal

The run’s journal.

Return type:

plesty.lib.experiment.journal.Journal

status() → dict[str, Any]

The run status by journal replay (see Journal.status()).

Return type:

dict[str, Any]

records() → Iterable[dict[str, Any]]

Iterate the committed step records, oldest first.

Return type:

Iterable[dict[str, Any]]

_run_started() → dict[str, Any]
Return type:

dict[str, Any]

property data_dir: str | None

The shared root as the acquiring host sees it.

Journaled by the run; falls back to the plan’s top-level data_dir for runs written before the journal carried it, then to the data_dir given at construction.

Return type:

Optional[str]

property raw_dir: str | None

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

Return type:

Optional[str]

property data_mount: str | None

The shared root as this machine sees it (see the class docstring).

Return type:

Optional[str]

local(host_path: str) → pathlib.Path

Return host_path (as a record spells it) as a path on this machine.

Untranslated when the run has no data root or this machine has no mount for it — the case of a viewer on the acquiring host itself.

Parameters:

host_path (str)

Return type:

pathlib.Path

source(mapper: plesty.lib.monitor.sources.FrameMapper | None = None) → plesty.lib.monitor.sources.RunSource | plesty.lib.monitor.sources.ReplaySource

Return a source over this run’s records — live or replayed.

A RunSource when replay_rows is 0, else a ReplaySource handing out that many records per poll. Every source made here is kept in sources.

Make one source per consumer: a source hands each record out once, so two views polling the same source would each see half the rows. The mapper is what they share.

Parameters:

mapper (Optional[plesty.lib.monitor.sources.FrameMapper])

Return type:

Union[plesty.lib.monitor.sources.RunSource, plesty.lib.monitor.sources.ReplaySource]

property exhausted: bool

Whether every source made here has handed out its last record.

Live sources never are (the next record may still be coming); a run without sources counts as exhausted so a watcher does not wait forever.

Return type:

bool

__repr__() → str

Return Run('<path>').

Return type:

str

plesty.lib.experiment.latest_run(run_root: str | pathlib.Path, name: str | Sequence[str] | None = None) → pathlib.Path

Return the most recently started run under run_root.

Raises:

FileNotFoundError – If there is no such run yet — a viewer opened before its experiment is a mistake worth naming.

Parameters:

  • run_root (str | pathlib.Path)

  • name (Union[str, Sequence[str], None])

Return type:

pathlib.Path

plesty.lib.experiment.list_runs(run_root: str | pathlib.Path, name: str | Sequence[str] | None = None) → list[pathlib.Path]

Return the run directories under run_root, oldest first.

Parameters:

  • run_root (str | pathlib.Path) – The directory runs are created in.

  • name (Union[str, Sequence[str], None]) – Keep only runs of this experiment name (or any of several — an experiment that was renamed still owns its old runs). None keeps every run.

Return type:

list[pathlib.Path]

plesty.lib.experiment.parse_run_id(text: str) → tuple[str, datetime.datetime] | None

Split a run id into its experiment name and start time.

Parameters:

text (str) – A directory name.

Returns:

(name, started), or None when text is not a run id — the name may itself contain underscores, so the split is at the last one.

Return type:

Optional[tuple[str, datetime.datetime]]

plesty.lib.experiment.run_id(name: str, when: datetime.datetime | None = None) → str

Return the run id for experiment name started at when (now by default).

Parameters:

  • name (str)

  • when (Optional[datetime.datetime])

Return type:

str

class plesty.lib.experiment.Plan(steps: list[Step], config: dict[str, Any] | None = None)

A frozen, hashable schedule of atomic steps plus the run configuration.

Usage:

plan = Plan(
    steps=[Step(id=f"scan[{i}]", op="scan_point", params={"x": i}) for i in range(10)],
    config={"exposure_s": 0.1},
)
plan.save("runs/run_001/plan.json")

Initialize the plan with its steps and optional configuration.

Parameters:

  • steps (list[Step]) – Ordered atomic steps; step ids must be unique.

  • config (dict[str, Any] | None) – Experiment configuration recorded alongside the schedule.

Raises:

ValueError – If two steps share the same id.

steps: list[Step]
config: dict[str, Any]
to_dict() → dict[str, Any]

Return a JSON-serializable representation of the plan.

Return type:

dict[str, Any]

classmethod from_dict(data: dict[str, Any]) → Plan

Reconstruct a plan from to_dict() output.

Parameters:

data (dict[str, Any])

Return type:

Plan

content_hash() → str

Return a stable SHA-256 hash of the schedule and configuration.

Used on resume to verify that the persisted plan matches the plan the experiment would generate now — resuming under changed parameters is refused rather than silently mixing two schedules.

Return type:

str

save(path: str | pathlib.Path) → pathlib.Path

Write the plan as JSON to path, creating parent directories.

Parameters:

path (str | pathlib.Path)

Return type:

pathlib.Path

classmethod load(path: str | pathlib.Path) → Plan

Load a plan previously written by save().

Parameters:

path (str | pathlib.Path)

Return type:

Plan

__len__() → int

Return the number of steps in the plan.

Return type:

int

class plesty.lib.experiment.Step

One atomic measurement step in an experiment plan.

Variables:

  • id – Stable, deterministic identifier unique within the plan, e.g. "scan[x=3,y=5]". Resume matches completed steps by this id, so it must not depend on run time or randomness.

  • op – Name of the experiment method to call for this step.

  • params – Keyword arguments passed to the method.

id: str
op: str
params: dict[str, Any]
to_dict() → dict[str, Any]

Return a JSON-serializable representation of the step.

Return type:

dict[str, Any]

classmethod from_dict(data: dict[str, Any]) → Step

Reconstruct a step from to_dict() output.

Parameters:

data (dict[str, Any])

Return type:

Step