wake_eval.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""Threshold-sweep metrics for wake-word tuning (#17).
Pure decision core. Given the wake detector's scores on a labeled set, it reports
false-accepts per hour and false-rejects per activation across a `wake_threshold`
sweep and recommends a default under a false-accept budget. No audio, no model, no
numpy is imported here, so the math is unit-testable on any box; the scoring driver
(eval_wake_threshold.py) loads the recordings and supplies the scores.
Metric definitions (matching the issue's acceptance criteria):
- false-rejects per activation: the fraction of real activations the detector
misses at a given threshold. One positive recording is one activation.
- false-accepts per hour: spurious fires during non-wake audio, per hour of that
audio. Near-word recordings ("computer"/"commuter"/"computing") and ambient
clips are both non-wake audio: each is scored frame by frame and threshold
crossings are counted with a refractory gap, so one sustained loud moment is
a single false accept, the way a live assistant does not re-trigger on the
same noise. Dividing the fire count by the non-wake audio duration gives a rate
that is comparable across sample sets of different sizes.
The recommendation walks the sweep for the lowest false-reject threshold whose
false-accept rate stays within a caller-supplied budget, and never returns a
silently-worse default: if nothing meets the budget it says so and hands back the
lowest achievable false-accept rate instead.
"""
from __future__ import annotations
from math import ceil, inf
from typing import NamedTuple, Sequence
# Audio seconds per capture frame: openWakeWord scores 80 ms (1280-sample) frames
# at 16 kHz, matching pipeline.FRAME_SIZE. The driver passes the real value; this
# default keeps the core usable and the tests readable without importing pipeline.
DEFAULT_FRAME_HOP_S = 1280 / 16000 # 0.08
class Activation(NamedTuple):
"""One real wake-word activation (a positive recording).
peak is the detector's highest score over the clip (pipeline.detect_wake's
third return). frame_scores and first_score_at_s let the sweep find the first
crossing separately for every evaluated threshold. The latter is the time,
relative to the real clip's start, when frame_scores[0] becomes available; it
can be negative when the scoring stream includes leading context padding.
latency_s remains as a compatibility path for callers that measured one fixed
latency. When frame_scores is present it takes precedence because a crossing's
latency changes with the threshold.
"""
peak: float
latency_s: float | None = None
frame_scores: Sequence[float] | None = None
first_score_at_s: float = 0.0
class NonWakeClip(NamedTuple):
"""A continuous stretch of audio that must NOT fire the wake word.
frame_scores is the detector's per-frame score across the whole clip (scored
without stopping at the first fire, so any threshold can be applied afterward).
seconds is the clip's real duration, the denominator for the per-hour rate.
"""
frame_scores: Sequence[float]
seconds: float
class ThresholdRow(NamedTuple):
"""The measured behavior at one threshold."""
threshold: float
activations: int
false_rejects: int
false_rejects_per_activation: float
false_accepts: int
nonwake_hours: float
false_accepts_per_hour: float
class Recommendation(NamedTuple):
"""The chosen default threshold, the row it came from, and why."""
threshold: float
row: ThresholdRow
meets_budget: bool
reason: str
def threshold_grid(
start: float = 0.1, stop: float = 0.9, step: float = 0.05
) -> list[float]:
"""Inclusive threshold sweep [start, stop] in `step` increments.
Rounded to 4 decimals so floating-point accumulation does not leave a value
like 0.35000000000000003 in the table. The last point never exceeds stop: when
step does not divide the span evenly the grid stops at the last multiple within
range (threshold_grid(0.1, 0.83, 0.05) ends at 0.8, not 0.85) so the sweep never
evaluates a threshold above the one asked for. Raises on a non-positive step or
an inverted range so a typo fails loud instead of producing an empty or backwards
sweep.
"""
if step <= 0:
raise ValueError("step must be positive")
if stop < start:
raise ValueError("stop must be >= start")
# Floor the step count so the grid stays within [start, stop]; the epsilon keeps
# an even division (e.g. 16.0 arriving as 15.9999999) from dropping the endpoint.
n = int((stop - start) / step + 1e-9)
return [round(start + i * step, 4) for i in range(n + 1)]
def _activation_accepted(
a: Activation,
threshold: float,
latency_tolerance_s: float | None,
frame_hop_s: float,
) -> bool:
"""True when this activation fires within the peak and (optional) latency budget.
Peak below threshold is always a miss. When a latency tolerance is set AND this
activation's latency was measured, a fire later than the tolerance is a miss too
(a wake that arrives after the user has given up is a reject in practice).
Frame scores are preferred so the first crossing is recomputed for this exact
threshold. An unmeasured latency is judged on peak alone rather than penalized,
so peak-only runs are unaffected by the latency feature.
"""
if a.peak < threshold:
return False
if latency_tolerance_s is None:
return True
if a.frame_scores is not None:
crossing = next(
(i for i, score in enumerate(a.frame_scores) if score >= threshold), None
)
if crossing is None:
return False
latency_s = max(0.0, a.first_score_at_s + crossing * frame_hop_s)
return latency_s <= latency_tolerance_s
if a.latency_s is not None:
return a.latency_s <= latency_tolerance_s
return True
def count_false_accepts(
frame_scores: Sequence[float], threshold: float, refractory_frames: int
) -> int:
"""Count refractory-gapped threshold crossings in one non-wake clip.
A live detector that fires does not immediately re-arm on the same noise, so a
sustained loud region should count once, not once per frame. After a fire at
frame i, further fires are suppressed until frame i + refractory_frames.
"""
fires = 0
next_allowed = 0
for i, score in enumerate(frame_scores):
if i >= next_allowed and score >= threshold:
fires += 1
next_allowed = i + refractory_frames
return fires
def sweep(
positives: Sequence[Activation],
nonwake: Sequence[NonWakeClip],
thresholds: Sequence[float],
*,
refractory_s: float = 1.0,
frame_hop_s: float = DEFAULT_FRAME_HOP_S,
latency_tolerance_s: float | None = None,
) -> list[ThresholdRow]:
"""Measure false-reject and false-accept behavior at each threshold.
Scores are computed once by the caller; this only compares them against each
threshold, so the sweep is cheap no matter how many thresholds it covers.
"""
if frame_hop_s <= 0:
raise ValueError("frame_hop_s must be positive")
if refractory_s < 0:
raise ValueError("refractory_s must be non-negative")
if latency_tolerance_s is not None and latency_tolerance_s < 0:
raise ValueError("latency_tolerance_s must be non-negative")
# 0 s means no suppression (every crossing counts); a fraction of a frame still
# rounds up to a 1-frame gap so a small positive value is not silently a no-op.
refractory_frames = ceil(refractory_s / frame_hop_s)
total_hours = sum(c.seconds for c in nonwake) / 3600.0
n_act = len(positives)
rows: list[ThresholdRow] = []
for t in sorted(thresholds):
fr = sum(
1
for a in positives
if not _activation_accepted(a, t, latency_tolerance_s, frame_hop_s)
)
fa = sum(
count_false_accepts(c.frame_scores, t, refractory_frames) for c in nonwake
)
# With no non-wake audio the rate is undefined: report inf if anything
# fired (so it never looks safe) and 0.0 only when nothing did.
if total_hours > 0:
fa_per_hour = fa / total_hours
else:
fa_per_hour = inf if fa else 0.0
rows.append(
ThresholdRow(
threshold=t,
activations=n_act,
false_rejects=fr,
false_rejects_per_activation=(fr / n_act if n_act else 0.0),
false_accepts=fa,
nonwake_hours=total_hours,
false_accepts_per_hour=fa_per_hour,
)
)
return rows
def recommend(
rows: Sequence[ThresholdRow], *, max_false_accepts_per_hour: float
) -> Recommendation:
"""Pick the lowest false-reject threshold within the false-accept budget.
Among thresholds whose false-accept rate is within budget, choose the smallest
false-rejects-per-activation; ties break toward the lower threshold, which
leaves more margin for a quiet or clipped activation. If no threshold meets the
budget, return the threshold with the lowest false-accept rate and flag that the
budget was not met, so the recommendation is never silently worse than asked.
"""
if not rows:
raise ValueError("no rows to recommend from")
within = [r for r in rows if r.false_accepts_per_hour <= max_false_accepts_per_hour]
if within:
best = min(within, key=lambda r: (r.false_rejects_per_activation, r.threshold))
return Recommendation(
best.threshold,
best,
True,
f"lowest false-reject threshold within "
f"{max_false_accepts_per_hour:g} false accepts/hour",
)
fallback = min(
rows,
key=lambda r: (
r.false_accepts_per_hour,
r.false_rejects_per_activation,
r.threshold,
),
)
return Recommendation(
fallback.threshold,
fallback,
False,
f"no threshold met the {max_false_accepts_per_hour:g}/hour budget; "
f"this is the lowest achievable false-accept rate "
f"({fallback.false_accepts_per_hour:g}/hour)",
)