"""High-level convenience API for ARCADE."""
from __future__ import annotations
from pathlib import Path
from typing import Any
import numpy as np
from ._logging import get_logger
log = get_logger(__name__)
[docs]
def load(
path: str | Path,
num_qubits: int,
*,
num_levels: int = 2,
format: str = "auto",
**kwargs: Any,
) -> dict[str, Any]:
"""Load and prepare readout data for classification (data stage).
Prefer YAML + :func:`arcade.run` for full bake-offs. Use :func:`load` when
you want an in-memory bundle for :func:`compare` or custom scripts.
Args:
path: Path to data file (HDF5, NPY, NPZ, or pickle).
num_qubits: Number of qubits in the data.
num_levels: Energy levels per qubit (default ``2``).
format: File format or ``"auto"`` for detection.
**kwargs: Extra ``data:`` config fields (e.g. ``hdf5_keys``,
``demodulation``, ``multiplexed``, ``qubit_bit_order``).
Returns:
Prepared data dict with traces, labels, splits, and related stage outputs.
Raises:
arcade.config.ArcadeConfigError: Invalid combined config.
FileNotFoundError: *path* does not exist (via loader).
"""
from .config import load_config
from .stages import run_data_stage
cfg_dict = {
"data": {
"path": str(path),
"num_qubits": num_qubits,
"num_levels": num_levels,
"format": format,
**kwargs,
}
}
cfg = load_config(cfg_dict)
return run_data_stage(cfg)
[docs]
def compare(
data: dict[str, Any] | str | Path,
classifiers: list[str] | None = None,
*,
num_qubits: int | None = None,
num_levels: int = 2,
sweep: bool = True,
hardware: bool = True,
report: bool = False,
output_dir: str = "./output",
**kwargs: Any,
) -> Any:
"""Run multiple classifiers and compare results.
For production zoo runs, prefer ``arcade run configs/….yaml`` so demod,
filters, and report sections stay in one file. :func:`compare` is the
compact Python helper.
Args:
data: Prepared dict from :func:`load`, or a path to a data file.
classifiers: Registry names. ``None`` defaults to
``["threshold", "svm", "fnn"]``.
num_qubits: Required when *data* is a file path.
num_levels: Levels per qubit (default ``2``).
sweep: Enable readout-duration sweep.
hardware: Enable analytical FPGA resource estimation.
report: Generate a summary PDF/HTML report when supported.
output_dir: Output directory for plots / reports.
**kwargs: Merged into the top-level config dict when *data* is a path
(advanced; prefer a YAML file for complex recipes).
Returns:
:class:`~arcade.results.PipelineResults` when available, otherwise a
plain results dict with ``data`` / ``classify`` / ``hardware`` keys.
Raises:
ValueError: File-path *data* without *num_qubits*.
"""
from .config import load_config
from .runner import run_pipeline
if classifiers is None:
classifiers = ["threshold", "svm", "fnn"]
if isinstance(data, (str, Path)):
if num_qubits is None:
raise ValueError("num_qubits is required when data is a file path")
cfg_dict: dict[str, Any] = {
"data": {
"path": str(data),
"num_qubits": num_qubits,
"num_levels": num_levels,
},
"classifiers": {"run": classifiers},
"sweep": {"enabled": sweep},
"hardware": {"enabled": hardware},
"visualization": {"output_dir": output_dir},
}
cfg_dict.update(kwargs)
return run_pipeline(cfg_dict)
from .stages import run_classify_stage, run_hardware_stage, run_visualization_stage
nq = num_qubits or 1
if isinstance(data.get("label_set"), dict):
nq = data["label_set"].get("num_qubits", nq)
cfg_dict = {
"data": {
"path": "memory://preloaded",
"num_qubits": nq,
"num_levels": num_levels,
},
"classifiers": {"run": classifiers},
"sweep": {"enabled": sweep},
"hardware": {"enabled": hardware},
"visualization": {"output_dir": output_dir},
}
cfg = load_config(cfg_dict)
classify_out = run_classify_stage(cfg, data)
hardware_out = (
run_hardware_stage(cfg, classify_out) if hardware else {"reports": {}, "scalability": {}}
)
report_path = None
if report:
report_path = run_visualization_stage(cfg, data, classify_out, hardware_out)
try:
from .results import PipelineResults
return PipelineResults(
{
"data": data,
"classify": classify_out,
"hardware": hardware_out,
"qec": {},
"report_path": report_path,
}
)
except ImportError:
return {
"data": data,
"classify": classify_out,
"hardware": hardware_out,
"report_path": report_path,
}
[docs]
def explore(data: dict[str, Any] | str | Path, *, num_qubits: int | None = None) -> None:
"""Print a human-readable data-quality summary to stdout.
Shows trace shapes, label histogram, split sizes, and whether transition
detection ran. Intended for interactive debugging (also via ``arcade explore``).
Args:
data: Prepared dict from :func:`load`, or a path to load first.
num_qubits: Required when *data* is a file path.
Raises:
ValueError: File-path *data* without *num_qubits*.
"""
if isinstance(data, (str, Path)):
if num_qubits is None:
raise ValueError("num_qubits required when data is a file path")
data = load(data, num_qubits)
traces = data.get("traces")
labels = data.get("labels")
splits = data.get("splits")
print("=== ARCADE Data Summary ===")
if traces is not None:
if isinstance(traces, dict):
for q, t in sorted(traces.items()):
print(f" Qubit {q}: {t.shape}")
else:
arr = np.asarray(traces)
print(f" Traces shape: {arr.shape}")
print(f" Dtype: {arr.dtype}")
if labels is not None:
labs = np.asarray(labels)
unique, counts = np.unique(labs, return_counts=True)
print(f" Labels: {len(labs)} total, {len(unique)} classes")
for u, c in zip(unique, counts):
print(f" Class {u}: {c} shots ({100 * c / len(labs):.1f}%)")
if splits is not None:
print(f" Train: {len(splits.train_labels)}")
print(f" Val: {len(splits.val_labels)}")
print(f" Test: {len(splits.test_labels)}")
tr = data.get("transition_result")
if tr is not None:
print(f" Transition detection: {tr.num_qubits} qubits, {tr.num_levels} levels")
[docs]
def describe(classifier_name: str) -> str:
"""Return a multi-line description of a registered classifier.
Args:
classifier_name: Registry name (e.g. ``"fnn"``, ``"threshold"``).
Returns:
Text including category, ``FEATURE_KIND``, and training / NN requirements.
If the name is unknown, returns an error string listing available names
(does not raise).
"""
from .classifiers.base import ensure_classifiers_registered, get_classifier, list_classifiers
ensure_classifiers_registered()
try:
cls = get_classifier(classifier_name)
except KeyError:
available = list_classifiers()
return f"Unknown classifier '{classifier_name}'. Available: {', '.join(available)}"
name = getattr(cls, "name", classifier_name)
desc = getattr(cls, "DESCRIPTION", "No description available.")
category = getattr(cls, "CATEGORY", "unknown")
feat = getattr(cls, "FEATURE_KIND", "filter_features")
needs_training = getattr(cls, "REQUIRES_TRAINING", True)
needs_nn = getattr(cls, "REQUIRES_NN_CONFIG", False)
lines = [
f"{name} ({category})",
f" {desc}",
f" Feature kind: {feat}",
f" Requires training: {needs_training}",
f" Requires NN config: {needs_nn}",
]
return "\n".join(lines)