test_capture_quality.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""Checks for the capture-device suitability warning (issue #34).
Three halves, if you will: the advertised-property verdicts, the spectral verdict
on captured audio, and the fact that a verdict reaches the operator at `--listen`
startup. That last one is what makes the feature real -- a correct classifier
nobody prints is worth nothing.
The spectral tests synthesize their own audio rather than shipping fixtures, so
the narrowband case is generated by band-limiting a voice-shaped signal the same
way a resampler does. That keeps the repo free of binary test assets and lets a
reader see exactly which property is being asserted.
Hardware-free by construction. capture_quality reads advertised fields and plain
sample arrays rather than a device, and the startup check stands a fake in for the
lazily-imported audio module, so everything here runs on a box with no PortAudio
and no microphone.
Run: .venv/bin/python test_capture_quality.py
Exit code is 0 only if every check passes.
"""
from __future__ import annotations
import contextlib
import io
import sys
import types
import numpy as np
import capture_quality
import pipeline
from capture_quality import (
CaptureRisk,
assess_capture_spectrum,
assess_input_device,
)
# audio.py only needs sounddevice for hardware calls, while these tests replace
# Microphone before exercising its capture verdict. Keep the test hardware-free
# on hosts where PortAudio is absent.
_saved_sounddevice = sys.modules.get("sounddevice")
sys.modules["sounddevice"] = types.ModuleType("sounddevice")
try:
import audio
finally:
if _saved_sounddevice is None:
sys.modules.pop("sounddevice", None)
else:
sys.modules["sounddevice"] = _saved_sounddevice
PASS, FAIL = "PASS", "FAIL"
results: list[bool] = []
def check(name: str, ok: bool, detail: str) -> bool:
results.append(bool(ok))
print(f" [{PASS if ok else FAIL}] {name}: {detail}")
return bool(ok)
def test_flags_bluetooth_hands_free_by_name() -> None:
"""The reported failure: a BT hands-free endpoint on Windows."""
# The real device name from issue #34's box, in the Windows form.
risk = assess_input_device("Headset (Anker PowerConf Hands-Free AG Audio)", 16000)
check(
"hands-free endpoint flagged",
risk is not None and risk.kind == "hands-free-profile",
f"kind={risk.kind if risk else None}",
)
check(
"hands-free message names the fix",
risk is not None and "USB" in risk.message,
"message points at USB Audio Class" if risk else "no message",
)
def test_hands_free_wins_over_rate() -> None:
"""Both signals present: report the profile, since that is the actionable one."""
risk = assess_input_device("Hands-Free AG Audio", 8000)
check(
"profile reported ahead of rate",
risk is not None and risk.kind == "hands-free-profile",
f"kind={risk.kind if risk else None}",
)
def test_flags_narrowband_rate() -> None:
"""A WASAPI capture format below the pipeline's 16 kHz."""
risk = assess_input_device("Some Telephony Mic", 8000, host_api="Windows WASAPI")
check(
"8 kHz device flagged",
risk is not None and risk.kind == "narrowband",
f"kind={risk.kind if risk else None}",
)
check(
"narrowband message states the measured rate",
risk is not None and "8000" in risk.message,
"message includes 8000 Hz" if risk else "no message",
)
def test_default_rate_is_advisory_outside_wasapi() -> None:
"""ALSA/CoreAudio defaults do not prove the opened stream is narrowband."""
for host_api in ("ALSA", "Core Audio", None):
risk = assess_input_device(
"Mic with a low preferred rate", 8000, host_api=host_api
)
check(
f"silent for {host_api or 'unknown'} default rate",
risk is None,
"no warning" if risk is None else f"unexpected {risk.kind}",
)
def test_microphone_passes_host_api_to_default_rate_check() -> None:
"""The live/test path preserves the host distinction from find_device."""
class FakeStream:
active = False
def start(self):
self.active = True
def stop(self):
self.active = False
def close(self):
pass
saved_find_device = audio.find_device
saved_open = audio.Microphone._open
info = {
"name": "Mic with a low preferred rate",
"default_samplerate": 8000,
"max_input_channels": 1,
}
audio.Microphone._open = lambda self, idx, host: (
FakeStream(),
self.target_sr,
1,
)
try:
for host_api, expected_kind in (
("ALSA", None),
("Windows WASAPI", "narrowband"),
):
audio.find_device = lambda name, kind, host=host_api: (0, info, host)
with audio.Microphone() as mic:
kind = mic.capture_risk.kind if mic.capture_risk else None
check(
f"microphone preserves {host_api}",
kind == expected_kind,
f"kind={kind}",
)
finally:
audio.find_device = saved_find_device
audio.Microphone._open = saved_open
def test_microphone_flags_an_actual_low_rate_fallback() -> None:
"""A concrete open below target is evidence even when the default was not."""
class FakeStream:
active = False
def start(self):
self.active = True
def stop(self):
self.active = False
def close(self):
pass
saved_find_device = audio.find_device
saved_open = audio.Microphone._open
info = {
"name": "Mic with an 8 kHz fallback",
"default_samplerate": 8000,
"max_input_channels": 1,
}
audio.find_device = lambda name, kind: (0, info, "ALSA")
audio.Microphone._open = lambda self, idx, host: (FakeStream(), 8000, 1)
try:
with audio.Microphone() as mic:
check(
"actual low-rate fallback is flagged",
mic.capture_risk is not None
and mic.capture_risk.kind == "narrowband"
and "8000" in mic.capture_risk.message,
f"risk={mic.capture_risk}",
)
finally:
audio.find_device = saved_find_device
audio.Microphone._open = saved_open
def test_good_devices_are_silent() -> None:
"""No warning for the devices the project runs on.
A warning here would be worse than no warning at all: the operator learns to
skip the line, and the real one gets skipped with it.
"""
cases = [
# The Pi production path, native 16 kHz mono over USB Audio Class. This is
# the boundary case -- 16 kHz must not read as narrowband.
("Anker PowerConf", 16000),
("Shure MV7", 48000),
("NVIDIA Broadcast", 48000),
("Microphone (USB Audio Device)", 44100),
# A USB headset is a fine STT mic, so "headset" alone must not trip it.
("Headset Microphone (Logitech USB Headset)", 48000),
]
for name, sr in cases:
risk = assess_input_device(name, sr)
check(
f"silent for {name}",
risk is None,
"no warning" if risk is None else f"unexpected {risk.kind}",
)
def test_unknown_rate_is_not_a_fault() -> None:
"""Absent rate information must not manufacture a warning."""
for sr in (None, 0, 0.0, "", "not-a-number"):
risk = assess_input_device("Mystery Device", sr)
check(
f"silent for default_sr={sr!r}",
risk is None,
"no warning" if risk is None else f"unexpected {risk.kind}",
)
def test_survives_a_missing_name() -> None:
"""A device with no usable name must not raise."""
try:
risk = assess_input_device(None, 48000)
check("no name does not raise", risk is None, "returned None")
except Exception as e: # noqa: BLE001 - any raise is the failure being checked
check("no name does not raise", False, f"raised {type(e).__name__}: {e}")
def test_target_rate_is_honored() -> None:
"""The threshold follows the caller's target, not a baked-in 16 kHz.
audio.py passes its TARGET_SR, so a pipeline retuned to a different frame rate
keeps a correct warning instead of one pinned to the old rate.
"""
risk = assess_input_device(
"Some Mic", 16000, target_sr=48000, host_api="Windows WASAPI"
)
check(
"16 kHz is narrowband against a 48 kHz target",
risk is not None and risk.kind == "narrowband",
f"kind={risk.kind if risk else None}",
)
check(
"default target is the pipeline's 16 kHz",
capture_quality.DEFAULT_TARGET_SR == 16000,
f"DEFAULT_TARGET_SR={capture_quality.DEFAULT_TARGET_SR}",
)
class _FakeMic:
"""audio.Microphone stand-in whose stream has already ended, so run_loop
reaches the startup print and then breaks immediately."""
device_label = "fake-mic"
def __init__(self, capture_risk):
self.capture_risk = capture_risk
def __enter__(self):
return self
def __exit__(self, *exc):
return False
def frames(self):
return iter(())
def pause(self):
pass
def resume(self):
pass
def flush(self):
pass
def active(self) -> bool:
return False
def _run_loop_output(capture_risk) -> str:
"""Drive pipeline.run_loop with a fake mic carrying `capture_risk`, and return
what it printed. run_loop imports `audio` lazily, so a fake module stands in
and no device, PortAudio, or model is touched."""
saved = {
"load_config": pipeline.load_config,
"warm_models": pipeline.warm_models,
"run_turn": pipeline.run_turn,
}
saved_audio = sys.modules.get("audio")
fake_audio = types.ModuleType("audio")
fake_audio.Microphone = lambda name=None: _FakeMic(capture_risk)
fake_audio.play_wav = lambda path, name=None: None
pipeline.load_config = lambda: {"wake_word": "computah", "wake_chime": False}
pipeline.warm_models = lambda cfg=None, wake_word=None: {}
pipeline.run_turn = lambda frames, **kw: None
sys.modules["audio"] = fake_audio
buf = io.StringIO()
try:
with contextlib.redirect_stdout(buf):
pipeline.run_loop()
finally:
pipeline.load_config = saved["load_config"]
pipeline.warm_models = saved["warm_models"]
pipeline.run_turn = saved["run_turn"]
if saved_audio is None:
sys.modules.pop("audio", None)
else:
sys.modules["audio"] = saved_audio
return buf.getvalue()
def test_startup_warns_about_a_bad_device() -> None:
"""The verdict has to reach the operator, not just exist.
This is the acceptance criterion from issue #34: warn at `--listen` startup.
A classifier that is right and silent leaves the original failure intact --
the loop looks healthy and only the transcript is wrong.
"""
risk = CaptureRisk("hands-free-profile", "narrowband and frame-dropped, use USB")
out = _run_loop_output(risk)
check(
"run_loop prints the warning at startup",
risk.message in out,
"message found" if risk.message in out else f"message absent from {out!r}",
)
check(
"the warning is labelled so it reads as a problem",
"WARNING" in out,
"labelled WARNING" if "WARNING" in out else "no WARNING label",
)
check(
"warned once, not per frame",
out.count(risk.message) == 1,
f"occurrences={out.count(risk.message)}",
)
check(
"the loop still starts on a flagged device",
"listening" in out,
"reached listening" if "listening" in out else f"never listened: {out!r}",
)
def test_startup_is_quiet_on_a_good_device() -> None:
"""No WARNING line when the device is fine, so the line stays meaningful."""
out = _run_loop_output(None)
check(
"no warning for a clean device",
"WARNING" not in out,
"no WARNING printed" if "WARNING" not in out else f"unexpected: {out!r}",
)
SR = 16000
def _speechlike(seconds: float = 3.0, tilt: float = 0.9, seed: int = 7):
"""Noise shaped like a voice: a natural spectral tilt plus formant bumps.
Real speech is the hard case for a spectral check, because its energy falls
off steeply with frequency all on its own. A flat signal would make any
threshold look good.
"""
rng = np.random.default_rng(seed)
n = int(seconds * SR)
spectrum = np.fft.rfft(rng.normal(0, 1, n))
freqs = np.fft.rfftfreq(n, 1.0 / SR)
shape = 1.0 / np.maximum(freqs, 50) ** tilt
for center, gain in ((500, 6), (1500, 4), (2600, 3), (5200, 1.6), (6800, 1.3)):
shape = shape * (1 + gain * np.exp(-((freqs - center) ** 2) / (2 * 320**2)))
return np.fft.irfft(spectrum * shape, n=n)
def _bandlimited(
signal, cutoff_hz: float, floor_db: float | None = None, seed: int = 3
):
"""Zero everything above cutoff, the way resampling from a narrower rate does.
`floor_db` adds the noise a real resampler leaves in the emptied band; None is
the ideal case that leaves it at exactly zero.
"""
spectrum = np.fft.rfft(signal)
freqs = np.fft.rfftfreq(len(signal), 1.0 / SR)
spectrum[freqs >= cutoff_hz] = 0
out = np.fft.irfft(spectrum, n=len(signal))
if floor_db is not None:
rng = np.random.default_rng(seed)
rms = float(np.sqrt((out**2).mean()))
out = out + rng.normal(0, rms * 10 ** (floor_db / 20), len(out))
return out
def test_spectrum_flags_upsampled_narrowband() -> None:
"""The Linux hole: audio that arrived at 16 kHz but started life at 8 kHz.
Neither the name nor the advertised rate can catch this -- PipeWire resamples
and reports 16 kHz -- so the audio is the only evidence left.
"""
voice = _speechlike()
cases = [
("ideal resampler", _bandlimited(voice, 4000)),
("-60 dB resampler floor", _bandlimited(voice, 4000, floor_db=-60)),
("-40 dB sloppy resampler", _bandlimited(voice, 4000, floor_db=-40)),
("CVSD narrowband (3.4 kHz)", _bandlimited(voice, 3400, floor_db=-50)),
]
for label, signal in cases:
assessment = assess_capture_spectrum(signal, SR)
risk = assessment.risk
check(
f"flags {label}",
risk is not None and risk.kind == "upsampled-narrowband",
f"kind={risk.kind if risk else None}",
)
def test_spectrum_is_silent_on_genuine_full_band_audio() -> None:
"""A false positive here tells someone to replace a mic that works.
The dull-voice case is the one that matters: it has very little energy up
high on its own, which is exactly what a naive high-band-energy check would
misread as narrowband.
"""
rng = np.random.default_rng(11)
cases = [
("speech-like full band", _speechlike()),
("very dull, heavily tilted voice", _speechlike(tilt=1.8)),
("white noise", rng.normal(0, 1, 3 * SR)),
("mild 7 kHz lowpass", _bandlimited(_speechlike(), 7000, floor_db=-50)),
]
for label, signal in cases:
assessment = assess_capture_spectrum(signal, SR)
check(
f"does not flag {label}",
assessment.risk is None,
(
"no warning"
if assessment.risk is None
else f"unexpected {assessment.risk.kind}"
),
)
check(
f"keeps {label} inconclusive",
assessment.status == "inconclusive",
f"status={assessment.status}",
)
def test_spectrum_reports_no_information_rather_than_guessing() -> None:
"""Nothing to measure must read as nothing to say, not as a fault."""
cases = [
("digital silence", np.zeros(3 * SR)),
("a constant offset", np.full(3 * SR, 1234.0)),
("too short to analyze", _speechlike(seconds=0.2)),
("empty buffer", np.zeros(0)),
]
for label, signal in cases:
assessment = assess_capture_spectrum(signal, SR)
check(
f"inconclusive verdict for {label}",
getattr(assessment, "status", None) == "inconclusive",
f"status={getattr(assessment, 'status', None)}",
)
def test_full_band_spectrum_does_not_clear_wideband_hfp() -> None:
"""A 4 kHz cliff check cannot rule out wideband HFP codec/drop artifacts."""
assessment = assess_capture_spectrum(_speechlike(), SR)
check(
"full-band spectrum remains inconclusive",
getattr(assessment, "status", None) == "inconclusive",
f"status={getattr(assessment, 'status', None)}",
)
message = getattr(assessment, "message", "")
check(
"inconclusive message names the undetected artifacts",
"wideband HFP" in message and "frame drops" in message,
f"message={message!r}",
)
def _run_mic_test(signal: np.ndarray, seconds: float) -> tuple[int, str]:
"""Run audio._test_mic against deterministic frames from `signal`."""
class FakeMicrophone:
device_label = "Unmarked PipeWire source (ALSA)"
device_default_sr = SR
host_api = "ALSA"
capture_risk = None
_open_sr = SR
_open_ch = 1
def __enter__(self):
return self
def __exit__(self, *exc):
return False
def frames(self):
usable = len(signal) - (len(signal) % audio.FRAME_SIZE)
for offset in range(0, usable, audio.FRAME_SIZE):
yield signal[offset : offset + audio.FRAME_SIZE].astype(np.int16)
saved_microphone = audio.Microphone
audio.Microphone = lambda name=None: FakeMicrophone()
output = io.StringIO()
try:
with contextlib.redirect_stdout(output):
exit_code = audio._test_mic("", seconds)
finally:
audio.Microphone = saved_microphone
return exit_code, output.getvalue()
def test_mic_returns_inconclusive_for_unanalyzable_audio() -> None:
"""A short or stuck nonzero stream must not receive exit code 0."""
cases = [
("too short", (_speechlike(seconds=0.2) * 8000), 0.2),
("constant nonzero", np.full(3 * SR, 1000), 3.0),
]
for label, signal, seconds in cases:
exit_code, output = _run_mic_test(signal.astype(np.int16), seconds)
check(
f"{label} capture exits inconclusive",
exit_code == 4,
f"exit_code={exit_code}",
)
check(
f"{label} capture reports inconclusive",
"inconclusive" in output.lower(),
f"output={output!r}",
)
def test_mic_does_not_pass_unmarked_wideband_hfp() -> None:
"""Full-band energy alone cannot make an unmarked PipeWire source usable."""
signal = (_speechlike() * 8000).astype(np.int16)
exit_code, output = _run_mic_test(signal, 3.0)
check(
"full-band unmarked source exits inconclusive",
exit_code == 4,
f"exit_code={exit_code}",
)
check(
"full-band unmarked source explains wideband HFP gap",
"wideband HFP" in output,
f"output={output!r}",
)
def test_spectrum_threshold_keeps_its_measured_margin() -> None:
"""Pin the separation the threshold was chosen from.
The docstring claims genuine audio sits well above the cutoff and narrowband
well below. If a future tweak narrows that gap, this fails while both verdict
tests above still pass -- the margin is the thing worth protecting, not just
the verdicts at today's values.
"""
# The dull voice, not the ordinary one: it is the genuine case with the least
# headroom, so it is the side of the threshold that gives out first. Measuring
# the comfortable case instead would leave the binding one unpinned.
genuine = capture_quality._cliff_ratio(
_speechlike(tilt=1.8), SR, capture_quality.DEFAULT_TARGET_SR
)
narrow = capture_quality._cliff_ratio(
_bandlimited(_speechlike(), 4000, floor_db=-40),
SR,
capture_quality.DEFAULT_TARGET_SR,
)
threshold = capture_quality._CLIFF_RATIO_THRESHOLD
check(
"genuine audio clears the threshold with margin",
genuine > threshold * 3,
f"ratio={genuine:.4f} vs threshold={threshold} (need >{threshold * 3:.3f})",
)
check(
"narrowband audio sits under it with margin",
narrow < threshold / 3,
f"ratio={narrow:.4f} vs threshold={threshold} (need <{threshold / 3:.3f})",
)
def test_spectrum_survives_int16_frames() -> None:
"""--test-mic hands it the concatenated int16 frames, not floats."""
frames = (_bandlimited(_speechlike(), 4000, floor_db=-60) * 8000).astype(np.int16)
risk = assess_capture_spectrum(frames, SR).risk
check(
"int16 input is judged the same as float",
risk is not None and risk.kind == "upsampled-narrowband",
f"kind={risk.kind if risk else None}",
)
def main() -> int:
print("capture-device suitability checks (issue #34)")
test_flags_bluetooth_hands_free_by_name()
test_hands_free_wins_over_rate()
test_flags_narrowband_rate()
test_default_rate_is_advisory_outside_wasapi()
test_microphone_passes_host_api_to_default_rate_check()
test_microphone_flags_an_actual_low_rate_fallback()
test_good_devices_are_silent()
test_unknown_rate_is_not_a_fault()
test_survives_a_missing_name()
test_target_rate_is_honored()
test_startup_warns_about_a_bad_device()
test_startup_is_quiet_on_a_good_device()
test_spectrum_flags_upsampled_narrowband()
test_spectrum_is_silent_on_genuine_full_band_audio()
test_spectrum_reports_no_information_rather_than_guessing()
test_full_band_spectrum_does_not_clear_wideband_hfp()
test_mic_returns_inconclusive_for_unanalyzable_audio()
test_mic_does_not_pass_unmarked_wideband_hfp()
test_spectrum_threshold_keeps_its_measured_margin()
test_spectrum_survives_int16_frames()
failed = results.count(False)
print(f"\n{len(results) - failed}/{len(results)} checks passed")
return 1 if failed else 0
if __name__ == "__main__":
raise SystemExit(main())