eval_wake_threshold.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""Sweep wake_threshold against a labeled sample set and recommend a default (#17).
This is the scoring driver for the pure metrics core in wake_eval.py. It reads the
gitignored samples/ directories that prep_wake_samples.py writes, scores each clip
with the same openWakeWord model the live loop uses, and prints false-accepts per
hour and false-rejects per activation across a threshold sweep, plus a recommended
default under a false-accept budget.
Sample layout (all under samples/, which is gitignored: personal voice data):
- positive/ real wake-word activations -> false-reject rate
- negative/ hard near-words (computer/commuter) -> non-wake audio
- background/ ambient / no-speech recordings -> non-wake audio
Record and prepare them with prep_wake_samples.py (see docs/recording-computah.md).
The script is committed; the recordings are not.
Usage:
.venv/bin/python eval_wake_threshold.py # defaults under samples/
.venv/bin/python eval_wake_threshold.py --model computah \
--min-threshold 0.1 --max-threshold 0.9 --step 0.05 --max-fa-per-hour 1.0 \
--max-latency-s 1.0
Record the recommended value in config.json (wake_threshold) for the shipped model;
see docs/wake-threshold-tuning.md.
"""
from __future__ import annotations
import argparse
import sys
from pathlib import Path
import pipeline
import wake_eval
AUDIO_EXTS = {".wav", ".flac", ".ogg", ".m4a", ".mp3"}
FRAME_HOP_S = pipeline.FRAME_SIZE / 16000 # 0.08 s per scored frame
# After a fire the live loop does not re-arm immediately: stream_detect_wake stops on
# the crossing and capture_request consumes the request before wake detection resumes,
# at least _NO_SPEECH_ONSET_FRAMES (~1.6 s) when no speech follows and longer when it
# does. Using that floor as the default post-fire gap makes one sustained near-word or
# noise burst count as a single wake event, matching production, rather than several,
# which would inflate false-accepts/hour and over-tighten the threshold.
DEFAULT_REFRACTORY_S = pipeline._NO_SPEECH_ONSET_FRAMES * FRAME_HOP_S # ~1.6 s
def _clips(directory: Path) -> list[Path]:
"""Every audio file directly under `directory`, sorted; empty if it is absent."""
if not directory.is_dir():
return []
return sorted(p for p in directory.iterdir() if p.suffix.lower() in AUDIO_EXTS)
def _positive_activation(path: Path, model_name: str) -> wake_eval.Activation:
"""The detector's score trace for one activation clip.
The sweep keeps the trace, not only its peak, so it can measure the first
crossing independently at every threshold. wake_frame_scores prepends context
padding; first_score_at_s puts each score back on the real clip's timeline.
"""
scores = pipeline.wake_frame_scores(str(path), model_name)
return wake_eval.Activation(
peak=max(scores) if scores else 0.0,
frame_scores=scores,
first_score_at_s=FRAME_HOP_S - pipeline._WAKE_LEAD_SAMPLES / 16000,
)
def _frame_scores(path: Path, model_name: str) -> tuple[list[float], float]:
"""Per-frame wake scores across a whole clip, and its real duration in seconds.
Reuses pipeline.wake_frame_scores, the same padded, freshly-reset scoring
detect_wake runs, so non-wake clips get the identical filled context as positives
and their opening ~2 s is not under-scored (which would undercount false accepts).
Unlike detect_wake it keeps every frame's score, not only the peak, so any
threshold can be applied afterward. The duration is the real, un-padded length,
the honest denominator for the per-hour rate.
"""
scores = pipeline.wake_frame_scores(str(path), model_name)
seconds = len(pipeline._load_pcm16(str(path))) / 16000
return scores, seconds
def _print_table(rows: list[wake_eval.ThresholdRow]) -> None:
print(
f"\n{'thresh':>7} {'FR/act':>7} {'FA/hr':>8} {'rejects':>8} {'accepts':>8}"
)
print(" " + "-" * 44)
for r in rows:
print(
f"{r.threshold:>7.3f} {r.false_rejects_per_activation:>7.3f} "
f"{r.false_accepts_per_hour:>8.2f} "
f"{r.false_rejects:>3}/{r.activations:<4} {r.false_accepts:>8}"
)
def main() -> int:
p = argparse.ArgumentParser(
description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
)
p.add_argument(
"--samples",
default="samples",
help="root of the labeled sample directories (default: samples)",
)
p.add_argument(
"--model", default=None, help="wake model name (default: config.json wake_word)"
)
p.add_argument("--min-threshold", type=float, default=0.1)
p.add_argument("--max-threshold", type=float, default=0.9)
p.add_argument("--step", type=float, default=0.05)
p.add_argument(
"--max-fa-per-hour",
type=float,
default=1.0,
help="false-accept budget for the recommendation (default: 1.0)",
)
p.add_argument(
"--refractory-s",
type=float,
default=DEFAULT_REFRACTORY_S,
help="post-fire gap before a new false accept can count, seconds "
"(default: the live re-arm floor, ~1.6s)",
)
p.add_argument(
"--max-latency-s",
type=float,
default=None,
help="count a positive as rejected when its first threshold crossing arrives "
"later than this many seconds after the clip starts (default: disabled)",
)
args = p.parse_args()
model_name = args.model or pipeline.load_config()["wake_word"]
root = Path(args.samples)
pos_dir, neg_dir, bg_dir = root / "positive", root / "negative", root / "background"
positives_paths = _clips(pos_dir)
nonwake_paths = _clips(neg_dir) + _clips(bg_dir)
if not positives_paths or not nonwake_paths:
print(
"not enough labeled audio to evaluate.\n"
f" positive activations: {len(positives_paths)} under {pos_dir}\n"
f" non-wake clips: {len(nonwake_paths)} under "
f"{neg_dir} + {bg_dir}\n"
"Record and prep samples first (prep_wake_samples.py; see "
"docs/recording-computah.md), then re-run.",
file=sys.stderr,
)
return 1
print(f"model: {model_name}")
print(
f"scoring {len(positives_paths)} activations, "
f"{len(nonwake_paths)} non-wake clips ..."
)
positives = [_positive_activation(path, model_name) for path in positives_paths]
nonwake = []
for path in nonwake_paths:
scores, seconds = _frame_scores(path, model_name)
nonwake.append(wake_eval.NonWakeClip(frame_scores=scores, seconds=seconds))
thresholds = wake_eval.threshold_grid(
args.min_threshold, args.max_threshold, args.step
)
rows = wake_eval.sweep(
positives,
nonwake,
thresholds,
refractory_s=args.refractory_s,
frame_hop_s=FRAME_HOP_S,
latency_tolerance_s=args.max_latency_s,
)
_print_table(rows)
rec = wake_eval.recommend(rows, max_false_accepts_per_hour=args.max_fa_per_hour)
total_hours = rows[0].nonwake_hours if rows else 0.0
label = "recommended" if rec.meets_budget else "best available (budget not met)"
print(f"\nnon-wake audio evaluated: {total_hours * 60:.1f} min")
if args.max_latency_s is not None:
print(f"activation latency tolerance: {args.max_latency_s:.2f} s")
print(f"{label} wake_threshold for {model_name}: {rec.threshold:.3f}")
print(f" {rec.reason}")
print(
f" at this threshold: false-rejects "
f"{rec.row.false_rejects_per_activation:.1%} of activations, "
f"false-accepts {rec.row.false_accepts_per_hour:.2f}/hour"
)
if rec.meets_budget:
print(f'\nrecord this in config.json: "wake_threshold": {rec.threshold:.3f},')
else:
# No threshold met the budget, so rec.threshold is only the least-bad value and
# still violates it. Don't emit a paste-ready config line a failed run could be
# copied into config.json from; point at the two real fixes instead.
print(
"\nno threshold met the false-accept budget, so there is nothing to "
"record yet. Collect more or cleaner negatives and re-run, or re-run "
f"with a higher --max-fa-per-hour if {rec.row.false_accepts_per_hour:.2f} "
f"false-accepts/hour is acceptable (best available was {rec.threshold:.3f})."
)
print("see docs/wake-threshold-tuning.md.")
return 0
if __name__ == "__main__":
raise SystemExit(main())