plesty.lib.experiment ===================== .. py:module:: plesty.lib.experiment .. autoapi-nested-parse:: Experiment orchestration: plans, journals, and the async Experiment ABC. Submodules ---------- .. toctree:: :maxdepth: 1 /reference/plesty/lib/experiment/base_experiment/index /reference/plesty/lib/experiment/journal/index /reference/plesty/lib/experiment/preflight/index /reference/plesty/lib/experiment/runs/index /reference/plesty/lib/experiment/schedule/index Exceptions ---------- .. autoapisummary:: plesty.lib.experiment.InvalidOperationError plesty.lib.experiment.PlanMismatchError plesty.lib.experiment.RigNotReadyError Classes ------- .. autoapisummary:: plesty.lib.experiment.Experiment plesty.lib.experiment.Journal plesty.lib.experiment.Run plesty.lib.experiment.Plan plesty.lib.experiment.Step Functions --------- .. autoapisummary:: plesty.lib.experiment.connect_rig plesty.lib.experiment.latest_run plesty.lib.experiment.list_runs plesty.lib.experiment.parse_run_id plesty.lib.experiment.run_id Package Contents ---------------- .. py:class:: Experiment(devices: Optional[plesty.lib.device.composite_device.CompositeDevice] = None, run_root: str | pathlib.Path | None = None, name: Optional[str] = None, max_retries: int = 3, retry_sleep: float = 3.0, data_dir: Optional[str] = None, data_mount: Optional[str] = None) Bases: :py:obj:`abc.ABC` Base class for PLESTY experiments with atomic scheduling and resume. Subclasses override exactly three lifecycle hooks: * :meth:`build_plan` — **required**: the atomic measurement schedule. * :meth:`setup` / :meth:`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 :class:`~plesty.lib.device.composite_device.CompositeDevice` passed at construction — an Experiment receives Device instances, never raw hardware. Initialize the experiment. :param devices: Composite device holding all instruments this experiment orchestrates; connected in :meth:`setup` and disconnected in :meth:`teardown`. :param run_root: 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``. :param name: Experiment name used in run ids; defaults to the subclass name in lowercase. :param max_retries: Attempts per step before the run aborts. :param retry_sleep: Seconds to wait between step retries. :param data_dir: The shared data root as the acquiring host sees it; ``None`` takes ``PLESTY_DATA_DIR``. Device blobs go to :attr:`raw_dir` beneath it. :param data_mount: The same root as this machine sees it; ``None`` takes ``PLESTY_DATA_MOUNT``. Journaled for readers. .. py:attribute:: devices :value: None .. py:attribute:: data_dir .. py:attribute:: data_mount .. py:attribute:: run_root .. py:attribute:: name :value: '' .. py:attribute:: max_retries :value: 3 .. py:attribute:: retry_sleep :value: 3.0 .. py:attribute:: run_id :type: Optional[str] :value: None .. py:attribute:: run_dir :type: Optional[pathlib.Path] :value: None .. py:attribute:: logger .. py:attribute:: _plan_length :value: 0 .. py:property:: raw_dir :type: Optional[str] Where this run's device-written blobs go, in the host's spelling. ``//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. .. py:method:: build_plan() -> plesty.lib.experiment.schedule.Plan :abstractmethod: 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. .. py:method:: setup() -> None :async: 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 :meth:`CompositeDevice.preflight`). Nothing is measured on a rig that is not what the plan assumes. .. py:method:: teardown() -> None :async: Release resources; the default disconnects all devices. Always awaited by :meth:`run`, whether the run completes, fails, or is canceled. .. py:method:: run(resume: Optional[str] = None) -> str :async: Execute the plan, journaling every step; optionally resume a run. :param resume: 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). :raises PlanMismatchError: If resuming and the stored plan differs from the plan the experiment generates now. :raises Exception: The final step error is re-raised after ``run_aborted`` is journaled and :meth:`teardown` has completed. .. py:method:: _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. .. py:method:: _prepare_run(plan: plesty.lib.experiment.schedule.Plan, resume: Optional[str]) -> 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. .. py:method:: _run_step(step: plesty.lib.experiment.schedule.Step, index: int, run_dir: pathlib.Path, journal: plesty.lib.experiment.journal.Journal) -> None :async: Execute one atomic step with retries and persist its result. .. py:exception:: InvalidOperationError Bases: :py:obj:`RuntimeError` Raised when a plan step references an invalid experiment operation. Initialize self. See help(type(self)) for accurate signature. .. py:exception:: PlanMismatchError Bases: :py:obj:`RuntimeError` Raised when resuming a run whose stored plan differs from the current one. Initialize self. See help(type(self)) for accurate signature. .. py:class:: Journal(path: str | pathlib.Path) Crash-safe append-only event log for one experiment run. Attach the journal to *path*, creating parent directories. :param path: Location of the ``journal.jsonl`` file; appended to if it already exists (the resume case). .. py:attribute:: TERMINAL_EVENTS :value: ('run_completed', 'run_canceled', 'run_aborted') .. py:attribute:: path .. py:method:: append(event: str, **fields: Any) -> dict[str, Any] Append one event line and fsync it to disk. :param event: Event name, e.g. ``"step_completed"``. :param fields: Additional payload stored on the event (step id, result path, error text, ...). :returns: The full event record that was written. .. py:method:: 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. .. py:method:: 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. .. py:method:: 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. .. py:exception:: RigNotReadyError(problems: dict[str, str], message: str) Bases: :py:obj:`RuntimeError` The rig failed its preflight and the operator did not (or could not) recover it. :ivar problems: Sub-device name → one-line problem. Keep the per-device problems next to the message. .. py:attribute:: problems .. py:function:: connect_rig(build: collections.abc.Callable[[frozenset[str]], plesty.lib.device.composite_device.CompositeDevice], *, optional: collections.abc.Iterable[str] = (), ask: Optional[collections.abc.Callable[[str], str]] = None, env_prefix: Optional[str] = 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. :param build: 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. :param optional: Sub-devices the experiment can run without; when only these fail, the operator may choose to continue without them. :param ask: 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 :class:`RigNotReadyError` raised). :param env_prefix: Deployment prefix of the ``_ADDRESS`` variables, for the diagnosis. :param out: 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. .. py:class:: Run(run_dir: str | pathlib.Path, *, mount: Optional[str] = None, data_dir: Optional[str] = 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 :meth:`status` because a run under way keeps appending to it. Attach to a run directory. :param run_dir: The run directory. :param mount: 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). :param data_dir: The shared root as the acquiring host sees it — only for runs that did not record it themselves. :param replay_rows: How :meth:`source` hands records out — ``0`` follows the run live, ``n`` rations a stored run *n* records per poll. .. py:attribute:: path .. py:attribute:: replay_rows :value: 0 .. py:attribute:: _mount :value: None .. py:attribute:: _data_dir :value: None .. py:attribute:: _plan :type: Optional[plesty.lib.experiment.schedule.Plan] :value: None .. py:attribute:: _started :type: Optional[dict[str, Any]] :value: None .. py:attribute:: sources :type: list[Any] :value: [] .. py:property:: id :type: str The run id (the directory name). .. py:property:: name :type: Optional[str] The experiment name the run id carries, if it parses as one. .. py:property:: started :type: Optional[datetime.datetime] The start time the run id carries, if it parses as one. .. py:property:: plan :type: plesty.lib.experiment.schedule.Plan The frozen plan (loaded once). .. py:property:: config :type: dict[str, Any] The frozen configuration; ``{}`` when the plan is unreadable. .. py:property:: journal :type: plesty.lib.experiment.journal.Journal The run's journal. .. py:method:: status() -> dict[str, Any] The run status by journal replay (see :meth:`Journal.status`). .. py:method:: records() -> Iterable[dict[str, Any]] Iterate the committed step records, oldest first. .. py:method:: _run_started() -> dict[str, Any] .. py:property:: data_dir :type: Optional[str] 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. .. py:property:: raw_dir :type: Optional[str] Where the run's device-written blobs went, in the host's spelling. .. py:property:: data_mount :type: Optional[str] The shared root as this machine sees it (see the class docstring). .. py:method:: 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. .. py:method:: source(mapper: Optional[plesty.lib.monitor.sources.FrameMapper] = None) -> Union[plesty.lib.monitor.sources.RunSource, plesty.lib.monitor.sources.ReplaySource] Return a source over this run's records — live or replayed. A :class:`~plesty.lib.monitor.sources.RunSource` when :attr:`replay_rows` is 0, else a :class:`~plesty.lib.monitor.sources.ReplaySource` handing out that many records per poll. Every source made here is kept in :attr:`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. .. py:property:: exhausted :type: 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. .. py:method:: __repr__() -> str Return ``Run('')``. .. py:function:: latest_run(run_root: str | pathlib.Path, name: Union[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. .. py:function:: list_runs(run_root: str | pathlib.Path, name: Union[str, Sequence[str], None] = None) -> list[pathlib.Path] Return the run directories under *run_root*, oldest first. :param run_root: The directory runs are created in. :param name: 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. .. py:function:: parse_run_id(text: str) -> Optional[tuple[str, datetime.datetime]] Split a run id into its experiment name and start time. :param text: 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. .. py:function:: run_id(name: str, when: Optional[datetime.datetime] = None) -> str Return the run id for experiment *name* started at *when* (now by default). .. py:class:: Plan(steps: list[Step], config: dict[str, Any] | None = None) A frozen, hashable schedule of atomic steps plus the run configuration. Usage: .. code-block:: python 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. :param steps: Ordered atomic steps; step ids must be unique. :param config: Experiment configuration recorded alongside the schedule. :raises ValueError: If two steps share the same id. .. py:attribute:: steps :type: list[Step] .. py:attribute:: config :type: dict[str, Any] .. py:method:: to_dict() -> dict[str, Any] Return a JSON-serializable representation of the plan. .. py:method:: from_dict(data: dict[str, Any]) -> Plan :classmethod: Reconstruct a plan from :meth:`to_dict` output. .. py:method:: 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. .. py:method:: save(path: str | pathlib.Path) -> pathlib.Path Write the plan as JSON to *path*, creating parent directories. .. py:method:: load(path: str | pathlib.Path) -> Plan :classmethod: Load a plan previously written by :meth:`save`. .. py:method:: __len__() -> int Return the number of steps in the plan. .. py:class:: Step One atomic measurement step in an experiment plan. :ivar 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. :ivar op: Name of the experiment method to call for this step. :ivar params: Keyword arguments passed to the method. .. py:attribute:: id :type: str .. py:attribute:: op :type: str .. py:attribute:: params :type: dict[str, Any] .. py:method:: to_dict() -> dict[str, Any] Return a JSON-serializable representation of the step. .. py:method:: from_dict(data: dict[str, Any]) -> Step :classmethod: Reconstruct a step from :meth:`to_dict` output.