The Experiment

Lesson 4 of 6. A second repository, and a second project. The answer is on branch step-4-experiment of plesty-demo-pol-pl.

take_rows.py took four rows and printed them. That is a script, not an experiment: nothing was recorded, nothing could be resumed, and nothing could watch it.

An experiment fixes that. It reads a configuration, works out the full list of steps up front, runs them in order, and writes each result to disk the moment it has it. This lesson builds one.

What the framework gives you

You do not write any of that machinery. plesty.lib.experiment provides it, and an experiment is a subclass that fills in two things — what the steps are, and what one step does:

The measurement is a new project, separate from the bench: the bench is the instruments, and one bench runs many experiments over its life.

Scaffold it

Same tool as lesson 1, different template:

mkdir plesty-demo-pol-pl && cd plesty-demo-pol-pl
uv run plesty init experiment

Installed pre-push hook: uv run plesty check
Initialized experiment project at: …/plesty-demo-pol-pl
Python module name: demo_pol_pl
Import as: plesty.demo_pol_pl

The name is taken from the directory again, and again there is no git init to do — the scaffold makes the repository and its first commit.

What experiment gives you that default did not:

It also prints what to do next, which is the outline of this lesson: implement build_plan() and the step methods, then preview the plan without hardware.

Take it at its word before writing anything:

uv sync
uv run python -m plesty.demo_pol_pl --dry-run

Plan: 3 step(s), content hash 909903064651…
  measure[x=0]: measure_point({'x': 0})
  measure[x=1]: measure_point({'x': 1})
  measure[x=2]: measure_point({'x': 2})

An experiment that runs before you have written a line of it — three example steps calling an empty method. Nothing is connected, which is what --dry-run is for.

What is missing is the measurement. Build it from the bottom: one row first, then the schedule that repeats it.

Bring the rig across

The step needs the rig, so the rig comes with you. Copy the file you wrote in lesson 3 into this package:

cp ../plesty-demo-bench/plesty/demo_bench/rig.py plesty/demo_pol_pl/rig.py

It needs a newer plesty-lib than the scaffold pins:

dependencies = [
  "plesty-lib>=0.3.5.dev2,<1",
]

Then build_devices in main.py, which the scaffold wrote to build a one-client composite from a DEVICE_ADDRESS:

def build_devices() -> DemoRig:
    """Build the rig this experiment measures with.

    No address appears here. ``DemoRig`` reads ``HWP_ADDRESS``,
    ``SPEC_ADDRESS`` and ``PM_ADDRESS`` from the environment itself, so a
    deployment is configured by its ``.env`` and the experiment never knows a
    hostname.
    """
    return DemoRig()

The import at the top becomes from .rig import DemoRig, and build_client, CompositeDevice, EnvSettings and the --device-address argument all go — the rig reads its own addresses, so this module has no use for them.

Which leaves .env, the last file. Replace the scaffold's single DEVICE_ADDRESS:

HWP_ADDRESS=tcp://localhost:5652
SPEC_ADDRESS=tcp://localhost:5653
PM_ADDRESS=tcp://localhost:5651

Use your own ports. plesty-server status prints what the bench allocated. Lesson 3 showed what a wrong one looks like, and it does not look like a wrong port.

uv sync

One row

This is the atomic unit: everything the measurement does, once. In plesty/demo_pol_pl/experiment.py, measure_row replaces the scaffold's measure_point:

    def measure_row(self, angle_deg: float, exposure_s: float) -> dict[str, Any]:
        """Measure one row: turn the plate, expose, read the power.

        The returned dictionary is persisted as one line of ``records.jsonl``,
        which is what the monitors read. Everything needed to interpret the row
        is in the row — including the angle the stage reports having reached,
        which is not always the angle that was asked for.

        Args:
            angle_deg:  Half-wave-plate angle for this row.
            exposure_s: Spectrometer exposure time in seconds.

        Returns:
            The row document: angles, frame path, power, and the window the
            reading covered.
        """
        reported_deg = self.devices.set_hwp(angle_deg)
        t_start = _now()
        row = self.devices.acquire_row(exposure_s)
        return {
            "hwp_deg": angle_deg,
            "hwp_reported_deg": reported_deg,
            "exposure_s": exposure_s,
            "t_start": t_start,
            "t_end": _now(),
            **row,
        }

with a timestamp helper at module level, above the class:

def _now() -> str:
    """The current UTC instant, as the records store it."""
    return datetime.datetime.now(datetime.UTC).isoformat()

Two things are worth pausing on.

Returning a dictionary is the whole of persistence. There is no "write the record" call to make. A step that returns something non-None has it appended to records.jsonl along with its provenance — the step id, the operation, the parameters it was called with. A step that returns None is journalled as complete and stores nothing, which is what you want for a step that only moves something.

The record carries both angles. hwp_deg is what the plan asked for; hwp_reported_deg is what the stage says it reached. On a real bench those differ, and the second one is the one the measurement actually happened at. Recording only the first is the kind of loss you discover a year later.

Notice also what measure_row does not contain: no address, no port, no device command. It asks the rig for two things. Everything about how those reach three servers was settled in lesson 3.

Set up before the first row

The exposure and the correction wavelength are the same for every row, so they are pushed once per run rather than once per row. The scaffold left a setup hook for exactly this:

    async def setup(self) -> None:
        """Connect the rig, then push the acquisition settings to it.

        The settings are pushed once here rather than per row: they do not
        change during a sweep, and a write per row would be one round trip
        per row buying nothing.
        """
        await super().setup()
        self.devices.configure(
            exposure_s=float(self.config.get("exposure_s", 0.3)),
            wavelength_nm=float(self.config.get("wavelength_nm", 780.0)),
        )

super().setup() is what connects the rig, so the configure call after it has something to talk to. teardown already disconnects; it needs nothing added.

The plan

One row is settled. A plan is how many of them, and in what order: the whole list of steps, worked out before the first one runs. Each step has an id, an operation, and the parameters that operation is called with.

Deciding everything up front is what makes a run resumable. The plan is written to disk and content-hashed; a resumed run rebuilds it and compares, so a configuration edited halfway through is refused rather than quietly stitched into the middle of a measurement.

It also means build_plan() must be deterministic and device-free — no timestamps, no random ids, and nothing asked of an instrument.

The configuration

The plan is built by build_plan(), in code. config/default.yaml is where its parameters are set — the values you change between runs without touching the module. Replace the example points: 3:

# The half-wave plate sweep. 0-180 degrees in 2-degree steps is 91 rows,
# which is the grid the replayed dataset was measured on.
hwp:
  start_deg: 0.0
  stop_deg: 180.0
  step_deg: 2.0

# Spectrometer exposure per row, and the wavelength the powermeter corrects
# for. The emission this measures sits around 780 nm.
exposure_s: 0.3
wavelength_nm: 780.0

Tuning values go here, in a tracked file, because they are part of what a run was. Addresses and credentials never do — those are .env, which is not tracked.

Building the steps

Back in experiment.py, replace build_plan:

    def build_plan(self) -> Plan:
        """Build the deterministic measurement schedule from the configuration.

        One step per half-wave-plate angle. Step ids must be stable across
        calls — no timestamps or randomness — and params must stay
        JSON-serializable, because the plan is written to disk and compared
        against on resume.
        """
        hwp = self.config.get("hwp", {})
        exposure_s = float(self.config.get("exposure_s", 0.3))
        steps = [
            Step(
                id=f"hwp[{angle:06.2f}]",
                op="measure_row",
                params={"angle_deg": angle, "exposure_s": exposure_s},
            )
            for angle in _angles(
                float(hwp.get("start_deg", 0.0)),
                float(hwp.get("stop_deg", 180.0)),
                float(hwp.get("step_deg", 2.0)),
            )
        ]
        return Plan(steps, config=self.config)

op="measure_row" is the method you just wrote — a step says what to call and with which arguments, nothing more. One of the contract tests checks that every op resolves to a real method, so a typo fails at plesty check rather than at row 47 of a measurement.

The angles come from a second module-level helper, beside _now:

def _angles(start_deg: float, stop_deg: float, step_deg: float) -> list[float]:
    """The sweep angles, endpoint included.

    Counted rather than accumulated: adding ``step_deg`` 90 times drifts, and
    a step id built from a drifted angle would not match the one the plan was
    resumed from.
    """
    count = int(round((stop_deg - start_deg) / step_deg))
    return [round(start_deg + index * step_deg, 6) for index in range(count + 1)]

The docstring is the whole reason it is not a while angle <= stop. Step ids are compared as text on resume, and hwp[180.00] built from an angle that drifted to 179.99999999 is a different step.

Preview it — still with nothing connected:

uv run python -m plesty.demo_pol_pl --dry-run

Plan: 91 step(s), content hash bcc1b48f020f…
  hwp[000.00]: measure_row({'angle_deg': 0.0, 'exposure_s': 0.3})
  hwp[002.00]: measure_row({'angle_deg': 2.0, 'exposure_s': 0.3})
  …
  hwp[180.00]: measure_row({'angle_deg': 180.0, 'exposure_s': 0.3})

91 steps, endpoints included, and no instrument was touched to produce them. This is the cheapest way to check a configuration before committing an instrument to it.

Name the experiment

Runs are named after the experiment, and the base class falls back to the class name — which here is Experiment, so every run would be called experiment_20260821-191701. Set it once, in init:

        # Without a name the base takes the class name, and every run would be
        # called `experiment_<timestamp>`. The name is what a run directory is
        # called and what the viewer filters on, so it says which measurement.
        kwargs.setdefault("name", "demo_pol_pl")
        super().__init__(devices=devices, run_root=run_root, **kwargs)

Lesson 6 filters runs by that name. It is easier to set now than to rename a directory of finished measurements later.

Run it

Bring the bench up if it is down, then:

uv run python -m plesty.demo_pol_pl

… | plesty.lib.device.composite_device | DemoRig: connecting sub-device(s) hwp, spec, pm
… | plesty.lib.device.composite_device | hwp: - -> connecting (tcp://localhost:5652, timeout 5000 ms, deadline 10 s)
… | demo_pol_pl | Run demo_pol_pl_20260821-191701: 91 step(s), run dir runs/demo_pol_pl_20260821-191701
… | demo_pol_pl | Setup: connecting devices
… | demo_pol_pl | Step 1/91 hwp[000.00] (measure_row)
… | demo_pol_pl | Step 1/91 done in 0.5 s
…
… | demo_pol_pl | Teardown: disconnecting devices
… | demo_pol_pl | Run demo_pol_pl_20260821-191701 completed (91 steps) in 42.3 s.
Run completed: demo_pol_pl_20260821-191701

About forty seconds. The third line is the one to keep: it names the run and the directory everything lands in.

What a run leaves behind

runs/demo_pol_pl_20260821-191701/
  plan.json        the frozen schedule and the configuration it came from
  journal.jsonl    what happened, in order: started, each step, finished
  records.jsonl    one line per step that returned something

Three files, all append-only, all plain text. The records are what the next two lessons read:

{"index": 0, "provenance": {"step_id": "hwp[000.00]", "op": "measure_row",
                            "params": {"angle_deg": 0.0, "exposure_s": 0.3}},
 "value": {"hwp_deg": 0.0, "hwp_reported_deg": 0.0, "exposure_s": 0.3,
           "t_start": "…", "t_end": "…",
           "data_file": "…/frames/…/mock_20260821-191701-731414.spe",
           "power_w": 6.112988e-06}}

Everything needed to interpret that row is in the row, and the frame itself is not: only its path travels, because the frame stays on the disk the spectrometer wrote it to.

Check a few rows against the measurement being replayed:

uv run python -c "
import json
rows = [json.loads(l) for l in open('runs/<run_id>/records.jsonl')]
for i in (0, 6, 30, 90):
    v = rows[i]['value']
    print(f\"{v['hwp_deg']:6.1f}  {v['power_w']*1e6:8.3f} uW\")
"

   0.0     6.113 uW
  12.0     0.263 uW
  60.0    24.149 uW
 180.0     5.784 uW

Those are the powers measured on 4 August 2026, to the microwatt. You have re-run a real measurement.

Fix the tests

uv run pytest

3 failed, 6 passed

Not a problem — the arithmetic of having changed the measurement. The six that pass are the contract tests, which check the shape of an experiment and keep passing throughout. The three that fail are the scaffold's own, written against measure_point and the old build_devices.

Replacing them is mostly writing a fake rig, and the interesting part is how little it needs:

class FakeRig:
    """Duck-typed stand-in for :class:`~plesty.demo_pol_pl.rig.DemoRig`."""

    def connect_all(self) -> None: ...
    def disconnect_all(self) -> None: ...
    def configure(self, exposure_s: float, wavelength_nm: float) -> None: ...
    def set_hwp(self, angle_deg: float) -> float: ...
    def acquire_row(self, exposure_s: float) -> dict[str, object]: ...

Five methods, no server, no ZMQ, no dataset. The experiment was written against the rig's interface rather than against instruments, so the whole run — plan, journal, records, resume — is testable in a quarter of a second. The full version is on the answer branch.

Then the gates:

uv run plesty check

Checking against standard: pixel (no release tag yet — prototype default)
  ✓ Metadata & Namespace
  ✓ Code Hygiene (lint)
  ✓ Code Hygiene (format)
  ✓ Code Hygiene (types)

All checks passed.

Resume

Stop a run with Ctrl-C partway through, and it tells you how to pick it up:

Run interrupted — resume with --resume <run_id>.

uv run python -m plesty.demo_pol_pl --resume <run_id>

… | demo_pol_pl | Resuming run demo_pol_pl_20260821-191025: 25/91 steps already completed.
… | demo_pol_pl | Run demo_pol_pl_20260821-191025 completed (91 steps) in 30.7 s.

The journal is what makes that work: steps recorded as complete are skipped, and the plan is rebuilt and compared against the stored one first. Edit config/default.yaml and try to resume, and it refuses — a run is one measurement or it is nothing.

On a real bench this is the difference between losing an evening to a stalled instrument and losing four minutes.

What you left out

This experiment is deliberately smaller than plesty-pol-pl, the production module for this measurement. What that one adds, and this one does not:

None of that changes the shape you have built. It is the same build_plan, the same step methods returning documents, the same journal.

Next

You have a measurement that records what it did. The next two lessons are about watching it: the monitor contract, and the viewer that docks three of them into one window.