arcade.data

Data loading and processing for ARCADE.

class arcade.data.BaseDataLoader[source]

Bases: ABC

Interface for loading readout data from any source.

Subclass this and decorate with @register_loader("name") to make the loader discoverable by the pipeline.

Example:

@register_loader("my_format")
class MyLoader(BaseDataLoader):
    def load(self, path, **kwargs):
        ...
        return DataBundle(traces=traces, labels=labels)
classmethod from_config(config)[source]

Construct a loader instance from a config dict.

Return type:

BaseDataLoader

Parameters:

config (dict[str, Any])

abstractmethod load(path, **kwargs)[source]

Load traces and labels from path.

Parameters:

  • path (str | Path) – File or directory to load from.

  • **kwargs (Any) – Loader-specific options.

Return type:

DataBundle

Returns:

A DataBundle containing the loaded data.

name: str = 'base'
supported_formats: tuple[str, ...] = ()
class arcade.data.BaseDataProcessor[source]

Bases: ABC

Interface for data processing steps (demod, preprocess, split, etc.).

Subclass this and decorate with @register_processor("name").

name: str = 'base'
abstractmethod process(bundle, **kwargs)[source]

Apply a processing step to the bundle.

Parameters:

  • bundle (DataBundle) – Current data state.

  • **kwargs (Any) – Processor-specific options.

Return type:

DataBundle

Returns:

The (possibly modified) DataBundle.

requires: list[str] = []
arcade.data.BaseLoader

alias of BaseDataLoader

arcade.data.BaseProcessor

alias of BaseDataProcessor

class arcade.data.DataBundle(traces, labels=None, time=None, metadata=<factory>)[source]

Bases: object

Standardized container for loaded readout data.

Parameters:
traces

IQ trace array. Shape is (n_shots, n_time, 2) or (n_states, n_shots, n_time, 2) for state-sorted data.

labels

Integer class labels, or None if not available.

time

Time-axis array, or None.

metadata

Provenance information (source path, format, etc.).

labels: ndarray | None = None
time: ndarray | None = None
traces: ndarray
metadata: dict[str, Any]
class arcade.data.DataStageOutput(bundle, traces, labels, label_set, splits, transition_result=None, spectral_labels=None, relaxation_result=None, metadata=<factory>)[source]

Bases: object

Output of the data preparation stage.

Parameters:
relaxation_result: Any | None = None
replace(**kwargs)[source]
Return type:

DataStageOutput

Parameters:

kwargs (Any)

spectral_labels: dict[int, ndarray] | None = None
transition_result: Any | None = None
bundle: Any
traces: Union[ndarray, dict[int, ndarray]]
labels: ndarray
label_set: LabelSet
splits: Any
metadata: dict[str, Any]
class arcade.data.FeatureStageOutput(filter_set=None, filter_types=(), transition_result=None, metadata=<factory>)[source]

Bases: object

Output of the feature / filter-bank stage.

Parameters:
filter_set: Any | None = None
filter_types: tuple[str, ...] = ()
replace(**kwargs)[source]
Return type:

FeatureStageOutput

Parameters:

kwargs (Any)

transition_result: Any | None = None
metadata: dict[str, Any]
class arcade.data.ClassifyStageOutput(classifiers=<factory>, sweep=None, metadata=<factory>)[source]

Bases: object

Output of the classifier training and evaluation stage.

Parameters:
replace(**kwargs)[source]
Return type:

ClassifyStageOutput

Parameters:

kwargs (Any)

sweep: dict[str, Any] | None = None
classifiers: dict[str, Any]
metadata: dict[str, Any]
class arcade.data.LabelSet(joint, per_qubit, num_qubits, num_levels, qubit_bit_order='lsb0')[source]

Bases: object

Joint and per-qubit labels with bookkeeping metadata.

Parameters:
qubit_bit_order: str = 'lsb0'
replace(**kwargs)[source]
Return type:

LabelSet

Parameters:

kwargs (Any)

joint: ndarray
per_qubit: ndarray
num_qubits: int
num_levels: int
class arcade.data.SplitIndices(train_indices, val_indices, test_indices)[source]

Bases: object

Train / validation / test index arrays.

Parameters:
replace(**kwargs)[source]
Return type:

SplitIndices

Parameters:

kwargs (Any)

train_indices: ndarray
val_indices: ndarray
test_indices: ndarray
exception arcade.data.LevelMismatchError[source]

Bases: ValueError

Raised when detected cluster count does not match configured levels.

Expected clusters equal num_levels, or num_levels + 1 when leakage detection is enabled.

exception arcade.data.InsufficientShotsError[source]

Bases: ValueError

Raised when a labeled population is too small for a detector fit.

class arcade.data.HDF5Loader(traces_key='DD', labels_key='y', time_key='t_bin')[source]

Bases: BaseDataLoader

Load readout traces from HDF5 files.

Parameters:

  • traces_key (str)

  • labels_key (str)

  • time_key (str)

classmethod from_config(config)[source]

Construct a loader instance from a config dict.

Return type:

HDF5Loader

Parameters:

config (dict[str, Any])

load(path, **kwargs)[source]

Load traces and labels from path.

Parameters:

  • path (str | Path) – File or directory to load from.

  • **kwargs (Any) – Loader-specific options.

Return type:

DataBundle

Returns:

A DataBundle containing the loaded data.

name: str = 'hdf5'
supported_formats: tuple[str, ...] = ('.h5', '.hdf5')
class arcade.data.NPZLoader[source]

Bases: BaseDataLoader

Load readout traces from .npy or .npz files.

load(path, **kwargs)[source]

Load traces and labels from path.

Parameters:

  • path (str | Path) – File or directory to load from.

  • **kwargs (Any) – Loader-specific options.

Return type:

DataBundle

Returns:

A DataBundle containing the loaded data.

name: str = 'npz'
supported_formats: tuple[str, ...] = ('.npy', '.npz')
class arcade.data.PathSignatureLoader(dataset_name='RIKEN2_Q56_simultaneous', time_truncation=2.0, label_source='expected', max_per_class=None, subsample_seed=42)[source]

Bases: BaseDataLoader

Load Cao et al. path-signature benchmark datasets.

Parameters:

  • dataset_name (str)

  • time_truncation (float)

  • label_source (str)

  • max_per_class (int | None)

  • subsample_seed (int)

classmethod from_config(config)[source]

Construct a loader instance from a config dict.

Return type:

PathSignatureLoader

Parameters:

config (dict[str, Any])

load(path, **kwargs)[source]

Load traces and labels from path.

Parameters:

  • path (str | Path) – File or directory to load from.

  • **kwargs (Any) – Loader-specific options.

Return type:

DataBundle

Returns:

A DataBundle containing the loaded data.

name: str = 'path_signature'
supported_formats: tuple[str, ...] = ()
class arcade.data.PickleLoader[source]

Bases: BaseDataLoader

Load readout traces from pickle files.

load(path, **kwargs)[source]

Load traces and labels from path.

Parameters:

  • path (str | Path) – File or directory to load from.

  • **kwargs (Any) – Loader-specific options.

Return type:

DataBundle

Returns:

A DataBundle containing the loaded data.

name: str = 'pickle'
supported_formats: tuple[str, ...] = ('.pkl', '.pickle')
class arcade.data.DemodulationProcessor(frequencies=None, sampling_rate=None)[source]

Bases: BaseDataProcessor

Demodulate raw ADC traces into per-qubit IQ data.

Parameters:

  • frequencies (dict | None)

  • sampling_rate (float | None)

name: str = 'demodulation'
process(bundle, **kwargs)[source]

Apply a processing step to the bundle.

Parameters:

  • bundle (DataBundle) – Current data state.

  • **kwargs (Any) – Processor-specific options.

Return type:

DataBundle

Returns:

The (possibly modified) DataBundle.

class arcade.data.BoxcarProcessor(window=25)[source]

Bases: BaseDataProcessor

Non-overlapping boxcar (moving-average) downsampling.

Parameters:

window (int)

name: str = 'boxcar'
process(bundle, **kwargs)[source]

Apply a processing step to the bundle.

Parameters:

  • bundle (DataBundle) – Current data state.

  • **kwargs (Any) – Processor-specific options.

Return type:

DataBundle

Returns:

The (possibly modified) DataBundle.

class arcade.data.SplitProcessor[source]

Bases: BaseDataProcessor

Stratified train / validation / test splitting (processor wrapper).

name: str = 'split'
process(bundle, **kwargs)[source]

Apply a processing step to the bundle.

Parameters:

  • bundle (DataBundle) – Current data state.

  • **kwargs (Any) – Processor-specific options.

Return type:

DataBundle

Returns:

The (possibly modified) DataBundle.

class arcade.data.TransitionProcessor(method='centroid', spectral=False, include_leakage=False)[source]

Bases: BaseDataProcessor

Detect transition trajectories (relaxation, excitation, leakage).

Parameters:
name: str = 'transitions'
process(bundle, **kwargs)[source]

Apply a processing step to the bundle.

Parameters:

  • bundle (DataBundle) – Current data state.

  • **kwargs (Any) – Processor-specific options.

Return type:

DataBundle

Returns:

The (possibly modified) DataBundle.

class arcade.data.TransitionResult(num_qubits, num_levels, per_qubit=<factory>)[source]

Bases: object

Per-qubit separation of clean, relaxation, excitation, and leakage traces.

Parameters:
centroids(qubit)[source]
Return type:

dict[int, tuple[float, float]]

Parameters:

qubit (int)

clean_traces(qubit, state)[source]
Return type:

ndarray

Parameters:
num_qubits: int
num_levels: int
per_qubit: dict[int, dict[str, Any]]
arcade.data.get_loader(name)[source]

Look up a loader by registry name or dotted import path.

Return type:

type[BaseDataLoader]

Parameters:

name (str)

arcade.data.get_processor(name)[source]

Look up a processor by registry name or dotted import path.

Return type:

type[BaseDataProcessor]

Parameters:

name (str)

arcade.data.list_loaders()[source]

Return the names of all registered loaders.

Return type:

list[str]

arcade.data.list_processors()[source]

Return the names of all registered processors.

Return type:

list[str]

arcade.data.load_data(path, config)[source]

Dispatch to the configured loader (default, registered name, or dotted path).

Return type:

DataBundle

Parameters:
arcade.data.register_loader(name)[source]

Decorator to register a data loader class by name.

Parameters:

name (str)

arcade.data.register_processor(name)[source]

Decorator to register a data processor class by name.

Example:

@register_processor("my_step")
class MyStep(BaseDataProcessor):
    def process(self, bundle, **kwargs):
        ...
        return bundle
Parameters:

name (str)