"""Typed contracts for data-stage boundaries."""
from __future__ import annotations
from dataclasses import dataclass, field, replace as _dc_replace
from typing import Any, Union
import numpy as np
TraceLike = Union[np.ndarray, dict[int, np.ndarray]]
[docs]
class LevelMismatchError(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.
"""
[docs]
class InsufficientShotsError(ValueError):
"""Raised when a labeled population is too small for a detector fit."""
[docs]
@dataclass(frozen=True)
class LabelSet:
"""Joint and per-qubit labels with bookkeeping metadata."""
joint: np.ndarray
per_qubit: np.ndarray
num_qubits: int
num_levels: int
qubit_bit_order: str = "lsb0"
[docs]
def replace(self, **kwargs: Any) -> "LabelSet":
return _dc_replace(self, **kwargs)
[docs]
@dataclass(frozen=True)
class SplitIndices:
"""Train / validation / test index arrays."""
train_indices: np.ndarray
val_indices: np.ndarray
test_indices: np.ndarray
[docs]
def replace(self, **kwargs: Any) -> "SplitIndices":
return _dc_replace(self, **kwargs)
def __post_init__(self) -> None:
for name in ("train_indices", "val_indices", "test_indices"):
arr = getattr(self, name)
if not isinstance(arr, np.ndarray):
object.__setattr__(self, name, np.asarray(arr, dtype=np.intp))
[docs]
@dataclass(frozen=True)
class DataStageOutput:
"""Output of the data preparation stage."""
bundle: Any
traces: TraceLike
labels: np.ndarray
label_set: LabelSet
splits: Any
transition_result: Any | None = None
spectral_labels: dict[int, np.ndarray] | None = None
relaxation_result: Any | None = None
metadata: dict[str, Any] = field(default_factory=dict)
[docs]
def replace(self, **kwargs: Any) -> "DataStageOutput":
return _dc_replace(self, **kwargs)
[docs]
@dataclass(frozen=True)
class FeatureStageOutput:
"""Output of the feature / filter-bank stage."""
filter_set: Any | None = None
filter_types: tuple[str, ...] = ()
transition_result: Any | None = None
metadata: dict[str, Any] = field(default_factory=dict)
[docs]
def replace(self, **kwargs: Any) -> "FeatureStageOutput":
return _dc_replace(self, **kwargs)
[docs]
@dataclass(frozen=True)
class ClassifyStageOutput:
"""Output of the classifier training and evaluation stage."""
classifiers: dict[str, Any] = field(default_factory=dict)
sweep: dict[str, Any] | None = None
metadata: dict[str, Any] = field(default_factory=dict)
[docs]
def replace(self, **kwargs: Any) -> "ClassifyStageOutput":
return _dc_replace(self, **kwargs)