"""Configuration loading and Pydantic validation."""
from __future__ import annotations
import copy
from pathlib import Path
from typing import Any, Literal, Union
import yaml
from pydantic import BaseModel, Field, ValidationError, field_validator, model_validator
[docs]
class ArcadeConfigError(ValueError):
"""Raised when ARCADE configuration validation fails."""
# ---------------------------------------------------------------------------
# Stage 1: Data config models
# ---------------------------------------------------------------------------
[docs]
class HDF5KeysConfig(BaseModel):
"""HDF5 dataset key names."""
traces: str = "DD"
labels: str = "y"
time: str = "t_bin"
[docs]
class DemodulationConfig(BaseModel):
"""Digital demodulation settings."""
enabled: bool = False
frequencies: list[float] = []
sampling_rate: float = 500e6
time_bin_width: float | None = None
skip_samples: int = 0
correct_iq_offset: bool = True
correct_iq_amplitude: bool = True
pre_demodulated: bool = False
cache_path: str | None = None
cache_force_rebuild: bool = False
cache_auto_save: bool = False
[docs]
class PreprocessingConfig(BaseModel):
"""Preprocessing pipeline settings."""
boxcar_window: int | None = None
truncate_at: int | None = None
normalize: bool = False
[docs]
class TransitionsConfig(BaseModel):
"""Transition detection settings."""
enabled: bool = True
auto: bool = True
method: Literal["centroid_radius", "spectral", "gmm"] = "centroid_radius"
radius_scale: float = 1.0
n_clusters: int = 3
cache_path: str | None = None
cache_force_rebuild: bool = False
spectral_cache_path: str | None = None
spectral_cache_force_rebuild: bool = False
leakage: bool = False
leakage_from_state: int = 1
effective_levels: int | None = None
[docs]
class SplitConfig(BaseModel):
"""Train / val / test splitting (ratio-based or fixed per-class counts)."""
mode: Literal["ratio", "per_class"] = "ratio"
train_ratio: float = 0.60
val_ratio: float = 0.15
test_ratio: float = 0.25
train_per_class: int | None = None
val_per_class: int | None = None
test_per_class: int | None = None
stratified: bool = True
shuffle: bool = True
seed: int = 42
@model_validator(mode="after")
def _ratios_when_ratio_mode(self) -> "SplitConfig":
if self.mode == "ratio":
total = self.train_ratio + self.val_ratio + self.test_ratio
if abs(total - 1.0) > 1e-6:
raise ValueError(f"Split ratios must sum to 1.0, got {total:.6f}")
return self
[docs]
class DataConfig(BaseModel):
"""Full data-stage configuration.
Describes how to load traces and labels, whether to demodulate ADC data,
preprocessing, transition detection, and train/val/test splits. See the
docs config schema and ``configs/example.yaml``.
"""
path: str
format: Literal["auto", "hdf5", "npy", "npz", "pickle"] = "auto"
loader: str = "default"
path_signature_dataset: str = "RIKEN2_Q56_simultaneous"
path_signature_time_truncation: float = 2.0
path_signature_label_source: Literal["expected", "gmm", "post_measured"] = "expected"
path_signature_max_per_class: int | None = None
hdf5_keys: HDF5KeysConfig = HDF5KeysConfig()
num_qubits: int
num_levels: int = 2
addressing: Literal["group", "individual"] = "group"
multiplexed: bool = True
label_format: Literal["state_sorted", "joint", "per_qubit"] = "joint"
qubit_bit_order: Literal["lsb0", "msb0"] = "lsb0"
demodulation: DemodulationConfig = DemodulationConfig()
preprocessing: PreprocessingConfig = PreprocessingConfig()
transitions: TransitionsConfig = TransitionsConfig()
split: SplitConfig = SplitConfig()
custom_processor: str | None = None
@field_validator("num_qubits")
@classmethod
def _positive_qubits(cls, v: int) -> int:
if v < 1:
raise ValueError(f"num_qubits must be >= 1, got {v}")
return v
@field_validator("num_levels")
@classmethod
def _positive_levels(cls, v: int) -> int:
if v < 2:
raise ValueError(f"num_levels must be >= 2, got {v}")
return v
@model_validator(mode="after")
def _demod_requires_config(self) -> "DataConfig":
if self.demodulation.enabled and not self.multiplexed:
raise ValueError(
"Demodulation enabled but multiplexed=False; "
"set multiplexed=True or disable demodulation.",
)
if self.demodulation.enabled and not self.demodulation.pre_demodulated:
if not self.demodulation.frequencies:
raise ValueError(
"Demodulation.enabled requires frequencies (or use pre_demodulated=True)",
)
return self
# ---------------------------------------------------------------------------
# Stage 2: Classify config models
# ---------------------------------------------------------------------------
[docs]
class ClassifiersConfig(BaseModel, extra="allow"):
"""Classifier selection and per-classifier options.
``run`` lists registry names or dotted ``package.module.ClassName`` paths.
Additional keys (e.g. ``fnn:``, ``herqules:``) hold method-specific options
and are allowed via ``extra="allow"``.
"""
run: list[str] = ["threshold"]
custom_classifier: str | None = None
filter_types: list[str] = Field(default_factory=lambda: ["MF"])
threshold_mode: Literal["per_qubit", "best_separation", "otsu"] = "best_separation"
threshold_percentile: float | None = None
name_style: Literal["short", "full"] = "short"
@field_validator("run", mode="before")
@classmethod
def _expand_all_magic(cls, v: Any) -> list[str]:
if isinstance(v, str) and v.lower() == "all":
from arcade.classifiers.base import ensure_classifiers_registered, list_classifiers
ensure_classifiers_registered()
return list_classifiers()
if isinstance(v, list) and len(v) == 1 and isinstance(v[0], str) and v[0].lower() == "all":
from arcade.classifiers.base import ensure_classifiers_registered, list_classifiers
ensure_classifiers_registered()
return list_classifiers()
return v
[docs]
class TuningConfig(BaseModel, extra="allow"):
"""Hyperparameter tuning settings."""
enabled: bool = False
method: Literal["optuna", "grid"] = "optuna"
n_trials: int = 50
metric: str = "val_accuracy"
search_space: dict[str, Any] = {}
[docs]
class ReadoutTimingConfig(BaseModel):
"""Timing model for readout_time = acquisition + transfer + inference."""
sampling_period_ns: float = 2.0
data_transfer_ns: float = 0.0
inference_ns: float = 0.0
[docs]
class SweepConfig(BaseModel):
"""Readout-duration sweep settings."""
enabled: bool = True
n_points: int = 10
lengths: list[int] | None = None
timing: ReadoutTimingConfig = ReadoutTimingConfig()
# ---------------------------------------------------------------------------
# Stage 3: Hardware config models
# ---------------------------------------------------------------------------
[docs]
class DSEConfig(BaseModel):
"""Design-space exploration sweep settings."""
enabled: bool = False
reuse_factors: list[int] = Field(default_factory=lambda: [1, 2, 4, 8, 16])
precisions: list[str] = Field(
default_factory=lambda: ["ap_fixed<16,6>", "ap_fixed<8,3>", "ap_fixed<4,2>"],
)
io_types: list[str] = Field(default_factory=lambda: ["io_parallel", "io_stream"])
[docs]
class HardwareConfig(BaseModel, extra="allow"):
"""FPGA synthesis and resource estimation settings."""
enabled: bool = True
target_fpga: str = "xc7z020"
backend: str = "Vivado"
clock_period_ns: float = 5.0
io_type: Literal["io_parallel", "io_stream"] = "io_parallel"
reuse_factor: int = 1
precision: str = "ap_fixed<16,6>"
run_synthesis: bool = False
run_vivado: bool = False
export_hls4ml: bool = False
export_dir: str = "./output/hls"
dse_enabled: bool = False
dse: DSEConfig = DSEConfig()
[docs]
class PruningConfig(BaseModel):
"""Structured / iterative pruning settings."""
enabled: bool = False
method: Literal["iterative", "structured"] = "iterative"
prune_fraction_per_step: float = 0.1
n_steps: int = 5
finetune_epochs: int = 10
[docs]
class DistillationConfig(BaseModel):
"""Knowledge-distillation settings."""
enabled: bool = False
temperature: float = 3.0
student_sizes: list[int] = Field(default_factory=lambda: [64, 32, 16])
[docs]
class QuantizationConfig(BaseModel):
"""Quantization-aware training settings."""
enabled: bool = False
precision: Literal["int8", "int16", "fp16"] = "int8"
finetune_epochs: int = 5
[docs]
class NASConfig(BaseModel):
"""Neural architecture search settings."""
enabled: bool = False
network_type: Literal["fnn", "hybrid", "transformer"] = "fnn"
n_trials: int = 50
max_params: int | None = None
max_hidden_layers: int = 4
min_hidden_dim: int = 8
max_hidden_dim: int = 512
hardware_aware: bool = False
[docs]
class OptimizationConfig(BaseModel):
"""Post-training classifier optimization (pruning / distill / NAS / quant).
Canonical implementation lives in :mod:`arcade.classifieroptimization`.
"""
enabled: bool = False
target_classifiers: list[str] = Field(default_factory=list)
error_budget: float = 0.01
pruning: PruningConfig = PruningConfig()
distillation: DistillationConfig = DistillationConfig()
quantization: QuantizationConfig = QuantizationConfig()
nas: NASConfig = NASConfig()
# ---------------------------------------------------------------------------
# Stage 4: Visualization config models
# ---------------------------------------------------------------------------
SUMMARY_SECTION_IDS = frozenset(
{
"summary_table",
"confusion_pages",
"sweep_overlay",
"fidelity_table",
"per_qubit_fidelity_table",
"cross_fidelity",
"recommendation",
"per_qubit_bar",
},
)
[docs]
class VisualizationConfig(BaseModel):
"""Visualization and reporting settings."""
output_dir: str = "./output"
format: Literal["pdf", "png", "html"] = "pdf"
dpi: int = 200
stages: list[str] = ["data", "classify", "hardware"]
show_iq_clusters: bool = True
show_averaged_traces: bool = True
report_detail: Literal["summary", "standard", "full"] = "standard"
target_infidelity: float | None = None
summary_sections: list[str] = Field(
default_factory=lambda: [
"summary_table",
"confusion_pages",
"sweep_overlay",
"fidelity_table",
"per_qubit_fidelity_table",
"cross_fidelity",
"per_qubit_bar",
"recommendation",
],
)
@field_validator("stages")
@classmethod
def _reject_old_stage_names(cls, v: list[str]) -> list[str]:
old = {"memory", "compute"} & set(v)
if old:
raise ValueError(
f"visualization.stages no longer accepts {sorted(old)}; "
'use "data" and/or "classify" instead (hard rename, no alias)'
)
return v
@field_validator("summary_sections")
@classmethod
def _check_section_ids(cls, v: list[str]) -> list[str]:
invalid = [s for s in v if s not in SUMMARY_SECTION_IDS]
if invalid:
raise ValueError(
f"Invalid summary_sections: {invalid}. Valid options: {sorted(SUMMARY_SECTION_IDS)}"
)
return v
[docs]
class PipelineStepConfig(BaseModel, extra="allow"):
"""One composable pipeline block (Phase 2 mix-and-match)."""
block: str
enabled: bool = True
[docs]
class PipelineConfig(BaseModel):
"""Optional block pipeline overriding implicit stage wiring."""
steps: list[PipelineStepConfig] = Field(default_factory=list)
[docs]
class QECCodeConfig(BaseModel, extra="allow"):
"""Per-code stim parameters."""
distance: int = 3
rounds: int = 100
[docs]
class QECConfig(BaseModel, extra="allow"):
"""Stim-based logical error rate + FTQC timing curves (Phase 6).
Use for readout-error ↔ accuracy maps, LER vs readout duration (cycle-time
tradeoff), and decoder-latency vs readout time (wall-clock memory
experiments) on surface / color / BB codes.
"""
enabled: bool = False
shots: int = 100_000
seed: int = 42
codes: list[str] = Field(default_factory=lambda: ["surface"])
experiments: list[str] = Field(default_factory=lambda: ["memory"])
surface: QECCodeConfig = QECCodeConfig()
bb: QECCodeConfig = Field(default_factory=lambda: QECCodeConfig(distance=6))
color: QECCodeConfig = QECCodeConfig()
walking_surface: QECCodeConfig = QECCodeConfig()
readout_binding: dict[str, Any] = Field(default_factory=dict)
# Per-code decoder override; defaults: surface→pymatching, color→sinter_internal, bb→belief_matching
decoders: dict[str, str] = Field(default_factory=dict)
accuracy_error_curve: bool = True
ler_duration_curve: bool = True
synthetic_duration_curve: bool = False
curve_points: int = 8
curve_shots: int = 5_000
# Idle / control overhead for LER (ns). Idle p_phys scales with
# readout + control_overhead (+ decode if include_decode_in_idle).
control_overhead_ns: float = 100.0
tau_idle_ns: float = 50_000.0
include_decode_in_idle: bool = False
idle_p_floor: float = 1e-4
[docs]
class ArcadeConfig(BaseModel):
"""Top-level ARCADE configuration, validated via Pydantic."""
data: DataConfig
classifiers: ClassifiersConfig = ClassifiersConfig()
tuning: TuningConfig = TuningConfig()
sweep: SweepConfig = SweepConfig()
hardware: HardwareConfig = HardwareConfig()
visualization: VisualizationConfig = VisualizationConfig()
optimization: OptimizationConfig = OptimizationConfig()
pipeline: PipelineConfig | None = None
qec: QECConfig = QECConfig()
model_config = {"extra": "allow"}
# ---------------------------------------------------------------------------
# Loading
# ---------------------------------------------------------------------------
[docs]
def load_config(source: Union[str, Path, dict]) -> ArcadeConfig:
"""Load and validate an ARCADE configuration.
Args:
source: Path to a YAML file, or an already-parsed ``dict``.
Returns:
A validated :class:`ArcadeConfig` instance.
Raises:
FileNotFoundError: YAML path does not exist.
TypeError: *source* is neither a path nor a dict.
ArcadeConfigError: Pydantic validation failed (message lists field paths).
"""
if isinstance(source, dict):
raw = copy.deepcopy(source)
elif isinstance(source, (str, Path)):
path = Path(source)
if not path.exists():
raise FileNotFoundError(f"Config file not found: {path}")
with open(path, "r") as fh:
raw = yaml.safe_load(fh)
if raw is None:
raw = {}
else:
raise TypeError(f"Expected a dict or file path, got {type(source).__name__}")
try:
return ArcadeConfig(**raw)
except ValidationError as exc:
errors = exc.errors()
parts = []
for err in errors:
loc = " → ".join(str(l) for l in err["loc"])
parts.append(f" • {loc}: {err['msg']}")
raise ArcadeConfigError(
f"Invalid ARCADE configuration ({len(errors)} error(s)):\n"
+ "\n".join(parts)
+ "\n\nCheck your YAML/dict against the ARCADE config schema."
) from exc
# ---------------------------------------------------------------------------
# Template generation
# ---------------------------------------------------------------------------
[docs]
def generate_template(num_qubits: int, classifiers: list[str] | None = None) -> str:
"""Return a commented YAML config string with all sections.
Args:
num_qubits: Number of qubits for the data section.
classifiers: List of classifier names to include; defaults to ["threshold"].
Returns:
A YAML string with comments suitable for writing to a config file.
"""
if classifiers is None:
classifiers = ["threshold"]
clf_list = "\n".join(f" - {c}" for c in classifiers)
return (
"# =============================================================================\n"
"# ARCADE Configuration Template\n"
f"# Generated for {num_qubits}-qubit system\n"
"# =============================================================================\n"
"\n"
"# --- Stage 1: Data ---\n"
"data:\n"
" # Path to raw measurement data (HDF5, npy, npz, or pickle)\n"
' path: "./data/measurements.h5"\n'
" format: auto\n"
f" num_qubits: {num_qubits}\n"
" num_levels: 2\n"
" addressing: group # group | individual\n"
" multiplexed: true\n"
" label_format: joint # state_sorted | joint | per_qubit\n"
" qubit_bit_order: lsb0 # lsb0 | msb0\n"
"\n"
" hdf5_keys:\n"
" traces: DD\n"
' labels: "y"\n'
" time: t_bin\n"
"\n"
" demodulation:\n"
" enabled: false\n"
" frequencies: []\n"
" sampling_rate: 500e6\n"
" skip_samples: 0\n"
" correct_iq_offset: true\n"
" correct_iq_amplitude: true\n"
" pre_demodulated: false\n"
"\n"
" preprocessing:\n"
" boxcar_window: null # Set integer for boxcar smoothing\n"
" truncate_at: null # Truncate traces at this sample index\n"
" normalize: false\n"
"\n"
" transitions:\n"
" enabled: true\n"
" auto: true\n"
" method: centroid_radius # centroid_radius | spectral | gmm\n"
" radius_scale: 1.0\n"
" n_clusters: 3\n"
" cache_path: null\n"
" cache_force_rebuild: false\n"
" leakage: false\n"
"\n"
" split:\n"
" mode: ratio # ratio | per_class\n"
" train_ratio: 0.60\n"
" val_ratio: 0.15\n"
" test_ratio: 0.25\n"
" stratified: true\n"
" shuffle: true\n"
" seed: 42\n"
"\n"
"# --- Stage 2: Classify ---\n"
"classifiers:\n"
" run:\n"
f"{clf_list}\n"
" filter_types:\n"
" - MF\n"
" threshold_mode: best_separation # per_qubit | best_separation | otsu\n"
" name_style: short # short | full\n"
"\n"
"tuning:\n"
" enabled: false\n"
" method: optuna # optuna | grid\n"
" n_trials: 50\n"
" metric: val_accuracy\n"
" search_space: {}\n"
"\n"
"sweep:\n"
" enabled: true\n"
" n_points: 10\n"
" timing:\n"
" sampling_period_ns: 2.0\n"
" data_transfer_ns: 0.0\n"
" inference_ns: 0.0\n"
"\n"
"# --- Stage 3: Hardware ---\n"
"hardware:\n"
" enabled: true\n"
" target_fpga: xc7z020\n"
" backend: Vivado\n"
" clock_period_ns: 5.0\n"
" io_type: io_parallel # io_parallel | io_stream\n"
" reuse_factor: 1\n"
' precision: "ap_fixed<16,6>"\n'
" run_synthesis: false\n"
"\n"
"# --- Stage 4: Visualization ---\n"
"visualization:\n"
" output_dir: ./output\n"
" format: pdf # pdf | png | html\n"
" dpi: 200\n"
" report_detail: standard # summary | standard | full\n"
" target_infidelity: null # Optional target infidelity threshold\n"
" stages:\n"
" - data\n"
" - classify\n"
" - hardware\n"
" show_iq_clusters: true\n"
" show_averaged_traces: true\n"
" summary_sections:\n"
" - summary_table\n"
" - confusion_pages\n"
" - sweep_overlay\n"
" - fidelity_table\n"
" - per_qubit_fidelity_table\n"
" - cross_fidelity\n"
" - per_qubit_bar\n"
" - recommendation\n"
"\n"
"# --- QEC (optional) ---\n"
"qec:\n"
" enabled: false\n"
" shots: 100000\n"
" seed: 42\n"
" codes:\n"
" - surface\n"
" experiments:\n"
" - memory\n"
)
# ---------------------------------------------------------------------------
# Validation helpers
# ---------------------------------------------------------------------------
[docs]
def validate_and_explain(source: Union[str, Path, dict]) -> list[str]:
"""Validate a config source and return human-readable error messages.
Args:
source: A dict, file path string, or Path to a YAML config.
Returns:
A list of human-readable error strings. Empty list means valid.
"""
if isinstance(source, dict):
raw = copy.deepcopy(source)
elif isinstance(source, (str, Path)):
path = Path(source)
if not path.exists():
return [f"Config file not found: {path}"]
with open(path, "r") as fh:
raw = yaml.safe_load(fh)
if raw is None:
raw = {}
else:
return [f"Expected a dict or file path, got {type(source).__name__}"]
try:
ArcadeConfig(**raw)
except ValidationError as exc:
messages = []
for err in exc.errors():
loc = " -> ".join(str(part) for part in err["loc"])
err_type = err.get("type", "unknown")
msg = err["msg"]
if err_type == "missing":
messages.append(f"Missing required field '{loc}': please provide a value.")
elif err_type == "value_error":
messages.append(f"Invalid value at '{loc}': {msg}")
elif err_type in ("string_type", "int_type", "float_type", "bool_type"):
expected = err_type.replace("_type", "")
messages.append(f"Wrong type at '{loc}': expected {expected}, {msg}")
elif err_type == "literal_error":
messages.append(f"Invalid choice at '{loc}': {msg}")
else:
messages.append(f"Error at '{loc}' ({err_type}): {msg}")
return messages
return []