Composite Device

Lesson 3 of 6. The answer is on branch step-3-composite.

Three servers are running, and lesson 2 drove two of them by hand. That works right up until the measurement needs two instruments to act at the same moment — and then every script has to solve the same problems: which thread owns which socket, what happens when one server stalls, how an exposure and a power reading are made to cover one window.

CompositeDevice solves them once. You write a rig — the lab word for a set of instruments assembled to do one job — as a subclass of it. The rig offers the measurement a short list of things it can ask for ("turn the filter to 40°", "take a reading"), and the measurement never talks to an individual instrument again.

It goes in the package beside the two scripts you have already written, plesty/demo_bench/rig.py, and is imported as from plesty.demo_bench import DemoRig — under the shared plesty namespace, like every other module on the platform.

What CompositeDevice gives you

Before writing anything, the handful of methods a rig is built out of. A subclass inherits all of them; the ones this lesson uses are the first three.

call and submit are the same operation with one difference: call waits, submit does not. Everything a rig adds is written in terms of those two.

The full surface is in the plesty-lib documentation.

Check what the scaffold pinned

Two of the keys this lesson uses — requires and connect_deadline_s — are recent, and a plesty-lib older than they are rejects them by name:

ValueError: Unknown configuration key(s) ['connect_deadline_s', 'requires'] for sub-device 'hwp';
accepted keys are ['address', 'attempts', 'backoff_s', 'env', 'timeout_ms'].

That is a good error — the library refuses a key it would otherwise ignore, so a preflight you think you configured cannot silently not happen. You should not see it: plesty init pins the newest plesty-lib published on the day it scaffolds, so lesson 1 already asked for a library that has both. If you do see it, the project was scaffolded by an older SDK; raise the floor in pyproject.toml to the version the error is missing, and uv sync.

Write the rig

Create plesty/demo_bench/rig.py. It starts as a subclass of CompositeDevice with nothing in it, and the rest of this lesson fills it in:

"""Three device servers as one instrument."""

from __future__ import annotations

import time
from typing import Any

from plesty.lib.device.composite_device import CompositeDevice


class DemoRig(CompositeDevice):
    """The half-wave plate, the spectrometer and the powermeter, together."""

Export it from the package so the scripts can reach it, in plesty/demo_bench/__init__.py:

from .rig import DemoRig

__all__ = ["DemoRig", "load_config"]

What the rig is made of

The sub-devices are declared as data, not code — a module-level DEVICES above the class:

DEVICES: dict[str, Any] = {
    "hwp": {
        "timeout_ms": 5000,
        "connect_deadline_s": 10,
        "requires": ["home_stage", "move_absolute", "get_position"],
    },
    "spec": {
        "timeout_ms": 5000,
        "connect_deadline_s": 10,
        "requires": ["acquire", "get_recent_file", "write", "query"],
    },
    "pm": {
        "timeout_ms": 5000,
        "connect_deadline_s": 10,
        "requires": ["measure_power", "write"],
    },
}

No address appears. Each is read from <NAME>_ADDRESS in the environment — HWP_ADDRESS, SPEC_ADDRESS, PM_ADDRESS. Write them in a file called .env at the top of the project:

HWP_ADDRESS=tcp://localhost:5552
SPEC_ADDRESS=tcp://localhost:5553
PM_ADDRESS=tcp://localhost:5551

Use your own ports. plesty-server status prints what the bench allocated; declared in lesson 2's order they come out 5551, 5552, 5553, but any of them may have been skipped. This is the third file whose addresses have to match the bench, and it is the one that fails least helpfully — a wrong address connects to something, and you find out at the first row.

The scaffold's .gitignore already excludes .env, which is the platform's convention for anything environment-shaped: on a real bench that file holds the addresses of instruments on a lab network, and sometimes a credential. A module meant for others ships a tracked .env.example beside it as the template.

A deployment is configured by its .env; the module never knows a hostname.

What requires is for

A wrong address still connects. Something is listening on that port, the socket opens, and nothing complains — you find out at the first row, when the stage is asked to move and answers that it has no such operation.

requires is how the rig checks that the thing at the other end is the instrument it wanted. Call preflight() and, for each sub-device, it asks the server describe(), compares the answer with the list, and confirms identity() responds:

rig = DemoRig()
print(rig.preflight())   # {} when every server is what it claims

Swap two addresses in .env — the stage's port under PM_ADDRESS, say — and it says so before anything moves:

hwp: server at tcp://localhost:5561 is not the expected device:
     missing home_stage, move_absolute, get_position
pm:  server at tcp://localhost:5562 is not the expected device: missing measure_power

It returns a dictionary of name → problem rather than raising, so the caller decides what to do: the experiment in lesson 4 refuses to start and shows them. Nothing calls it for you — take_rows.py below does not, which is why getting .env wrong there fails at the first row instead.

Worth knowing what that looks like, because it does not look like an address problem. configure writes a spectrometer setting, so if SPEC_ADDRESS is really the powermeter's port, the powermeter is asked for a camera setting it has never heard of:

RuntimeError: KeyError: "Unknown group prefix 'CameraSettings' in
key 'CameraSettings.HardwareIOTriggerResponse'."

Nothing in that names a port, and it reads like a broken schema. It means the connection went somewhere real and wrong. plesty-server status prints what each port actually is; .env has to agree with it. preflight() would have said so in one line — that is the argument for calling it.

That is the whole trick: an address says where to connect, requires says whether you connected to the right thing.

One thread per instrument

The constructor is one line of work, and one line saying what it is:

    def __init__(self) -> None:
        """Build the rig: the three servers, each pinned to its own thread."""
        super().__init__(DEVICES, thread_affinity=True)

The docstring is not decoration. This file travels into lesson 4, which is where the compliance gates run for the first time, and the lint gate wants one on every public method — including __init__.

thread_affinity=True gives each sub-device its own single-worker thread — dev-hwp, dev-spec, dev-pm — and pins that sub-device's connection to it: opened there, and only ever used there. It does two jobs, and neither is speed.

It keeps each socket on one thread. The connections underneath are ZMQ sockets, which are not thread-safe; even moving one between threads is fragile. Pinning makes the mistake impossible rather than unlikely.

It is what makes submit concurrent. With the flag off, a submitted call runs inline and comes back already finished — the same code stays correct, just sequential. That is the quiet one: acquire_row below would block on the exposure before ever reaching the power reading, so the power would describe a moment the spectrum was not taken in, and nothing would look wrong.

Every call is bounded

The budget for that retrying is declared per sub-device. The timeout_ms: 5000 in DEVICES fails fast everywhere, because these servers heal themselves and a quick retry beats a long wait.

Operations that are genuinely slow raise their own window per call, which is what the constant at the top of the file is for:

MOVE_TIMEOUT_S = 130.0   # a long rotation is mechanical; 5 s is not enough

A routine move takes about a second and a long one over a minute. Both go through the same code path, with different budgets.

Turning the plate, and setting up

Two of the three methods are the plain kind — ask one sub-device to do one thing:

    def set_hwp(self, angle_deg: float) -> float:
        """Move the plate to an absolute angle; return where it stopped."""
        future = self.submit("hwp", "move_absolute", angle_deg, timeout=MOVE_TIMEOUT_S)
        return float(future.result())

    def configure(self, exposure_s: float, wavelength_nm: float) -> None:
        """Push the acquisition settings to the spectrometer and the meter."""
        # Leftover state — an external trigger, a shutter held closed for
        # darks — otherwise arms an acquisition that never completes, so both
        # are pinned rather than assumed.
        for key, value in (
            ("CameraSettings.HardwareIOTriggerResponse", "NoResponse"),
            ("CameraSettings.ShutterTimingMode", "Normal"),
            ("CameraSettings.ShutterTimingExposureTime", exposure_s * 1000.0),
        ):
            self.submit("spec", "write", key, value).result()
        self.submit("pm", "write", "wavelength", float(wavelength_nm)).result()

set_hwp returns what the stage reports, not what it was told: that is the angle the row was measured at. Those three CameraSettings keys are the spectrometer's own vocabulary — the sort of thing a device module exists to hide, and the sort of thing you look up in its docs once and never again.

The one that has to overlap

This is what a composite buys you that three clients do not. Each row of the sweep needs the powermeter to read during the spectrometer's exposure, not before or after it — otherwise the power reference does not describe the frame it is attached to.

That needs one more constant beside MOVE_TIMEOUT_S, and this one is about the instrument rather than the network:

#: The spectrometer needs a moment of setup before its shutter actually opens,
#: so a power sample taken immediately would land before the exposure began.
PM_DELAY_S = 0.4

Then the method:

    def acquire_row(self, exposure_s: float) -> dict[str, Any]:
        """Expose the spectrometer and read the power during that exposure."""
        spec_future = self.submit("spec", "acquire", timeout=exposure_s + 10.0)

        # Waited for here rather than on the powermeter's own thread, which
        # the reading itself needs; the exposure is already running.
        time.sleep(PM_DELAY_S)
        power_w = float(self.submit("pm", "measure_power", averaging=10).result()["power"])

        spec_future.result()
        # The frame stays on the acquiring host's disk; only its path travels.
        data_file = self.submit("spec", "get_recent_file").result()
        if not data_file:
            raise RuntimeError("The spectrometer reported no saved frame.")
        return {"data_file": str(data_file), "power_w": power_w}

Both calls are started before either is waited on, which is the whole trick: the exposure is running while the power is read. Wait on the first .result() before submitting the second and the reading lands after the shutter has closed, describing a moment the spectrum was not taken in.

The sleep waits on the rig's own thread rather than the powermeter's, which the reading itself is about to need — the exposure is already running on another, so nothing is lost by pausing here.

The rig's own vocabulary

That is the file. What a measurement sees of it is three methods:

Write plesty/demo_bench/take_rows.py against them:

"""Take a few polarization rows through the rig."""

from plesty.demo_bench import DemoRig

#: Angles to measure, in degrees of half-wave plate.
ANGLES = [0.0, 20.0, 40.0, 60.0]

#: Exposure per row, and the wavelength the powermeter corrects for.
EXPOSURE_S = 0.3
WAVELENGTH_NM = 780.0


def main() -> None:
    """Measure one row at each angle and print what came back."""
    rig = DemoRig()
    try:
        rig.configure(exposure_s=EXPOSURE_S, wavelength_nm=WAVELENGTH_NM)
        print(f"{'angle (deg)':>12}  {'power (uW)':>11}  frame")
        for angle in ANGLES:
            reached = rig.set_hwp(angle)
            row = rig.acquire_row(exposure_s=EXPOSURE_S)
            frame = row["data_file"].rsplit("/", 1)[-1]
            print(f"{reached:12.2f}  {row['power_w'] * 1e6:11.3f}  {frame}")
    finally:
        rig.disconnect_all()


if __name__ == "__main__":
    main()

Compare it with lesson 2's sweep_power.py: no addresses, no clients, no ports. It opens a rig and asks for rows. Run it:

uv run python -m plesty.demo_bench.take_rows

 angle (deg)   power (uW)  frame
        0.00        6.113  mock_20260820-133513-679173.spe
       20.00        0.611  mock_20260820-133514-143603.spe
       40.00       13.591  mock_20260820-133514-615042.spe
       60.00       24.149  mock_20260820-133515-080956.spe

The production module for this measurement, plesty-pol-pl, has a rig that does the same and more — it re-issues a move that settled off target, so a row is never recorded at the wrong angle. You meet it in the next lesson.

Images never cross the network

A frame is one readout of the spectrometer's detector — the raw image behind a spectrum. acquire_row returns a path to it, not the image itself:

{"data_file": ".../raw/mock_20260820-120807-809632.spe",
 "power_w": 6.112988e-06,
 "pm_averaging": 10}

The spectrometer writes the file and reports only where it put it. That is a platform-wide rule, and it is why a sweep's network traffic is a few hundred bytes per row whether the detector has 1340 pixels or four million. On a real bench the file lands on a disk shared between the machines, so whoever needs the image reads it directly rather than having it posted to them.

Look at what it wrote:

ls frames/

The rig is assembled. Now run the measurement.