Measurement Workflows

Write measurement workflows that orchestrate devices, acquire data, and persist results using save_result.

A measurement workflow is the core of an experiment. It orchestrates devices, acquires data, and persists results. PLESTY experiments follow a structured pattern to ensure reproducibility and error handling.

Workflow structure

A typical measurement workflow follows these steps:

  1. Initialize — connect to devices and set initial parameters
  2. Configure — set device parameters for the measurement
  3. Acquire — read data from devices
  4. Process — optionally run analyzers on the data
  5. Save — persist results
  6. Cleanup — disconnect devices

Example: Power sweep

class PowerSweep:
    """Measure power output across a range of laser drive currents."""

    def __init__(self, composite: CompositeDevice) -> None:
        self.composite = composite
        self.results = []

    def run(self, current_range: list[float]) -> dict:
        """Run a power sweep measurement."""
        for current in current_range:
            # Configure
            self.composite.call("laser", "_write_", key="current", value=current)

            # Acquire
            power = self.composite.call("meter", "_query_", key="power")

            # Save
            self.save_result({
                "current": current,
                "power": power,
            })

        return {"measurements": self.results}

    def save_result(self, data: dict) -> None:
        """Persist a measurement result."""
        self.results.append(data)

Result persistence

Use save_result() to persist measurement data. The SDK provides this method for experiment classes. Never use ad-hoc persistence — always use save_result (this is enforced by gate e2).

def save_result(self, data: dict) -> None:
    """Persist a measurement result.

    Args:
        data: Dictionary of measurement data to save.
    """
    ...

Error handling

Wrap device calls in try-except blocks to handle communication failures:

def run(self) -> dict:
    try:
        self.composite.call("laser", "connect")
        # ... measurement logic ...
    except ConnectionError:
        self.composite.reconnect("laser")
        # Retry logic
    finally:
        self.composite.call("laser", "disconnect")

Next steps