Experiments & Database

Experiments & Sweeps

This section documents the core classes and utilities for managing experiments, sweeps, and measurement results in PyTestLab.


Database & Results

pytestlab.experiments.database.Database = MeasurementDatabase module-attribute

pytestlab.experiments.results.MeasurementResult(values, instrument, units, measurement_type, timestamp=None, envelope=None, sampling_rate=None, **kwargs)

MeasurementResult(
    values: float,
    instrument: str,
    units: str,
    measurement_type: str,
    timestamp: float | None = ...,
    envelope: dict[str, Any] | None = ...,
    sampling_rate: float | None = ...,
    **kwargs: Any,
)
MeasurementResult(
    values: np.ndarray
    | pl.DataFrame
    | np.float64
    | list[Any]
    | UFloat,
    instrument: str,
    units: str,
    measurement_type: str,
    timestamp: float | None = ...,
    envelope: dict[str, Any] | None = ...,
    sampling_rate: float | None = ...,
    **kwargs: Any,
)

A class to represent a collection of measurement values.

ATTRIBUTE DESCRIPTION
values

The measurement data.

TYPE: Union[ndarray, DataFrame, float64, TypingList[Any], UFloat]

units

The units of the measurements.

TYPE: str

instrument

The name of the instrument used for the measurements.

TYPE: str

measurement_type

The type of measurement.

TYPE: str

timestamp

Timestamp of when the result was created.

TYPE: float

Source code in pytestlab/experiments/results.py
def __init__(
    self,
    values: np.ndarray | pl.DataFrame | np.float64 | list[Any] | UFloat | float,
    instrument: str,
    units: str,
    measurement_type: str,
    timestamp: float | None = None,  # Allow optional timestamp override
    envelope: dict[str, Any] | None = None,  # Add envelope as an explicit argument
    sampling_rate: float | None = None,  # Add sampling_rate for FFT
    **kwargs: Any,
) -> None:  # Added **kwargs and type hint
    if isinstance(values, float) and not isinstance(values, np.floating):
        # Normalize plain Python float to numpy float64 for internal consistency
        values = np.float64(values)
    self.values: np.ndarray | pl.DataFrame | np.float64 | list[Any] | UFloat = cast(
        np.ndarray | pl.DataFrame | np.float64 | list[Any] | UFloat, values
    )
    self.units: str = units
    self.instrument: str = instrument
    self.measurement_type: str = measurement_type
    self.timestamp: float = timestamp if timestamp is not None else time.time()
    # Envelope logic: always provide an envelope attribute
    if envelope is not None:
        self.envelope = envelope
    else:
        # Default: minimal valid envelope (empty dict, or customize as needed)
        self.envelope = {}

    # Store sampling rate for FFT calculations
    self.sampling_rate = sampling_rate

    # Store any additional kwargs as attributes
    for key, value in kwargs.items():
        setattr(self, key, value)

Attributes

envelope = envelope instance-attribute

instrument = instrument instance-attribute

measurement_type = measurement_type instance-attribute

nominal property

sampling_rate = sampling_rate instance-attribute

sigma property

timestamp = timestamp if timestamp is not None else time.time() instance-attribute

units = units instance-attribute

values = cast(np.ndarray | pl.DataFrame | np.float64 | list[Any] | UFloat, values) instance-attribute

Functions

__delitem__(index)

Allows deleting an item from 'values' if it's a list or ndarray.

Source code in pytestlab/experiments/results.py
def __delitem__(self, index: int) -> None:
    """Allows deleting an item from 'values' if it's a list or ndarray."""
    if isinstance(self.values, list):
        del self.values[index]
    elif isinstance(self.values, np.ndarray):
        self.values = np.delete(self.values, index, axis=0)
    else:
        raise TypeError(f"Deletion by index not supported for type {type(self.values)}")

__getitem__(key)

Allow dictionary-style access or integer indexing into values.

Source code in pytestlab/experiments/results.py
def __getitem__(self, key):
    """Allow dictionary-style access or integer indexing into values."""
    if isinstance(key, int):
        if isinstance(self.values, np.ndarray | list):
            return self.values[key]
        if isinstance(self.values, pl.DataFrame):
            return self.values[key]
        if isinstance(self.values, (np.float64 | UFloat)) and key == 0:
            return self.values
        raise IndexError(
            f"Index {key} out of range or type {type(self.values)} not directly indexable."
        )
    return self.to_dict()[key]

__iter__()

Allows iteration over the 'values' attribute.

Source code in pytestlab/experiments/results.py
def __iter__(self) -> Iterator[Any]:
    """Allows iteration over the 'values' attribute."""
    if isinstance(self.values, np.ndarray | list):
        return iter(self.values)
    elif isinstance(self.values, pl.DataFrame):
        return iter(self.values.iter_rows())
    elif isinstance(self.values, np.float64 | UFloat):
        return iter([self.values])
    raise TypeError(f"Iteration not supported for type {type(self.values)}")

__len__()

Source code in pytestlab/experiments/results.py
def __len__(self) -> int:
    if isinstance(self.values, np.ndarray | list):
        return len(self.values)
    elif isinstance(self.values, np.float64 | UFloat):
        return 1
    elif isinstance(self.values, pl.DataFrame):
        return self.values.height  # Number of rows
    return 0  # Default for unknown types

__repr__()

Detailed representation of the measurement result.

For backward compatibility with tests, this matches str behavior. In typical libraries, repr would show construction details instead.

Source code in pytestlab/experiments/results.py
def __repr__(self) -> str:
    """Detailed representation of the measurement result.

    For backward compatibility with tests, this matches __str__ behavior.
    In typical libraries, repr would show construction details instead.
    """
    return self.__str__()

__str__()

String representation of the measurement result.

For backward compatibility with tests, returns a newline-separated list for arrays. For other types, uses a more descriptive representation.

Source code in pytestlab/experiments/results.py
def __str__(self) -> str:
    """String representation of the measurement result.

    For backward compatibility with tests, returns a newline-separated list for arrays.
    For other types, uses a more descriptive representation.
    """
    if isinstance(self.values, UFloat):
        return f"{self.values} {self.units}"
    elif isinstance(self.values, np.float64):
        return f"{self.values} {self.units}"
    elif isinstance(self.values, pl.DataFrame):
        return str(self.values)
    elif isinstance(self.values, np.ndarray):
        # For numpy arrays, handle 1D arrays specially for backward compatibility
        if self.values.ndim == 1:
            return "\n".join([f"{val} {self.units}" for val in self.values])
        # For multi-dimensional arrays, provide a concise representation
        return (
            f"NumPy Array (shape: {self.values.shape}, dtype: {self.values.dtype}) {self.units}"
        )
    elif isinstance(self.values, list):
        # For lists, special handling for backward compatibility
        if all(isinstance(x, int | float) for x in self.values):
            return "\n".join([f"{val} {self.units}" for val in self.values])
        # For lists with mixed types or nested lists, show first few items if long
        if len(self.values) > 5:
            return f"List (first 5 of {len(self.values)}): {self.values[:5]}... {self.units}"
        return f"List: {self.values} {self.units}"

    # Fallback for other types
    return f"Values: {str(self.values)[:100]}... Type: {type(self.values)} {self.units}"

add(value)

Adds a new value to the collection. Behavior depends on self.values type.

Source code in pytestlab/experiments/results.py
def add(self, value: Any) -> None:
    """Adds a new value to the collection. Behavior depends on self.values type."""
    if isinstance(self.values, np.ndarray):
        # This might be inefficient for frequent additions. Consider list then convert.
        self.values = np.append(self.values, value)
    elif isinstance(self.values, list):
        self.values.append(value)
    elif isinstance(self.values, np.float64):
        # Convert to list or ndarray if adding to a single float
        self.values = np.array([self.values, value])
        print("Warning: Added value to np.float64, converted 'values' to np.ndarray.")
    elif isinstance(self.values, UFloat):
        # If current value is UFloat, adding another value implies creating a list/array of UFloats
        self.values = [self.values, value]
        print(
            "Warning: Added value to UFloat, converted 'values' to a list. Consider using a list of UFloats initially."
        )
    elif isinstance(self.values, pl.DataFrame):
        # Appending to Polars DataFrame is complex; typically done by creating a new DF and vstacking.
        # This simple 'add' might not be suitable.
        raise NotImplementedError(
            "Direct 'add' to Polars DataFrame not supported. Use 'set_values' or manage DataFrame externally."
        )
    else:
        raise TypeError(f"Cannot 'add' to type {type(self.values)}")

clear()

Clears all the MeasurementValues from the collection, resetting to an empty/default state.

Source code in pytestlab/experiments/results.py
def clear(self) -> None:
    """Clears all the MeasurementValues from the collection, resetting to an empty/default state."""
    if isinstance(self.values, np.ndarray):
        self.values = np.array([])
    elif isinstance(
        self.values, (np.float64 | UFloat)
    ):  # Reset UFloat to a default float or ufloat(0,0)
        self.values = np.float64(0.0)  # Or ufloat(0,0) if preferred default for UFloat
    elif isinstance(self.values, pl.DataFrame):
        self.values = pl.DataFrame()
    elif isinstance(self.values, list):
        self.values = []
    else:  # Fallback for unknown types, attempt to set to a default float64
        print(
            f"Warning: Clearing unknown type {type(self.values)}, setting to np.float64(0.0)."
        )
        self.values = np.float64(0.0)

get(index)

Gets the MeasurementValue at a specified index. Assumes indexable values.

Source code in pytestlab/experiments/results.py
def get(self, index: int) -> Any:
    """Gets the MeasurementValue at a specified index. Assumes indexable values."""
    if isinstance(self.values, np.ndarray | list):
        return self.values[index]
    elif isinstance(self.values, pl.DataFrame):
        # For DataFrame, 'get' by index might mean row.
        # This returns a new DataFrame with one row.
        return self.values[index]
    elif isinstance(self.values, (np.float64 | UFloat)) and index == 0:
        return self.values
    raise IndexError(
        f"Index {index} out of range or type {type(self.values)} not directly indexable by single int."
    )

get_all()

Returns all the MeasurementValues in the collection.

Source code in pytestlab/experiments/results.py
def get_all(self) -> np.ndarray | pl.DataFrame | np.float64 | list[Any] | UFloat:
    """Returns all the MeasurementValues in the collection."""
    return self.values

items()

Return items for dict-like behavior.

Source code in pytestlab/experiments/results.py
def items(self):
    """Return items for dict-like behavior."""
    return self.to_dict().items()

keys()

Return the keys for dict-like behavior.

Source code in pytestlab/experiments/results.py
def keys(self):
    """Return the keys for dict-like behavior."""
    return self.to_dict().keys()

perform_fft()

Perform Fast Fourier Transform on the measurement data.

Requires: - self.values to be a numpy array of time-domain data - self.sampling_rate to be set (in Hz)

RETURNS DESCRIPTION
MeasurementResult

A new MeasurementResult containing the FFT data, with frequency in Hz

MeasurementResult

and magnitude in the same units as the original data.

Source code in pytestlab/experiments/results.py
def perform_fft(self) -> MeasurementResult:
    """Perform Fast Fourier Transform on the measurement data.

    Requires:
    - self.values to be a numpy array of time-domain data
    - self.sampling_rate to be set (in Hz)

    Returns:
        A new MeasurementResult containing the FFT data, with frequency in Hz
        and magnitude in the same units as the original data.
    """
    if self.sampling_rate is None:
        raise ValueError("Sampling rate must be set to perform FFT")

    if not isinstance(self.values, np.ndarray):
        raise TypeError(f"FFT requires numpy array, got {type(self.values)}")

    # Ensure we're working with a 1D array
    values = self.values.flatten() if self.values.ndim > 1 else self.values

    # Perform FFT
    fft_values = np.fft.rfft(values)
    fft_magnitude = np.abs(fft_values)

    # Create frequency axis
    freqs = np.fft.rfftfreq(len(values), 1 / self.sampling_rate)

    # Create result with frequency and magnitude
    result_df = pl.DataFrame({"frequency": freqs, "magnitude": fft_magnitude})

    return MeasurementResult(
        values=result_df,
        instrument=self.instrument,
        units=self.units,
        measurement_type="FFT",
        timestamp=time.time(),
        original_type=self.measurement_type,
        sampling_rate=self.sampling_rate,
    )

plot(spec=None, **kwargs)

Plot this measurement result.

  • If values is a Polars DataFrame, plots columns per PlotSpec.
  • If values is a 1D numpy array, uses plot_ndarray; honors sampling_rate and units.
  • If values is a scalar/list, attempts to plot as 1D series.
PARAMETER DESCRIPTION
spec

Optional PlotSpec. If not provided, built from kwargs.

TYPE: PlotSpec | None DEFAULT: None

**kwargs

Fields for PlotSpec.

DEFAULT: {}

RETURNS DESCRIPTION

A matplotlib Figure object.

Source code in pytestlab/experiments/results.py
def plot(self, spec: PlotSpec | None = None, **kwargs):
    """
    Plot this measurement result.

    - If values is a Polars DataFrame, plots columns per PlotSpec.
    - If values is a 1D numpy array, uses plot_ndarray; honors sampling_rate and units.
    - If values is a scalar/list, attempts to plot as 1D series.

    Args:
        spec: Optional PlotSpec. If not provided, built from kwargs.
        **kwargs: Fields for PlotSpec.

    Returns:
        A matplotlib Figure object.
    """
    import numpy as np  # local import
    import polars as pl  # local import

    from ..plotting import PlotSpec  # noqa: E402
    from ..plotting import plot_dataframe  # noqa: E402
    from ..plotting import plot_ndarray  # noqa: E402

    pspec = spec or (PlotSpec(**kwargs) if kwargs else PlotSpec())

    if isinstance(self.values, pl.DataFrame):
        # If ylabel not provided and a single numeric y is chosen, prefer units
        if pspec.ylabel is None:
            try:
                # The helper will set y label to the series name by default; we can override
                pspec = PlotSpec(
                    kind=pspec.kind,
                    title=pspec.title or self.measurement_type,
                    x=pspec.x,
                    y=pspec.y,
                    xlabel=pspec.xlabel,
                    ylabel=self.units if self.units else pspec.ylabel,
                    legend=pspec.legend,
                    grid=pspec.grid,
                )
            except Exception:
                pass
        return plot_dataframe(self.values, pspec)

    # Convert scalar/list to numpy array for plotting
    arr: np.ndarray
    if isinstance(self.values, np.ndarray):
        arr = self.values
    elif isinstance(self.values, list):
        arr = np.asarray(self.values)
    else:
        arr = np.asarray([self.values])

    if arr.ndim != 1:
        raise ValueError("Only 1D arrays are supported for simple plotting in Phase 1.")

    title = pspec.title or self.measurement_type
    pspec = PlotSpec(
        kind=pspec.kind,
        title=title,
        x=pspec.x,
        y=pspec.y,
        xlabel=pspec.xlabel,
        ylabel=self.units or pspec.ylabel,
        legend=pspec.legend,
        grid=pspec.grid,
    )
    return plot_ndarray(arr, pspec, sampling_rate=self.sampling_rate, units=self.units)

save(path)

Saves the measurement data to a file.

If the data is a numpy array, it will be saved as a .npy file. If the data is a Polars DataFrame, it will be saved as a .parquet file. Other list-like data will be converted to numpy array and saved as .npy. np.float64 will be saved as a 0-D numpy array. UFloat objects will be saved as a two-element numpy array [nominal, std_dev] in a .npy file.

Source code in pytestlab/experiments/results.py
def save(self, path: str) -> None:
    """Saves the measurement data to a file.

    If the data is a numpy array, it will be saved as a .npy file.
    If the data is a Polars DataFrame, it will be saved as a .parquet file.
    Other list-like data will be converted to numpy array and saved as .npy.
    np.float64 will be saved as a 0-D numpy array.
    UFloat objects will be saved as a two-element numpy array [nominal, std_dev] in a .npy file.
    """
    default_ext = ".npy"
    if isinstance(self.values, pl.DataFrame):
        default_ext = ".parquet"

    if not path.endswith((".npy", ".parquet")):
        path += default_ext
        print(f"Warning: File extension not specified. Saving as {path}")

    if isinstance(self.values, np.ndarray):
        np.save(path, self.values)
    elif isinstance(self.values, pl.DataFrame):
        if not path.endswith(".parquet"):
            print(
                f"Warning: Saving Polars DataFrame to non-parquet file '{path}'. Consider using .parquet for DataFrames."
            )
        self.values.write_parquet(path)
    elif isinstance(self.values, UFloat):
        if not path.endswith(".npy"):
            print(f"Warning: Saving UFloat to non-npy file '{path}'. Consider using .npy.")
        np.save(path, np.array([self.values.nominal_value, self.values.std_dev]))
    elif isinstance(self.values, list | np.float64):  # Convert list or float64 to ndarray
        if not path.endswith(".npy"):
            print(
                f"Warning: Saving {type(self.values).__name__} to non-npy file '{path}'. Consider using .npy."
            )
        np.save(path, np.array(self.values))
    else:
        raise TypeError(
            f"Unsupported data type for saving: {type(self.values)}. Can save np.ndarray, pl.DataFrame, list, np.float64, or UFloat."
        )
    print(f"Measurement saved to {path}")

set_values(values)

Sets the MeasurementValues in the collection.

Source code in pytestlab/experiments/results.py
def set_values(
    self, values: np.ndarray | pl.DataFrame | np.float64 | list[Any] | UFloat | float
) -> None:
    """Sets the MeasurementValues in the collection."""
    if isinstance(values, float) and not isinstance(values, np.floating):
        values = np.float64(values)
    self.values = cast(np.ndarray | pl.DataFrame | np.float64 | list[Any] | UFloat, values)

to_dict()

Convert MeasurementResult to a dict for DataFrame conversion.

This allows MeasurementResult objects to be directly used in Experiment.add_trial.

Source code in pytestlab/experiments/results.py
def to_dict(self) -> dict[str, Any]:
    """Convert MeasurementResult to a dict for DataFrame conversion.

    This allows MeasurementResult objects to be directly used in Experiment.add_trial.
    """
    if isinstance(self.values, pl.DataFrame):
        # If values is already a DataFrame, convert to dict representation
        result = {}
        for col in self.values.columns:
            result[col] = self.values[col].to_list()
        return result
    elif isinstance(self.values, np.ndarray | list):
        # Convert array or list to a dict with a 'values' key
        return {"values": self.values}
    elif isinstance(self.values, np.float64 | UFloat):
        # Convert scalar value to a dict with a 'value' key
        return {"value": self.values}
    else:
        # Default fallback
        return {"values": self.values}

Experiments & Sweeps

pytestlab.experiments.experiments.Experiment(name, description='', notes='')

Experiment tracker to store measurements and parameters.

This class maintains an internal Polars DataFrame (self.data) for trial data, regardless of whether the input is provided as a Polars DataFrame, dict, or list.

It provides two export functionalities
  • save_parquet(file_path): Saves the internal data as a Parquet file.

Additionally, printing the Experiment instance (via str) shows a summary and the head (first few rows) of the data.

Source code in pytestlab/experiments/experiments.py
def __init__(self, name: str, description: str = "", notes: str = "") -> None:
    self.name: str = name
    self.description: str = description
    self.notes: str = notes
    self.parameters: dict[str, ExperimentParameter] = {}
    self.data: pl.DataFrame = pl.DataFrame()

Attributes

data = pl.DataFrame() instance-attribute

description = description instance-attribute

name = name instance-attribute

notes = notes instance-attribute

parameters = {} instance-attribute

Functions

__iter__()

Iterate over each trial (row) as a dictionary.

Source code in pytestlab/experiments/experiments.py
def __iter__(self) -> Iterator[dict[str, Any]]:
    """Iterate over each trial (row) as a dictionary."""
    yield from self.data.to_dicts()

__len__()

Return the number of trials.

Source code in pytestlab/experiments/experiments.py
def __len__(self) -> int:
    """Return the number of trials."""
    return self.data.height

__str__()

Return a string representation of the experiment.

This includes a summary of the experiment details and prints the first 5 rows of the trial data (the head).

Source code in pytestlab/experiments/experiments.py
def __str__(self) -> str:
    """
    Return a string representation of the experiment.

    This includes a summary of the experiment details and prints the first 5 rows
    of the trial data (the head).
    """
    param_str = ", ".join(str(param) for param in self.parameters.values())
    head_data: pl.DataFrame | str
    if not self.data.is_empty():
        head_data = self.data.head(5)
    else:
        head_data = "No trial data available."

    return (
        f"Experiment: {self.name}\n"
        f"Description: {self.description}\n"
        f"Notes: {self.notes or 'No notes'}\n"
        f"Parameters: {param_str}\n"
        f"Trial Data (first 5 rows):\n{head_data}"
    )

add_parameter(name, units, notes='')

Add a new parameter to the experiment.

PARAMETER DESCRIPTION
name

Name of the parameter.

TYPE: str

units

Units for the parameter.

TYPE: str

notes

Additional notes.

TYPE: str DEFAULT: ''

Source code in pytestlab/experiments/experiments.py
def add_parameter(self, name: str, units: str, notes: str = "") -> None:
    """
    Add a new parameter to the experiment.

    Args:
        name (str): Name of the parameter.
        units (str): Units for the parameter.
        notes (str, optional): Additional notes.
    """
    self.parameters[name] = ExperimentParameter(name, units, notes)

add_trial(measurement_result, **parameter_values)

Add a new trial to the experiment.

Accepts measurement data in various formats (list, dict, Polars DataFrame, or MeasurementResult) and converts it into a Polars DataFrame if needed. Additional parameter values are added as new columns.

PARAMETER DESCRIPTION
measurement_result

The measurement data.

TYPE: Union[DataFrame, Dict[str, Any], List[Any], MeasurementResult]

**parameter_values

Additional parameters to include with this trial.

TYPE: Any DEFAULT: {}

RAISES DESCRIPTION
ValueError

If the conversion to a Polars DataFrame fails or if a provided parameter is not defined.

Source code in pytestlab/experiments/experiments.py
def add_trial(
    self,
    measurement_result: pl.DataFrame | dict[str, Any] | list[Any] | MeasurementResult,
    **parameter_values: Any,
) -> None:
    """
    Add a new trial to the experiment.

    Accepts measurement data in various formats (list, dict, Polars DataFrame, or MeasurementResult)
    and converts it into a Polars DataFrame if needed. Additional parameter values
    are added as new columns.

    Args:
        measurement_result (Union[pl.DataFrame, Dict[str, Any], List[Any], MeasurementResult]): The measurement data.
        **parameter_values: Additional parameters to include with this trial.

    Raises:
        ValueError: If the conversion to a Polars DataFrame fails or if a
                    provided parameter is not defined.
    """
    trial_df: pl.DataFrame

    # Special handling for MeasurementResult objects
    if hasattr(measurement_result, "values") and hasattr(measurement_result, "to_dict"):
        # If it's a MeasurementResult, extract its values
        if isinstance(measurement_result.values, pl.DataFrame):
            trial_df = measurement_result.values
        else:
            # Convert to dict and then to DataFrame
            try:
                trial_df = pl.DataFrame(measurement_result.to_dict(), strict=False)
            except Exception as e:
                raise ValueError(
                    f"Failed to convert MeasurementResult to DataFrame: {e}"
                ) from e
    elif not isinstance(measurement_result, pl.DataFrame):
        try:
            trial_df = pl.DataFrame(measurement_result, strict=False)
        except Exception as e:
            raise ValueError(
                f"Failed to convert measurement_result to a Polars DataFrame: {e}"
            ) from e
    else:
        trial_df = measurement_result

    for param_name, value in parameter_values.items():
        if param_name not in self.parameters:
            raise ValueError(
                f"Parameter '{param_name}' is not defined in the experiment. Add it first using add_parameter()."
            )
        trial_df = trial_df.with_columns(pl.lit(value).alias(param_name))

    if self.data.is_empty():
        self.data = trial_df
    else:
        try:
            self.data = self.data.vstack(trial_df)
        except Exception as e:
            raise ValueError(
                f"Failed to stack new trial data. Check for schema compatibility. Error: {e}"
            ) from e

list_trials()

Print the full trials DataFrame.

Source code in pytestlab/experiments/experiments.py
def list_trials(self) -> None:
    """Print the full trials DataFrame."""
    print(self.data)

plot(spec=None, **kwargs)

Plot the experiment's data using the lightweight plotting layer.

PARAMETER DESCRIPTION
spec

Optional PlotSpec. If not provided, a PlotSpec is built from kwargs.

TYPE: PlotSpec | None DEFAULT: None

**kwargs

Convenience fields for PlotSpec (e.g., kind, x, y, title, xlabel, ylabel, legend, grid).

DEFAULT: {}

RETURNS DESCRIPTION

A matplotlib Figure object.

Source code in pytestlab/experiments/experiments.py
def plot(self, spec: PlotSpec | None = None, **kwargs):
    """
    Plot the experiment's data using the lightweight plotting layer.

    Args:
        spec: Optional PlotSpec. If not provided, a PlotSpec is built from kwargs.
        **kwargs: Convenience fields for PlotSpec (e.g., kind, x, y, title, xlabel, ylabel, legend, grid).

    Returns:
        A matplotlib Figure object.
    """
    # Local import to keep plotting optional and to avoid import cycles at module import time
    from ..plotting import PlotSpec  # noqa: E402
    from ..plotting import plot_dataframe  # noqa: E402

    if self.data.is_empty():
        raise ValueError("Experiment has no data to plot.")

    spec_to_use = spec or (PlotSpec(**kwargs) if kwargs else PlotSpec())
    return plot_dataframe(self.data, spec_to_use)

save_parquet(file_path)

Save the internal Polars DataFrame as a Parquet file.

PARAMETER DESCRIPTION
file_path

The file path (including filename) where the Parquet file will be saved.

TYPE: str

Source code in pytestlab/experiments/experiments.py
def save_parquet(self, file_path: str) -> None:
    """
    Save the internal Polars DataFrame as a Parquet file.

    Args:
        file_path (str): The file path (including filename) where the Parquet file will be saved.
    """
    self.data.write_parquet(file_path)
    print(f"Data saved to Parquet file at: {file_path}")

pytestlab.experiments.sweep

Attributes

R = TypeVar('R') module-attribute

T = TypeVar('T') module-attribute

grid_sweep_impl = _grid_sweep_impl module-attribute

gwass_impl = _gwass_impl module-attribute

monte_carlo_sweep_impl = _monte_carlo_sweep_impl module-attribute

Classes

ParameterSpace(ranges='auto', names=None, constraint=None)

Represents a parameter space for sweep operations.

This class helps define and manage parameter spaces for various sweep strategies, including parameter ranges, constraints, and integration with MeasurementSession.

Initialize a parameter space.

PARAMETER DESCRIPTION
ranges

Parameter ranges in one of these formats: - List of (min, max) tuples: [(min1, max1), (min2, max2), ...] - Dict of {name: (min, max)}: {"x": (0, 10), "y": (-5, 5), ...} - "auto" to extract from MeasurementSession

TYPE: list[tuple[float, float]] | str | dict[str, tuple[float, float]] DEFAULT: 'auto'

names

Parameter names (required if ranges is a list of tuples)

TYPE: list[str] | None DEFAULT: None

constraint

Optional function that takes a dict of parameter values and returns True if the combination is valid

TYPE: Callable[[dict[str, float]], bool] | None DEFAULT: None

Source code in pytestlab/experiments/sweep.py
def __init__(
    self,
    ranges: list[tuple[float, float]] | str | dict[str, tuple[float, float]] = "auto",
    names: list[str] | None = None,
    constraint: Callable[[dict[str, float]], bool] | None = None,
):
    """
    Initialize a parameter space.

    Args:
        ranges: Parameter ranges in one of these formats:
            - List of (min, max) tuples: [(min1, max1), (min2, max2), ...]
            - Dict of {name: (min, max)}: {"x": (0, 10), "y": (-5, 5), ...}
            - "auto" to extract from MeasurementSession
        names: Parameter names (required if ranges is a list of tuples)
        constraint: Optional function that takes a dict of parameter values
                    and returns True if the combination is valid
    """
    normalized_ranges: list[tuple[float, float]] | str
    derived_names = names or []

    if isinstance(ranges, dict):
        derived_names = list(ranges.keys())
        normalized_ranges = [ranges[name] for name in derived_names]
    elif isinstance(ranges, list):
        normalized_ranges = ranges
    elif isinstance(ranges, str):
        normalized_ranges = ranges
    else:
        raise TypeError(f"Unsupported parameter space type: {type(ranges)}")

    self.ranges: list[tuple[float, float]] | str = normalized_ranges
    self.names = derived_names or (
        [f"param_{idx}" for idx in range(len(normalized_ranges))]
        if isinstance(normalized_ranges, list)
        else []
    )
    self.constraint = constraint
    self._session: MeasurementSession | None = None

    # Validate ranges and names
    if (
        isinstance(normalized_ranges, list)
        and self.names
        and len(normalized_ranges) != len(self.names)
    ):
        raise ValueError("Number of ranges must match number of parameter names")
Attributes
constraint = constraint instance-attribute
names = derived_names or ([f'param_{idx}' for idx in (range(len(normalized_ranges)))] if isinstance(normalized_ranges, list) else []) instance-attribute
ranges = normalized_ranges instance-attribute
Functions
from_session(session, constraint=None) classmethod

Create a ParameterSpace from a MeasurementSession.

PARAMETER DESCRIPTION
session

A MeasurementSession with defined parameters

TYPE: MeasurementSession

constraint

Optional constraint function

TYPE: Callable | None DEFAULT: None

RETURNS DESCRIPTION
ParameterSpace

A configured parameter space

TYPE: ParameterSpace

Source code in pytestlab/experiments/sweep.py
@classmethod
def from_session(
    cls, session: MeasurementSession, constraint: Callable | None = None
) -> ParameterSpace:
    """
    Create a ParameterSpace from a MeasurementSession.

    Args:
        session: A MeasurementSession with defined parameters
        constraint: Optional constraint function

    Returns:
        ParameterSpace: A configured parameter space
    """
    space = cls("auto", constraint=constraint)
    space._session = session

    # Extract parameter information
    param_names = []
    param_ranges = []

    for name, param in session._parameters.items():
        param_names.append(name)
        values = param.values
        # Calculate numeric range from values; ignore non-numeric entries
        numeric_values = [float(v) for v in values if isinstance(v, int | float)]
        if numeric_values:
            min_val = min(numeric_values)
            max_val = max(numeric_values)
        else:
            min_val = 0.0
            max_val = 0.0
        param_ranges.append((min_val, max_val))

    space.names = param_names
    space.ranges = param_ranges

    return space
get_parameters()

Get parameter information.

RETURNS DESCRIPTION
tuple

(names, ranges) where: - names is a list of parameter names - ranges is a list of (min, max) tuples

TYPE: tuple[list[str], list[tuple[float, float]]]

Source code in pytestlab/experiments/sweep.py
def get_parameters(
    self,
) -> tuple[list[str], list[tuple[float, float]]]:
    """
    Get parameter information.

    Returns:
        tuple: (names, ranges) where:
            - names is a list of parameter names
            - ranges is a list of (min, max) tuples
    """
    # If auto, extract from session
    if self.ranges == "auto":
        if not self._session:
            raise ValueError("'auto' ranges require a MeasurementSession")
        return ParameterSpace.from_session(self._session, self.constraint).get_parameters()

    ranges = self.ranges
    if isinstance(ranges, str):
        raise ValueError(f"Unsupported ranges specification: {ranges}")
    return self.names, ranges
is_valid(param_values)

Check if a parameter combination is valid according to the constraint.

PARAMETER DESCRIPTION
param_values

Parameter values as a list or dict

TYPE: list[float] | dict[str, float]

RETURNS DESCRIPTION
bool

True if valid, False otherwise

TYPE: bool

Source code in pytestlab/experiments/sweep.py
def is_valid(self, param_values: list[float] | dict[str, float]) -> bool:
    """
    Check if a parameter combination is valid according to the constraint.

    Args:
        param_values: Parameter values as a list or dict

    Returns:
        bool: True if valid, False otherwise
    """
    if not self.constraint:
        return True

    # Convert list to dict if needed
    if isinstance(param_values, list):
        param_dict = dict(zip(self.names, param_values, strict=False))
    else:
        param_dict = param_values

    return self.constraint(param_dict)
wrap_function(func)

Wrap a function to handle parameter passing and session integration.

PARAMETER DESCRIPTION
func

The measurement function to wrap

TYPE: Callable

RETURNS DESCRIPTION
Callable

A wrapped function that handles parameters appropriately

TYPE: Callable

Source code in pytestlab/experiments/sweep.py
def wrap_function(self, func: Callable) -> Callable:
    """
    Wrap a function to handle parameter passing and session integration.

    Args:
        func: The measurement function to wrap

    Returns:
        Callable: A wrapped function that handles parameters appropriately
    """
    # Get parameter information
    names, _ = self.get_parameters()

    # Define wrapper function for session usage
    def wrapped_func(*params):
        # Convert positional params to dict
        param_dict = dict(zip(names, params, strict=False))

        # Apply constraint if any
        if self.constraint and not self.constraint(param_dict):
            # Return a default value for invalid combinations
            return float("nan")

        # Call the original function with named parameters
        return func(**param_dict)

    return wrapped_func

Sweep

Dummy Sweep class for documentation compatibility. This is not used in runtime code, but allows mkdocstrings to resolve 'pytestlab.experiments.Sweep' for API docs.

Functions

f_evaluate(params, f)

Source code in pytestlab/experiments/sweep.py
def f_evaluate(params: tuple[Any, ...], f: Callable[..., Any]) -> Any:
    return f(*params)

grid_sweep(param_space=None, points=10)

Apply a grid sweep to a measurement function.

PARAMETER DESCRIPTION
param_space

Parameter space definition, one of: - ParameterSpace object - List of (min, max) tuples - Dict of {name: (min, max)} - "auto" to extract from MeasurementSession

TYPE: ParameterSpace | list[tuple[float, float]] | dict[str, tuple[float, float]] | str | None DEFAULT: None

points

Points per dimension, either: - Single integer (same for all dimensions) - List of integers (one per dimension)

TYPE: int | list[int] DEFAULT: 10

RETURNS DESCRIPTION
Callable

A decorator that applies a grid sweep

TYPE: Callable

Example

@grid_sweep({"voltage": (0, 10), "current": (0, 1)}, 20) def measure(voltage, current): # Measurement code return result

Or with auto parameter extraction from session

@session.acquire @grid_sweep(points=15) def measure(voltage, current, instrument): # Measurement code return result

With constraint

def valid_region(params): return params["voltage"] > 2 * params["current"]

@grid_sweep( ParameterSpace({"voltage": (0, 10), "current": (0, 1)}, constraint=valid_region), points=15 ) def measure(voltage, current): # Measurement code return result

Source code in pytestlab/experiments/sweep.py
def grid_sweep(
    param_space: ParameterSpace
    | list[tuple[float, float]]
    | dict[str, tuple[float, float]]
    | str
    | None = None,
    points: int | list[int] = 10,
) -> Callable:
    """
    Apply a grid sweep to a measurement function.

    Args:
        param_space: Parameter space definition, one of:
            - ParameterSpace object
            - List of (min, max) tuples
            - Dict of {name: (min, max)}
            - "auto" to extract from MeasurementSession
        points: Points per dimension, either:
            - Single integer (same for all dimensions)
            - List of integers (one per dimension)

    Returns:
        Callable: A decorator that applies a grid sweep

    Example:
        @grid_sweep({"voltage": (0, 10), "current": (0, 1)}, 20)
        def measure(voltage, current):
            # Measurement code
            return result

        # Or with auto parameter extraction from session
        @session.acquire
        @grid_sweep(points=15)
        def measure(voltage, current, instrument):
            # Measurement code
            return result

        # With constraint
        def valid_region(params):
            return params["voltage"] > 2 * params["current"]

        @grid_sweep(
            ParameterSpace({"voltage": (0, 10), "current": (0, 1)}, constraint=valid_region),
            points=15
        )
        def measure(voltage, current):
            # Measurement code
            return result
    """
    # Handle different param_space types
    if param_space is None:
        param_space = "auto"

    if not isinstance(param_space, ParameterSpace):
        param_space = ParameterSpace(param_space)

    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # Handle session as first argument
            if args and hasattr(args[0], "_parameters") and hasattr(args[0], "acquire"):
                session = args[0]
                local_space = ParameterSpace.from_session(session, param_space.constraint)

                # Create measurement function that closes over the original func
                def measure_func(**params):
                    return func(**params, **kwargs)

                # Get parameters and run grid sweep
                _, ranges = local_space.get_parameters()
                wrapped_func = local_space.wrap_function(measure_func)

                # Run the original grid_sweep function
                results = grid_sweep_impl(wrapped_func, ranges, points)

                # Format results if needed
                return results
            else:
                # Standard usage without session
                if param_space.ranges == "auto":
                    raise ValueError("'auto' parameter space requires a MeasurementSession")

                # Get parameters and run grid sweep
                _, ranges = param_space.get_parameters()
                wrapped_func = param_space.wrap_function(func)

                # Run the original grid_sweep function
                return grid_sweep_impl(wrapped_func, ranges, points)

        return wrapper

    return decorator

gwass(param_space=None, budget=100, initial_percentage=0.1)

Apply gradient-weighted adaptive stochastic sampling to a measurement function.

PARAMETER DESCRIPTION
param_space

Parameter space definition, one of: - ParameterSpace object - List of (min, max) tuples - Dict of {name: (min, max)} - "auto" to extract from MeasurementSession

TYPE: ParameterSpace | list[tuple[float, float]] | dict[str, tuple[float, float]] | str | None DEFAULT: None

budget

Total number of function evaluations allowed

TYPE: int DEFAULT: 100

initial_percentage

Percentage of budget to use for initial grid

TYPE: float DEFAULT: 0.1

RETURNS DESCRIPTION
Callable

A decorator that applies GWASS

TYPE: Callable

Example

@gwass({"voltage": (0, 10), "current": (0, 1)}, budget=200) def measure(voltage, current): # Measurement code return result

Or with constraint

def valid_region(params): # Only accept points where voltage > 2*current return params["voltage"] > 2 * params["current"]

@gwass( ParameterSpace({"voltage": (0, 10), "current": (0, 1)}, constraint=valid_region), budget=150 ) def measure(voltage, current): # Measurement code return result

Source code in pytestlab/experiments/sweep.py
def gwass(
    param_space: ParameterSpace
    | list[tuple[float, float]]
    | dict[str, tuple[float, float]]
    | str
    | None = None,
    budget: int = 100,
    initial_percentage: float = 0.1,
) -> Callable:
    """
    Apply gradient-weighted adaptive stochastic sampling to a measurement function.

    Args:
        param_space: Parameter space definition, one of:
            - ParameterSpace object
            - List of (min, max) tuples
            - Dict of {name: (min, max)}
            - "auto" to extract from MeasurementSession
        budget: Total number of function evaluations allowed
        initial_percentage: Percentage of budget to use for initial grid

    Returns:
        Callable: A decorator that applies GWASS

    Example:
        @gwass({"voltage": (0, 10), "current": (0, 1)}, budget=200)
        def measure(voltage, current):
            # Measurement code
            return result

        # Or with constraint
        def valid_region(params):
            # Only accept points where voltage > 2*current
            return params["voltage"] > 2 * params["current"]

        @gwass(
            ParameterSpace({"voltage": (0, 10), "current": (0, 1)}, constraint=valid_region),
            budget=150
        )
        def measure(voltage, current):
            # Measurement code
            return result
    """
    # Handle different param_space types
    if param_space is None:
        param_space = "auto"

    if not isinstance(param_space, ParameterSpace):
        param_space = ParameterSpace(param_space)

    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # Handle session as first argument
            if args and hasattr(args[0], "_parameters") and hasattr(args[0], "acquire"):
                session = args[0]
                local_space = ParameterSpace.from_session(session, param_space.constraint)

                # Create measurement function that closes over the original func
                def measure_func(**params):
                    return func(**params, **kwargs)

                # Get parameters and run GWASS
                _, ranges = local_space.get_parameters()
                wrapped_func = local_space.wrap_function(measure_func)

                # Call the original GWASS function
                return gwass_impl(wrapped_func, ranges, budget, initial_percentage)
            else:
                # Standard usage without session
                if param_space.ranges == "auto":
                    raise ValueError("'auto' parameter space requires a MeasurementSession")

                # Get parameters and run GWASS
                _, ranges = param_space.get_parameters()
                wrapped_func = param_space.wrap_function(func)

                return gwass_impl(wrapped_func, ranges, budget, initial_percentage)

        return wrapper

    return decorator

monte_carlo_sweep(param_space=None, samples=50)

Apply a Monte Carlo sweep to a measurement function.

PARAMETER DESCRIPTION
param_space

Parameter space definition, one of: - ParameterSpace object - List of (min, max) tuples - Dict of {name: (min, max)} - "auto" to extract from MeasurementSession

TYPE: ParameterSpace | list[tuple[float, float]] | dict[str, tuple[float, float]] | str | None DEFAULT: None

samples

Number of samples, either: - Single integer (total samples) - List of integers (samples per dimension)

TYPE: int | list[int] DEFAULT: 50

RETURNS DESCRIPTION
Callable

A decorator that applies a Monte Carlo sweep

TYPE: Callable

Example

@monte_carlo_sweep({"voltage": (0, 10), "current": (0, 1)}, 100) def measure(voltage, current): # Measurement code return result

With constraint function

def valid_region(params): return params["voltage"] > 0.5 and params["current"] < 0.8

@monte_carlo_sweep( ParameterSpace({"voltage": (0, 10), "current": (0, 1)}, constraint=valid_region), samples=200 ) def measure(voltage, current): # Measurement code return result

Source code in pytestlab/experiments/sweep.py
def monte_carlo_sweep(
    param_space: ParameterSpace
    | list[tuple[float, float]]
    | dict[str, tuple[float, float]]
    | str
    | None = None,
    samples: int | list[int] = 50,
) -> Callable:
    """
    Apply a Monte Carlo sweep to a measurement function.

    Args:
        param_space: Parameter space definition, one of:
            - ParameterSpace object
            - List of (min, max) tuples
            - Dict of {name: (min, max)}
            - "auto" to extract from MeasurementSession
        samples: Number of samples, either:
            - Single integer (total samples)
            - List of integers (samples per dimension)

    Returns:
        Callable: A decorator that applies a Monte Carlo sweep

    Example:
        @monte_carlo_sweep({"voltage": (0, 10), "current": (0, 1)}, 100)
        def measure(voltage, current):
            # Measurement code
            return result

        # With constraint function
        def valid_region(params):
            return params["voltage"] > 0.5 and params["current"] < 0.8

        @monte_carlo_sweep(
            ParameterSpace({"voltage": (0, 10), "current": (0, 1)}, constraint=valid_region),
            samples=200
        )
        def measure(voltage, current):
            # Measurement code
            return result
    """
    # Handle different param_space types
    if param_space is None:
        param_space = "auto"

    if not isinstance(param_space, ParameterSpace):
        param_space = ParameterSpace(param_space)

    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # Handle session as first argument
            if args and hasattr(args[0], "_parameters") and hasattr(args[0], "acquire"):
                session = args[0]
                local_space = ParameterSpace.from_session(session, param_space.constraint)

                # Create measurement function that closes over the original func
                def measure_func(**params):
                    return func(**params, **kwargs)

                # Get parameters and run Monte Carlo sweep
                _, ranges = local_space.get_parameters()
                wrapped_func = local_space.wrap_function(measure_func)

                # Convert samples to per-dimension if needed
                if isinstance(samples, int):
                    # Equal distribution among parameters
                    per_dim = int(samples ** (1 / len(ranges))) if ranges else samples
                    samples_list: list[int] = [per_dim] * len(ranges)
                else:
                    samples_list = list(samples)

                # Run the original Monte Carlo sweep function
                return monte_carlo_sweep_impl(wrapped_func, ranges, samples_list)
            else:
                # Standard usage without session
                if param_space.ranges == "auto":
                    raise ValueError("'auto' parameter space requires a MeasurementSession")

                # Get parameters and run Monte Carlo sweep
                _, ranges = param_space.get_parameters()
                wrapped_func = param_space.wrap_function(func)

                # Convert samples to per-dimension if needed
                if isinstance(samples, int):
                    # Equal distribution among parameters
                    per_dim = int(samples ** (1 / len(ranges))) if ranges else samples
                    samples_list = [per_dim] * len(ranges)
                else:
                    samples_list = list(samples)

                return monte_carlo_sweep_impl(wrapped_func, ranges, samples_list)

        return wrapper

    return decorator