steady_state

steady_state

Steady-state detection for energy measurement at thermal equilibrium.

Classes

SteadyStateConfig dataclass

SteadyStateConfig(
    warmup_samples: int = 5,
    window_size: int = 5,
    cv_threshold: float = 0.05,
    min_steady_samples: int = 3,
    metric: str = "throughput",
)

Configuration for steady-state detection.

SteadyStateResult dataclass

SteadyStateResult(
    total_samples: int = 0,
    warmup_samples: int = 0,
    steady_state_samples: int = 0,
    steady_state_reached: bool = False,
    warmup_throughputs: List[float] = list(),
    warmup_energies: List[float] = list(),
    steady_throughputs: List[float] = list(),
    steady_energies: List[float] = list(),
)

Result of steady-state detection.

SteadyStateDetector

SteadyStateDetector(
    config: SteadyStateConfig | None = None,
)

Detect steady state using coefficient of variation over a sliding window.

The first warmup_samples recordings are always classified as warmup. After warmup, the CV (stdev / mean) of the last window_size values is checked. When CV < cv_threshold for min_steady_samples consecutive checks, steady state is declared.

Source code in src/diapason/telemetry/steady_state.py

def __init__(self, config: SteadyStateConfig | None = None) -> None:
    self._config = config or SteadyStateConfig()
    self._throughputs: List[float] = []
    self._energies: List[float] = []
    self._consecutive_stable: int = 0
    self._steady_state_reached: bool = False

Attributes

result property

result: SteadyStateResult

Return a snapshot of the detection state.

Methods:

record

record(
    throughput: float,
    energy: float = 0.0,
    latency: float = 0.0,
) -> bool

Record a sample. Returns True when steady state is reached.

Source code in src/diapason/telemetry/steady_state.py

def record(
    self,
    throughput: float,
    energy: float = 0.0,
    latency: float = 0.0,
) -> bool:
    """Record a sample.  Returns ``True`` when steady state is reached."""
    self._throughputs.append(throughput)
    self._energies.append(energy)

    cfg = self._config

    # Still in warmup phase
    if len(self._throughputs) <= cfg.warmup_samples:
        return False

    # Already declared steady
    if self._steady_state_reached:
        return True

    # Not enough post-warmup samples for a full window yet
    post_warmup = self._throughputs[cfg.warmup_samples :]
    if len(post_warmup) < cfg.window_size:
        return False

    # Compute CV over the last window_size values
    window = post_warmup[-cfg.window_size :]
    mean = statistics.mean(window)
    if mean == 0:
        self._consecutive_stable = 0
        return False

    cv = statistics.stdev(window) / mean if len(window) > 1 else 0.0

    if cv < cfg.cv_threshold:
        self._consecutive_stable += 1
    else:
        self._consecutive_stable = 0

    if self._consecutive_stable >= cfg.min_steady_samples:
        self._steady_state_reached = True
        return True

    return False

reset

reset() -> None

Clear all recorded state.

Source code in src/diapason/telemetry/steady_state.py

def reset(self) -> None:
    """Clear all recorded state."""
    self._throughputs.clear()
    self._energies.clear()
    self._consecutive_stable = 0
    self._steady_state_reached = False