"""End-to-end pipeline orchestrator (thin entrypoint over ``stages``)."""
from __future__ import annotations
from pathlib import Path
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from .results import PipelineResults
from .stages.base import (
BaseClassifyStage,
BaseDataStage,
BaseHardwareStage,
BaseOptimizationStage,
BaseVisualizationStage,
)
from ._logging import configure_logging, get_logger
log = get_logger(__name__)
[docs]
class ArcadePipeline:
"""Injectable pipeline wiring data → classify → optimization → hardware → viz.
Pass custom stage instances to override defaults; omitted stages use the
``Default*Stage`` implementations from :mod:`arcade.stages`.
"""
[docs]
def __init__(
self,
*,
data_stage: "BaseDataStage | None" = None,
classify_stage: "BaseClassifyStage | None" = None,
optimization_stage: "BaseOptimizationStage | None" = None,
hardware_stage: "BaseHardwareStage | None" = None,
visualization_stage: "BaseVisualizationStage | None" = None,
) -> None:
"""Wire optional custom stages into the pipeline.
Args:
data_stage: Data stage instance, or ``None`` for
:class:`~arcade.stages.DefaultDataStage`.
classify_stage: Classify stage instance, or ``None`` for
:class:`~arcade.stages.DefaultClassifyStage`.
optimization_stage: Optimization stage instance, or ``None`` for
:class:`~arcade.stages.DefaultOptimizationStage`.
hardware_stage: Hardware stage instance, or ``None`` for
:class:`~arcade.stages.DefaultHardwareStage`.
visualization_stage: Visualization stage instance, or ``None`` for
:class:`~arcade.stages.DefaultVisualizationStage`.
"""
from .stages import (
DefaultClassifyStage,
DefaultDataStage,
DefaultHardwareStage,
DefaultOptimizationStage,
DefaultVisualizationStage,
)
self.data_stage = data_stage or DefaultDataStage()
self.classify_stage = classify_stage or DefaultClassifyStage()
self.optimization_stage = optimization_stage or DefaultOptimizationStage()
self.hardware_stage = hardware_stage or DefaultHardwareStage()
self.visualization_stage = visualization_stage or DefaultVisualizationStage()
[docs]
def run(
self,
config_path: str | Path | dict,
*,
cache_dir: str | Path | None = None,
demod_cache_path: str | Path | None = None,
) -> "PipelineResults":
"""Execute the full pipeline from config through visualization (and QEC).
Args:
config_path: Config file path, dict, or
:class:`~arcade.config.ArcadeConfig`.
cache_dir: Optional root for demod / spectral / transition caches.
demod_cache_path: Optional explicit demod-cache file path.
Returns:
:class:`~arcade.results.PipelineResults` with ``data``,
``classify``, ``hardware``, ``optimization``, ``qec``, and
``report_path``.
Raises:
RuntimeError: If the data or classify stage fails fatally.
"""
configure_logging()
cfg = _resolve_cfg(config_path)
log.info("ARCADE pipeline starting")
import time as _time
t0 = _time.perf_counter()
try:
data_out = self.data_stage.run(
cfg,
cache_dir=cache_dir,
demod_cache_path=demod_cache_path,
classifiers_run=cfg.classifiers.run,
)
except Exception as exc:
log.error("Data stage failed: %s", exc)
raise RuntimeError(
f"Data stage failed: {exc}\n"
"Check your data path, file format, and label configuration."
) from exc
log.info("Data stage: %.1f s", _time.perf_counter() - t0)
out = self._run_post_data(cfg, data_out)
log.info("ARCADE pipeline finished")
return out
[docs]
def run_classify(
self,
config_path: str | Path | dict | Any,
data_out: dict[str, Any],
) -> "PipelineResults":
"""Run classify through visualization on an existing data-stage dict.
Args:
config_path: Config file path, dict, or
:class:`~arcade.config.ArcadeConfig`.
data_out: Previously produced data-stage output.
Returns:
:class:`~arcade.results.PipelineResults` for the post-data stages.
"""
configure_logging()
cfg = _resolve_cfg(config_path)
return self._run_post_data(cfg, data_out)
def _run_post_data(self, cfg: Any, data_out: dict[str, Any]) -> "PipelineResults":
import time as _time
from .results import PipelineResults
classify_out: dict[str, Any] = {}
hardware_out: dict[str, Any] = {}
qec_out: dict[str, Any] = {}
optimization_out: dict[str, Any] = {}
report_path = None
t0 = _time.perf_counter()
try:
classify_out = self.classify_stage.run(cfg, data_out)
except Exception as exc:
log.error("Classify stage failed: %s", exc)
raise RuntimeError(
f"Classify stage failed: {exc}\n"
"Check your classifier selection and training parameters."
) from exc
log.info("Classify stage: %.1f s", _time.perf_counter() - t0)
try:
classify_out_for_opt = {**classify_out, "_data_out": data_out}
optimization_out = self.optimization_stage.run(cfg, classify_out_for_opt)
classify_out = {**classify_out, "optimization": optimization_out}
except Exception as exc:
log.warning("Optimization stage failed (non-fatal): %s", exc)
t0 = _time.perf_counter()
try:
hardware_out = self.hardware_stage.run(cfg, classify_out)
except Exception as exc:
log.warning("Hardware stage failed (non-fatal): %s", exc)
hardware_out = {"reports": {}, "scalability": {}}
log.info("Hardware stage: %.1f s", _time.perf_counter() - t0)
try:
report_path = self.visualization_stage.run(
cfg, data_out, classify_out, hardware_out,
)
except Exception as exc:
log.warning("Visualization stage failed (non-fatal): %s", exc)
try:
if getattr(cfg.qec, "enabled", False):
from .qec.stage import run_qec_stage
qec_out = run_qec_stage(cfg, classify_out)
except Exception as exc:
log.warning("QEC stage failed (non-fatal): %s", exc)
return PipelineResults(
{
"data": data_out,
"classify": classify_out,
"hardware": hardware_out,
"optimization": optimization_out,
"qec": qec_out,
"report_path": report_path,
}
)
def _resolve_cfg(config_path: str | Path | dict | Any) -> Any:
from .config import ArcadeConfig, load_config
if isinstance(config_path, ArcadeConfig):
return config_path
return load_config(config_path)
[docs]
def run_classify_pipeline(
config_path: str | Path | dict | Any,
data_out: dict[str, Any],
) -> "PipelineResults":
"""Run classify (+ hardware/viz) on an existing data-stage dict.
Args:
config_path: Config file path, dict, or
:class:`~arcade.config.ArcadeConfig`.
data_out: Previously produced data-stage output.
Returns:
:class:`~arcade.results.PipelineResults` for the post-data stages.
"""
configure_logging()
cfg = _resolve_cfg(config_path)
return ArcadePipeline().run_classify(cfg, data_out)
[docs]
def run_pipeline(
config_path: str | Path | dict,
*,
cache_dir: str | Path | None = None,
demod_cache_path: str | Path | None = None,
) -> "PipelineResults":
"""Execute the full ARCADE pipeline.
Args:
config_path: Config file path or dict.
cache_dir: Optional root for demod / spectral / transition caches.
demod_cache_path: Optional explicit demod-cache file path.
Returns:
:class:`~arcade.results.PipelineResults` for the full run.
"""
return ArcadePipeline().run(
config_path,
cache_dir=cache_dir,
demod_cache_path=demod_cache_path,
)