"""Hardware estimation and export stage."""
from __future__ import annotations
from pathlib import Path
from typing import Any
from .._logging import get_logger
from .base import BaseHardwareStage
from .helpers import data_out_labels_placeholder
log = get_logger(__name__)
[docs]
class DefaultHardwareStage(BaseHardwareStage):
"""FPGA resource estimation, export, and design-space exploration."""
[docs]
def run(self, cfg: Any, classify_out: dict) -> dict[str, Any]:
"""Estimate resources, optionally export HLS, and run DSE.
Args:
cfg: Pipeline config (``hardware`` block).
classify_out: Classify-stage output (``classifiers``, optional
``optimization``).
Returns:
Dict with ``reports``, ``scalability``, ``exports``, ``dse``,
and ``comparisons``. Empty collections when hardware is disabled.
"""
log.info("=== Stage 3: Hardware ===")
if not cfg.hardware.enabled:
log.info("Hardware estimation disabled")
return {"reports": {}, "scalability": {}, "exports": {}, "dse": {}, "comparisons": {}}
from ..hardware.estimator import estimate_resources
from ..hardware.export import export_hls4ml
from ..hardware.scalability import project_scalability
hw_cfg = cfg.hardware
clf_results = classify_out.get("classifiers", {})
opt_results = classify_out.get("optimization", {}).get("optimized", {})
reports: dict[str, Any] = {}
comparisons: dict[str, Any] = {}
scalability: dict[str, Any] = {}
exports: dict[str, Path] = {}
dse_out: dict[str, Any] = {}
dse_enabled = bool(getattr(hw_cfg.dse, "enabled", False) or hw_cfg.dse_enabled)
export_enabled = bool(hw_cfg.export_hls4ml or hw_cfg.run_synthesis)
hw_dict = hw_cfg.model_dump()
def _estimate_for(clf_name: str, data: dict[str, Any], suffix: str = "") -> None:
model = data.get("model")
if model is None:
return
key = f"{clf_name}{suffix}"
report = estimate_resources(model, hw_dict, classifier_name=key)
reports[key] = report
log.info(
" %s: LUT=%s, DSP=%s, latency=%s ns",
key,
report.lut,
report.dsp,
report.latency_ns,
)
for clf_name, data in clf_results.items():
_estimate_for(clf_name, data)
if clf_name in opt_results and opt_results[clf_name].get("model") is not None:
opt_data = {**data, "model": opt_results[clf_name]["model"]}
_estimate_for(clf_name, opt_data, suffix="_optimized")
comparisons[clf_name] = {
"original": reports.get(clf_name),
"optimized": reports.get(f"{clf_name}_optimized"),
"method": opt_results[clf_name].get("method"),
}
model_sources = {**clf_results}
for name, opt in opt_results.items():
if opt.get("model") is not None:
model_sources[name] = {**clf_results.get(name, {}), "model": opt["model"]}
for clf_name, data in model_sources.items():
model = data.get("model")
if model is None:
continue
if export_enabled:
try:
out = Path(hw_cfg.export_dir) / clf_name
exports[clf_name] = export_hls4ml(
model,
out,
backend=hw_cfg.backend,
clock_period=hw_cfg.clock_period_ns,
io_type=hw_cfg.io_type,
reuse_factor=hw_cfg.reuse_factor,
precision=hw_cfg.precision,
)
if hw_cfg.run_vivado:
try:
from dataclasses import replace
from ..hardware.report_parser import parse_synth_report
synth = parse_synth_report(exports[clf_name])
base = reports.get(clf_name) or estimate_resources(
model, hw_dict, classifier_name=clf_name,
)
reports[clf_name] = replace(
base,
lut=synth.get("lut", base.lut),
ff=synth.get("ff", base.ff),
bram=synth.get("bram", base.bram),
dsp=synth.get("dsp", base.dsp),
latency_cycles=synth.get("latency_cycles", base.latency_cycles),
latency_ns=synth.get("latency_ns", base.latency_ns),
estimated=False,
)
log.info("Vivado synthesis report parsed for %s", clf_name)
except FileNotFoundError:
log.warning(
"Vivado synthesis report not found for %s; using analytical estimates",
clf_name,
)
except Exception as exc:
log.warning("Vivado build/report failed for %s: %s", clf_name, exc)
except Exception as exc:
log.warning("HLS4ML export failed for %s: %s", clf_name, exc)
if dse_enabled:
bench = data.get("benchmark_features")
metrics = data.get("metrics")
if bench is not None and metrics is not None:
try:
from ..hardware.dse import explore_design_space
dse_cfg = hw_cfg.dse
dse_out[clf_name] = explore_design_space(
model,
bench,
data_out_labels_placeholder(bench),
bench,
data_out_labels_placeholder(bench),
reuse_factors=list(dse_cfg.reuse_factors),
precisions=list(dse_cfg.precisions),
io_types=list(dse_cfg.io_types),
target_fpga=hw_cfg.target_fpga,
clock_period_ns=hw_cfg.clock_period_ns,
)
except Exception as exc:
log.warning("DSE failed for %s: %s", clf_name, exc)
if reports:
qubit_counts = [1, 2, 5, 10, 20, 50]
for rname, rpt in reports.items():
if rname.endswith("_optimized"):
continue
scalability[rname] = project_scalability(
rpt,
qubit_counts,
base_num_qubits=cfg.data.num_qubits,
num_levels=cfg.data.num_levels,
)
return {
"reports": reports,
"scalability": scalability,
"exports": exports,
"dse": dse_out,
"comparisons": comparisons,
}
[docs]
def run_hardware_stage(cfg: Any, classify_out: dict) -> dict[str, Any]:
"""Run the default hardware stage.
Thin wrapper around :class:`DefaultHardwareStage`.
Args:
cfg: Pipeline config (``hardware`` block).
classify_out: Classify-stage output.
Returns:
Hardware-stage dict (``reports``, ``scalability``, …).
"""
return DefaultHardwareStage().run(cfg, classify_out)