"""FPGA Design-Space Exploration (DSE) for readout classifiers."""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any
import numpy as np
from .._logging import get_logger
log = get_logger(__name__)
[docs]
@dataclass
class DSEPoint:
"""One point in the design space."""
reuse_factor: int
precision: str
io_type: str
lut: int | None = None
ff: int | None = None
dsp: int | None = None
bram: int | None = None
latency_ns: float | None = None
accuracy: float | None = None
[docs]
@dataclass
class DSEResult:
"""Result of design-space exploration."""
points: list[DSEPoint] = field(default_factory=list)
pareto_optimal: list[DSEPoint] = field(default_factory=list)
best_accuracy: DSEPoint | None = None
best_resources: DSEPoint | None = None
best_balanced: DSEPoint | None = None
[docs]
def explore_design_space(
model: Any,
train_features: np.ndarray,
train_labels: np.ndarray,
test_features: np.ndarray,
test_labels: np.ndarray,
*,
reuse_factors: list[int] | None = None,
precisions: list[str] | None = None,
io_types: list[str] | None = None,
target_fpga: str = "xc7z020",
clock_period_ns: float = 5.0,
) -> DSEResult:
"""Sweep reuse_factor x precision x io_type and estimate resources.
For each combination, estimates FPGA resources analytically.
When HLS4ML is available, uses actual synthesis estimates.
"""
if reuse_factors is None:
reuse_factors = [1, 2, 4, 8, 16]
if precisions is None:
precisions = ["ap_fixed<16,6>", "ap_fixed<8,3>", "ap_fixed<4,2>"]
if io_types is None:
io_types = ["io_parallel", "io_stream"]
from .estimator import estimate_resources
points: list[DSEPoint] = []
for rf in reuse_factors:
for prec in precisions:
for io in io_types:
hw_config = {
"target_fpga": target_fpga,
"clock_period_ns": clock_period_ns,
"reuse_factor": rf,
"precision": prec,
"io_type": io,
}
report = estimate_resources(model, hw_config, classifier_name="dse")
bits = _parse_precision_bits(prec)
rf_scale = 1.0 / rf
bit_scale = bits / 16.0
adjusted_lut = int((report.lut or 0) * rf_scale * bit_scale) if report.lut else None
adjusted_dsp = int((report.dsp or 0) * rf_scale) if report.dsp else None
adjusted_lat = (report.latency_ns or 0) * rf if report.latency_ns else None
point = DSEPoint(
reuse_factor=rf,
precision=prec,
io_type=io,
lut=adjusted_lut,
ff=int((report.ff or 0) * bit_scale) if report.ff else None,
dsp=adjusted_dsp,
bram=report.bram,
latency_ns=adjusted_lat,
)
points.append(point)
pareto = _find_pareto(points)
best_acc = None
best_res = (
min(points, key=lambda p: (p.lut or float("inf")) + (p.dsp or 0) * 100) if points else None
)
best_lat = min(points, key=lambda p: p.latency_ns or float("inf")) if points else None
log.info("DSE: explored %d configurations, %d Pareto-optimal", len(points), len(pareto))
return DSEResult(
points=points,
pareto_optimal=pareto,
best_accuracy=best_acc,
best_resources=best_res,
best_balanced=best_lat or (pareto[len(pareto) // 2] if pareto else None),
)
def _parse_precision_bits(precision: str) -> int:
"""Extract total bits from ap_fixed<W,I> format."""
import re
m = re.match(r"ap_fixed<(\d+)", precision)
return int(m.group(1)) if m else 16
def _find_pareto(points: list[DSEPoint]) -> list[DSEPoint]:
"""Find Pareto-optimal points minimizing (LUT+DSP, latency)."""
if not points:
return []
scored = []
for p in points:
cost = (p.lut or 0) + (p.dsp or 0) * 100
lat = p.latency_ns or 0
scored.append((cost, lat, p))
scored.sort(key=lambda x: x[0])
pareto: list[DSEPoint] = []
best_lat = float("inf")
for cost, lat, point in scored:
if lat < best_lat:
pareto.append(point)
best_lat = lat
return pareto