From 394a805245b06d559416145449f8690140c54d3a Mon Sep 17 00:00:00 2001 From: Joe Amditis <6799804+jamditis@users.noreply.github.com> Date: Mon, 27 Jul 2026 10:39:57 -0400 Subject: [PATCH 1/5] Warn when the capture device cannot carry continuous speech A Bluetooth conference mic on Windows connects over the hands-free profile, which downsamples to narrowband and drops frames. faster-whisper then returns garbled text. The same mic over USB Audio Class transcribes cleanly, so this is the transport, not the model. What made it cost an afternoon is that wake detection does not degrade with it. A live "computah ..." still scores 0.998, because the wake model matches a short fixed pattern that survives the damage while a full sentence does not. The loop looks healthy right up to the transcript, so the mic is the last thing suspected. Hence a warning at startup rather than only a README note. Two signals: a hands-free marker in the device name, and a native rate below the pipeline's 16 kHz. The rate read is the device's advertised rate, not the one the stream opened at -- _open() asks for 16 kHz first and WASAPI auto_convert grants it even on a narrowband device, so the opened rate reports 16000 while the audio is already ruined. Checking that field would have detected nothing. The classifier is its own module rather than part of audio.py because audio.py imports sounddevice, which loads native PortAudio and is absent on plenty of boxes. Judging a device's advertised properties needs no hardware, so keeping it separate makes it testable everywhere, matching the hardware-free-core split the project already draws. Also surfaced where someone choosing a mic will look: --list tags unsuitable inputs, and --test-mic now exits 3 instead of reporting a clean pass on a device it just warned about. A diagnostic that returns success on hardware it told you not to use is how this stayed confusing. Known blind spot, documented rather than papered over: wideband HFP negotiates 16 kHz and clears the rate check, and on Linux the name marker is usually absent, so a clean result means "nothing detected", not "device is fine". wake-20260727T1000-0b9a6b --- .github/workflows/ci.yml | 1 + CHANGELOG.md | 13 ++ CLAUDE.md | 2 + README.md | 43 ++++++ audio.py | 53 +++++++- capture_quality.py | 96 ++++++++++++++ pipeline.py | 5 + test_capture_quality.py | 266 ++++++++++++++++++++++++++++++++++++++ test_run_loop_playback.py | 3 + test_warm_models.py | 2 + 10 files changed, 480 insertions(+), 4 deletions(-) create mode 100644 capture_quality.py create mode 100644 test_capture_quality.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 3c11fee..93e2bbf 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -64,6 +64,7 @@ jobs: test_brain_bridge.py \ test_brain_dispatch.py \ test_sanitize_reply.py \ + test_capture_quality.py \ test_chime.py \ test_config_validation.py \ test_confidence_guard.py \ diff --git a/CHANGELOG.md b/CHANGELOG.md index 219974b..d10b6e6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,19 @@ All notable changes to computah are recorded here. The format follows ## [Unreleased] ### Added +- Capture-device suitability warning (#34): `capture_quality.py` judges whether a + mic can carry continuous speech, and the live loop says so at startup instead of + letting it surface as a garbled transcript. A Bluetooth hands-free (HFP) mic on + Windows downsamples to narrowband and drops frames, so faster-whisper returns + garbled text while wake detection keeps firing at 0.998 -- the loop looks + healthy right up to the transcript. Two signals: a hands-free marker in the + device name, and a native rate below the pipeline's 16 kHz. The check reads the + device's advertised rate rather than the opened rate, because WASAPI + `auto_convert` grants a 16 kHz open on a narrowband device and hides the + problem. `audio.py --list` tags unsuitable inputs, `--test-mic` exits 3 on one + (rather than reporting a clean pass, as it used to), and the module is + hardware-free so `test_capture_quality.py` runs with no PortAudio. Known blind + spot: wideband HFP at 16 kHz with no name marker, which needs a spectral check. - `prep_wake_samples.py --clean` (#8): removes leftover clips of a take the run just re-recorded (a `_NNN.wav` clip whose stem this run wrote) from `--output`, so re-recording with fewer utterances no longer strands orphaned diff --git a/CLAUDE.md b/CLAUDE.md index 238841e..052af68 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -138,6 +138,7 @@ python -m venv .venv && .venv/bin/pip install -r requirements.txt .venv/bin/python test_confidence_guard.py # mishear guard decision + aggregation — fast, no models .venv/bin/python test_preroll.py # pre-roll buffer keeps a no-pause request's leading audio — fast, no models .venv/bin/python test_endpoint_config.py # endpoint_silence_ms / max_request_ms tune capture endpointing — fast, no models +.venv/bin/python test_capture_quality.py # capture-device suitability verdicts + the startup warning — fast, no models, no PortAudio .venv/bin/python test_chime.py # wake-acknowledgment cue generator + both-loop wiring — fast, no models .venv/bin/python test_live_driver.py # live_driver hardware path honors the guard — fast, no models .venv/bin/python test_pipeline_bridge.py # full chain + bridge brain (loads models) @@ -178,6 +179,7 @@ bug. - `pipeline.py` — stages, chain, CLI, and the live-streaming turn (`stream_detect_wake`, `capture_request`, `run_turn`). `run_turn` exposes `on_wake` (fires at the wake->capture boundary, for the chime) and `on_capture` (fires after capture, for half-duplex mic pause); `run_loop` is the desktop live loop and wires both. - `live_driver.py` — always-on live loop: real mic (arecord on stdin) -> wake -> chime -> STT -> brain -> spoken reply, re-arming after each turn. Gates the transcript through `pipeline.guard_transcript` before dispatch, the same mishear guard as `run_turn`. +- `capture_quality.py` — whether a capture device can carry continuous speech (issue #34): flags a Bluetooth hands-free (HFP) endpoint by name and any device whose native rate is below the pipeline's 16 kHz. Hardware-free (no sounddevice) so it is testable anywhere; `audio.py` feeds it what `sd.query_devices()` reports and stores the verdict on `Microphone.capture_risk`, which `run_loop` prints at startup and `--list`/`--test-mic` surface. Judges the device's *native* rate, not the opened one — WASAPI `auto_convert` grants a 16 kHz open on a narrowband device. - `chime.py` — the wake-acknowledgment cue (issue #41): a pure-DSP generator for the two-tone rising chime played the instant the wake fires, before capture. Backend-free (no sounddevice/aplay) so both live loops can render the cue and play it through their own output. Gated by the `wake_chime` config key (opt-in, default off — it regresses the no-pause case on a half-duplex device); half-duplex handling keeps the cue out of the captured request when it is on. - `brain_bridge.py` — bridge plus transports. - `sim_persona.py` — test stand-in for the assistant. diff --git a/README.md b/README.md index 8fea8dd..c6c36ab 100644 --- a/README.md +++ b/README.md @@ -138,6 +138,49 @@ Run the pipeline on a wav file: The output path receives the spoken reply. +## Choosing a microphone + +The live loop needs a capture device that delivers full-band audio. Use a USB +Audio Class mic (or another wired/USB mic). **Do not use a Bluetooth mic on +Windows for continuous speech**, including a conference mic like the Anker +PowerConf paired over Bluetooth. + +Windows captures from a Bluetooth mic over the hands-free profile (HFP), which +downsamples and recompresses the stream to narrowband and drops frames. +faster-whisper then receives degraded audio and returns garbled transcripts. The +same PowerConf over USB delivers its DSP-cleaned full-band stream intact and +transcribes cleanly. + +This does not look like an audio problem. Wake detection keeps working over HFP -- +a live "computah ..." still scores 0.998 -- because the wake model matches a short +fixed pattern that survives the degradation. Only the sentence after it comes back +wrong, so the loop appears healthy right up to the transcript. + +To check a device before relying on it: + +```bash +.venv/bin/python audio.py --list # flags unsuitable input devices +.venv/bin/python audio.py --test-mic "powerconf" # see the exit codes below +``` + +`--list` tags any input device it can tell is unsuitable and prints why. The live +loop prints the same warning at startup and keeps running, since the device is +still good enough for wake detection. + +`--test-mic` exit codes, which point at different fixes: + +| Code | Meaning | +| --- | --- | +| 0 | Usable: frames arrived and nothing looks wrong with the device. | +| 2 | Nothing arrived (only zeros). The mic is muted, unplugged, or the wrong source. | +| 3 | Frames arrived, but the device is unsuitable for speech. It works; pick a different one. | + +One case the check cannot see: wideband HFP negotiates 16 kHz, so it passes the +rate test, and if the OS reports it without a hands-free marker in the name there +is nothing to match on. On Linux that marker is usually absent, so the rate check +is the only guard there. A clean `--list` is not proof a Bluetooth link is fine. +Prefer USB. + ## Configuration `config.json` contains safe defaults that can be committed: diff --git a/audio.py b/audio.py index 5a9aa61..2deec5d 100644 --- a/audio.py +++ b/audio.py @@ -36,6 +36,8 @@ import soundfile as sf from scipy.signal import resample_poly +import capture_quality + # Must match pipeline.py: openWakeWord's frame is 80 ms at 16 kHz = 1280 samples. TARGET_SR = 16000 FRAME_SIZE = 1280 @@ -146,6 +148,13 @@ def __init__( self._open_sr = None self._open_ch = None self.device_label = None + # What the device itself advertises, kept separate from _open_sr. On + # Windows, _open() asks for 16 kHz first and WASAPI auto_convert grants + # it even for a narrowband device, so _open_sr reads a healthy 16000 + # while the audio is already degraded. native_sr is the field that shows + # the truth, and capture_risk is its verdict (issue #34). + self.native_sr = None + self.capture_risk = None def _callback(self, indata, frames, time_info, status): if status: @@ -186,6 +195,10 @@ def _open(self, idx: int, host: str): def __enter__(self): idx, info, host = find_device(self.name, "input") self.device_label = f"{info['name']} ({host})" + self.native_sr = int(info["default_samplerate"]) + self.capture_risk = capture_quality.assess_input_device( + info["name"], self.native_sr, self.target_sr + ) self._stream, self._open_sr, self._open_ch = self._open(idx, host) self._stream.start() return self @@ -280,7 +293,12 @@ def play_wav(path, name=None) -> None: def list_devices() -> None: - """Print the audio device table (index, I/O, host API, default rate).""" + """Print the audio device table (index, I/O, host API, default rate). + + Tags input devices that cannot carry continuous speech, so the table helps + pick a mic rather than only listing what exists (issue #34). + """ + flagged = [] for i, d in enumerate(sd.query_devices()): io = [] if d["max_input_channels"] > 0: @@ -288,10 +306,20 @@ def list_devices() -> None: if d["max_output_channels"] > 0: io.append(f"OUT x{d['max_output_channels']}") host = sd.query_hostapis(d["hostapi"])["name"] + risk = None + if d["max_input_channels"] > 0: + risk = capture_quality.assess_input_device( + d["name"], d["default_samplerate"], TARGET_SR + ) + tag = f" <-- {risk.kind}, not for speech" if risk else "" + if risk: + flagged.append((i, risk)) print( f"[{i:2d}] {d['name'][:44]:44s} {','.join(io):10s} " - f"{host:18s} @{int(d['default_samplerate'])}" + f"{host:18s} @{int(d['default_samplerate'])}{tag}" ) + for idx, risk in flagged: + print(f"\n[{idx}] {risk.message}") def _test_mic(name: str, seconds: float) -> int: @@ -304,9 +332,15 @@ def _test_mic(name: str, seconds: float) -> int: n_samples = 0 target = int(seconds * TARGET_SR / FRAME_SIZE) with Microphone(subs) as mic: + # native_sr alongside open_sr on purpose: when they differ, something is + # converting, and open_sr=16000 on its own has fooled this check before + # (issue #34). print( - f"device: {mic.device_label} open_sr={mic._open_sr} open_ch={mic._open_ch}" + f"device: {mic.device_label} native_sr={mic.native_sr} " + f"open_sr={mic._open_sr} open_ch={mic._open_ch}" ) + if mic.capture_risk is not None: + print(f"WARNING: {mic.capture_risk.message}") for frame in mic.frames(): assert frame.dtype == np.int16 and frame.shape == (FRAME_SIZE,), ( f"bad frame: dtype={frame.dtype} shape={frame.shape}" @@ -325,6 +359,15 @@ def _test_mic(name: str, seconds: float) -> int: if peak == 0: print("RESULT: dead stream (only zeros) - check the mic's source") return 2 + if mic.capture_risk is not None: + # The frames are structurally perfect and the content is still unusable, + # so a bare "correct for the pipeline" here would read as a pass on a mic + # that garbles every sentence. Non-zero exit: this is a real finding. + print( + "RESULT: live frame stream, but this device is not usable for " + f"continuous speech ({mic.capture_risk.kind}) - see the warning above" + ) + return 3 print("RESULT: live frame stream - shape and dtype correct for the pipeline") return 0 @@ -337,7 +380,9 @@ def _cli() -> int: metavar="NAME", nargs="?", const="", - help="capture from a mic (name substring; empty = default)", + help="capture from a mic (name substring; empty = default). exits 0 if " + "usable, 2 if nothing arrived, 3 if the device works but is unsuitable " + "for continuous speech", ) p.add_argument("--play", metavar="WAV", help="play a WAV through an output device") p.add_argument( diff --git a/capture_quality.py b/capture_quality.py new file mode 100644 index 0000000..890afe5 --- /dev/null +++ b/capture_quality.py @@ -0,0 +1,96 @@ +#!/usr/bin/env python3 +"""Judge whether a capture device can carry continuous speech (issue #34). + +Wake detection and transcription do not fail together. A Bluetooth hands-free +(HFP) mic on Windows still fires the wake word reliably -- a live "computah ..." +scored 0.998 -- while faster-whisper receives narrowband, recompressed, +frame-dropped audio and returns garbled text. The wake model matches a short +fixed pattern that survives the degradation; a full sentence does not. The loop +therefore looks healthy right up to the transcript, which is why this warns up +front. README's "Choosing a microphone" is the operator-facing version. + +Hardware-free by design. audio.py is the one OS-specific seam (it imports +sounddevice, which loads native PortAudio), but judging a device's advertised +properties is plain logic, so it lives here and is testable on a box with no +PortAudio and no microphone. +""" + +from __future__ import annotations + +from typing import NamedTuple + +# The pipeline consumes 16 kHz frames (audio.TARGET_SR). A device whose own +# native rate is below that cannot supply them: something upstream is +# upsampling, and the detail is already gone. Callers pass their own target so +# this module stays free of audio.py and its native dependency. +DEFAULT_TARGET_SR = 16000 + +# Substrings that identify a Bluetooth hands-free endpoint by name. Windows +# exposes a BT mic as a separate hands-free capture endpoint (the "Hands-Free AG +# Audio" family), so the profile shows up in the device name before any audio is +# read. +# +# Excludes the bare word "headset" on purpose: a USB headset is a good STT +# capture device, and matching it would warn about working hardware. +# +# Platform asymmetry worth knowing, because production runs on the Pi: these are +# Windows names. A bluez/PipeWire HFP source on Linux typically carries no +# hands-free substring, so this check never fires there and the native-rate check +# below is the only guard -- and that one misses wideband mSBC at 16 kHz. On +# Linux, treat a clean result as "nothing detected", not "device is fine". +_HFP_NAME_MARKERS = ("hands-free", "hands free", "handsfree", "hfp") + + +class CaptureRisk(NamedTuple): + """A reason a device is unsuitable for continuous-speech capture. + + `kind` is the stable machine-readable tag ("hands-free-profile", + "narrowband"); `message` is the human line a CLI prints. Callers key logic + off `kind` so the wording can change without breaking them. + """ + + kind: str + message: str + + +def assess_input_device( + name: str | None, + native_sr: float | int | None, + target_sr: int = DEFAULT_TARGET_SR, +) -> CaptureRisk | None: + """Return a CaptureRisk if this input device will garble continuous speech. + + The name check runs first, and wins when both signals fire, because the + profile is the thing to act on and the rate is its symptom. + + Returns None when nothing is detectably wrong, which is weaker than "this + device is good". Wideband HFP negotiates 16 kHz, so it clears the rate check, + and an OS that reports it without a hands-free marker leaves nothing to match + on. Catching that needs the audio itself (a spectral check for an empty top + octave), which costs a capture at startup. + + An unknown or unparseable `native_sr` is treated as no information rather + than a fault, so a device that advertises no rate is not warned about. + """ + lowered = (name or "").lower() + if any(marker in lowered for marker in _HFP_NAME_MARKERS): + return CaptureRisk( + "hands-free-profile", + f"{name!r} is a Bluetooth hands-free (HFP) endpoint: narrowband and " + "frame-dropped, so transcription comes back garbled even though wake " + "detection keeps firing. Use this mic over USB, or another wired mic.", + ) + + try: + rate = float(native_sr) if native_sr is not None else 0.0 + except (TypeError, ValueError): + rate = 0.0 + if 0.0 < rate < target_sr: + return CaptureRisk( + "narrowband", + f"{name!r} is narrowband: {int(rate)} Hz native, below the " + f"{target_sr} Hz the pipeline needs, so transcription will be " + "garbled. Use a USB Audio Class mic.", + ) + + return None diff --git a/pipeline.py b/pipeline.py index a5e0d65..09c2640 100644 --- a/pipeline.py +++ b/pipeline.py @@ -1643,6 +1643,11 @@ def run_loop( print(f"computah listening -- wake word: {wake_word}. Ctrl-C to stop.") with audio.Microphone(mic_name) as mic: print(f"mic: {mic.device_label}") + # Say it once, here, while someone is still watching the terminal: a bad + # capture device does not announce itself later, since only the transcript + # is wrong (issue #34). Warn and continue -- it still serves wake fine. + if mic.capture_risk is not None: + print(f" WARNING: {mic.capture_risk.message}") frames = mic.frames() paused = False diff --git a/test_capture_quality.py b/test_capture_quality.py new file mode 100644 index 0000000..e7d4bda --- /dev/null +++ b/test_capture_quality.py @@ -0,0 +1,266 @@ +#!/usr/bin/env python3 +"""Checks for the capture-device suitability warning (issue #34). + +Two halves: the classifier's verdicts, and the fact that a verdict reaches the +operator at `--listen` startup. The second half is what makes the feature real -- +a correct classifier nobody prints is worth nothing. + +Hardware-free by construction. capture_quality takes the fields a device +advertises rather than reading 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 capture_quality +import pipeline +from capture_quality import CaptureRisk, assess_input_device + +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 device whose own rate is below the pipeline's 16 kHz.""" + risk = assess_input_device("Some Telephony Mic", 8000) + 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_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 native_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) + 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}", + ) + + +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_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() + 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()) diff --git a/test_run_loop_playback.py b/test_run_loop_playback.py index ed5bd16..ce34e2e 100644 --- a/test_run_loop_playback.py +++ b/test_run_loop_playback.py @@ -34,6 +34,9 @@ class _FakeMic: ended (active() is False), so run_loop breaks right after the single stubbed turn.""" device_label = "fake-mic" + # Part of the real Microphone contract the live loop reads (issue #34). None + # means "nothing wrong with this device", which is what a fake should claim. + capture_risk = None def __enter__(self): return self diff --git a/test_warm_models.py b/test_warm_models.py index dd9d175..5ebc40f 100644 --- a/test_warm_models.py +++ b/test_warm_models.py @@ -150,6 +150,8 @@ def test_run_loop_warms_before_listening() -> None: class _FakeMic: device_label = "fake-mic" + # Part of the real Microphone contract the live loop reads (issue #34). + capture_risk = None def __enter__(self): # Opening the mic is the moment the loop begins listening. From ff3b4a67f7d0aaf36bb453e3ee3f7201c8eb5e76 Mon Sep 17 00:00:00 2001 From: Joe Amditis <6799804+jamditis@users.noreply.github.com> Date: Mon, 27 Jul 2026 11:45:17 -0400 Subject: [PATCH 2/5] Match the narrowband verdict's confidence to its evidence The rate check read default_samplerate and then asserted the device 'is narrowband' and told the operator to use a different mic. That field is what the device advertises, not a measured capability: under WASAPI it is the shared-mode capture format and the limit is real, but under ALSA or CoreAudio it can be a default the device exceeds, so the flat claim could send someone to replace a working mic. Reports the advertised figure and what it does and does not establish per host, and softens --test-mic's RESULT line and exit-3 help to read as the classifier's verdict rather than as ground truth. The hands-free branch keeps its certainty, since a name match identifies the device outright. Kept consulting the advertised rate rather than the opened one: a successful open at 16 kHz is what auto_convert fakes, which is the bug this feature exists to catch. Separating an advertised default from a hard limit needs a spectral check on real audio, and that costs a capture at startup. Detection unchanged, so the 23 checks still pass. --- audio.py | 16 ++++++++++------ capture_quality.py | 23 ++++++++++++++++++++--- 2 files changed, 30 insertions(+), 9 deletions(-) diff --git a/audio.py b/audio.py index 2deec5d..68a59a2 100644 --- a/audio.py +++ b/audio.py @@ -360,12 +360,16 @@ def _test_mic(name: str, seconds: float) -> int: print("RESULT: dead stream (only zeros) - check the mic's source") return 2 if mic.capture_risk is not None: - # The frames are structurally perfect and the content is still unusable, + # The frames are structurally perfect and the content is still suspect, # so a bare "correct for the pipeline" here would read as a pass on a mic - # that garbles every sentence. Non-zero exit: this is a real finding. + # that garbles every sentence. Non-zero exit, and worded as the + # classifier's verdict rather than as ground truth: the hands-free match + # identifies the device, but the rate check reads an advertised figure that + # is conclusive only on WASAPI (see capture_quality.assess_input_device). print( - "RESULT: live frame stream, but this device is not usable for " - f"continuous speech ({mic.capture_risk.kind}) - see the warning above" + "RESULT: live frame stream, but this device is flagged as unsuitable " + f"for continuous speech ({mic.capture_risk.kind}) - see the warning " + "above" ) return 3 print("RESULT: live frame stream - shape and dtype correct for the pipeline") @@ -381,8 +385,8 @@ def _cli() -> int: nargs="?", const="", help="capture from a mic (name substring; empty = default). exits 0 if " - "usable, 2 if nothing arrived, 3 if the device works but is unsuitable " - "for continuous speech", + "usable, 2 if nothing arrived, 3 if the device works but is flagged as " + "unsuitable for continuous speech", ) p.add_argument("--play", metavar="WAV", help="play a WAV through an output device") p.add_argument( diff --git a/capture_quality.py b/capture_quality.py index 890afe5..1d18e1e 100644 --- a/capture_quality.py +++ b/capture_quality.py @@ -71,6 +71,20 @@ def assess_input_device( An unknown or unparseable `native_sr` is treated as no information rather than a fault, so a device that advertises no rate is not warned about. + + `native_sr` is what the device advertises, not a measured capability, and how + much that is worth depends on the host API. Under WASAPI it is the shared-mode + mix format, which is the rate capture actually happens at, so a low value + there is the fault itself. Under ALSA or CoreAudio it can be a default the + device is able to exceed, which makes the rate branch advisory rather than + conclusive -- hence the hedged wording, and why this returns a risk to report + rather than a verdict to act on. + + Opening a stream at `target_sr` is not the counter-evidence it looks like: + WASAPI `auto_convert` grants that request by resampling, which is the whole + reason this consults the advertised rate instead of `Microphone._open_sr`. + Distinguishing an advertised default from a hard limit needs the audio itself, + which costs a capture at startup. """ lowered = (name or "").lower() if any(marker in lowered for marker in _HFP_NAME_MARKERS): @@ -88,9 +102,12 @@ def assess_input_device( if 0.0 < rate < target_sr: return CaptureRisk( "narrowband", - f"{name!r} is narrowband: {int(rate)} Hz native, below the " - f"{target_sr} Hz the pipeline needs, so transcription will be " - "garbled. Use a USB Audio Class mic.", + f"{name!r} advertises {int(rate)} Hz, below the {target_sr} Hz the " + "pipeline needs, so transcription is likely to come back garbled. On " + "Windows that figure is the shared-mode capture format and the limit " + "is real; on other hosts it can be a default the device exceeds. If " + "this is a Bluetooth mic, use it over USB; otherwise check the rate " + "in the OS sound settings.", ) return None From 01f169cbc75cadd65b6b782644a84f62c47ddb60 Mon Sep 17 00:00:00 2001 From: Joe Amditis <6799804+jamditis@users.noreply.github.com> Date: Mon, 27 Jul 2026 12:49:50 -0400 Subject: [PATCH 3/5] Detect narrowband capture from the audio, not just the advertisement The name and rate checks only see what a device chooses to report, which is enough on Windows and not enough anywhere else. A bluez/PipeWire HFP source on Linux exposes no hands-free marker and PipeWire resamples to 16 kHz and reports 16 kHz, so both existing signals read clean while the audio underneath came off an 8 kHz transport. That hole was documented in the module and left open. --test-mic already captures a few seconds, so judging that audio costs nothing extra. Upsampling cannot invent content it never received, so an 8 kHz source leaves the band above 4 kHz empty and the cliff at that edge is the evidence. Measuring the cliff rather than total high-band energy is the part that took work. Speech falls off steeply on its own, so a raw high-band fraction mostly measures the tilt: on these signals a dull voice reads 0.00007% and a sloppily resampled 8 kHz signal reads 0.005%, putting the genuine case on the wrong side of the narrowband one. Comparing the 1 kHz band above the edge against the 1 kHz below cancels the tilt and separates them by an order of magnitude in the right order. The threshold is measured rather than guessed, and _cliff_ratio is split out from the verdict so a test can assert on the margin instead of only the verdict. That test earned its place immediately: the first threshold sat under 2x from the narrowband ceiling and had to move. Refs #34. wake-20260727T1200-f460a6 --- README.md | 25 ++++-- audio.py | 26 ++++-- capture_quality.py | 158 ++++++++++++++++++++++++++++++++--- test_capture_quality.py | 177 ++++++++++++++++++++++++++++++++++++++-- 4 files changed, 351 insertions(+), 35 deletions(-) diff --git a/README.md b/README.md index c6c36ab..8e520ca 100644 --- a/README.md +++ b/README.md @@ -163,9 +163,16 @@ To check a device before relying on it: .venv/bin/python audio.py --test-mic "powerconf" # see the exit codes below ``` -`--list` tags any input device it can tell is unsuitable and prints why. The live -loop prints the same warning at startup and keeps running, since the device is -still good enough for wake detection. +`--list` tags any input device it can tell is unsuitable and prints why, judging +the name and rate the device advertises. The live loop prints the same warning at +startup and keeps running, since the device is still good enough for wake +detection. + +`--test-mic` goes further, because it has audio to look at: after capturing, it +inspects the spectrum for the cliff that upsampling leaves behind. Audio that +arrived at 16 kHz but started at 8 kHz carries almost nothing above 4 kHz, and no +amount of resampling puts it back. That is the check that catches a device whose +advertisement looks clean, which is the usual Linux case. `--test-mic` exit codes, which point at different fixes: @@ -175,11 +182,13 @@ still good enough for wake detection. | 2 | Nothing arrived (only zeros). The mic is muted, unplugged, or the wrong source. | | 3 | Frames arrived, but the device is unsuitable for speech. It works; pick a different one. | -One case the check cannot see: wideband HFP negotiates 16 kHz, so it passes the -rate test, and if the OS reports it without a hands-free marker in the name there -is nothing to match on. On Linux that marker is usually absent, so the rate check -is the only guard there. A clean `--list` is not proof a Bluetooth link is fine. -Prefer USB. +The two layers fail differently, so a clean result means less from `--list` than +from `--test-mic`. `--list` sees only what the device advertises: wideband HFP +negotiates 16 kHz and passes the rate test, and on Linux the hands-free marker is +usually absent from the name, leaving nothing to match on. `--test-mic` reads the +audio instead, which is why it catches those -- but it too has a blind spot, a +resampler noisy enough to fill the empty band (around -30 dB, loud enough to hear +as hiss). A clean `--list` is not proof a Bluetooth link is fine. Prefer USB. ## Configuration diff --git a/audio.py b/audio.py index 68a59a2..d37e3e9 100644 --- a/audio.py +++ b/audio.py @@ -330,6 +330,7 @@ def _test_mic(name: str, seconds: float) -> int: peak = 0 sumsq = 0.0 n_samples = 0 + captured = [] target = int(seconds * TARGET_SR / FRAME_SIZE) with Microphone(subs) as mic: # native_sr alongside open_sr on purpose: when they differ, something is @@ -346,6 +347,7 @@ def _test_mic(name: str, seconds: float) -> int: f"bad frame: dtype={frame.dtype} shape={frame.shape}" ) n_frames += 1 + captured.append(frame) peak = max(peak, int(np.max(np.abs(frame)))) sumsq += float(np.sum(frame.astype(np.float32) ** 2)) n_samples += frame.size @@ -359,17 +361,29 @@ def _test_mic(name: str, seconds: float) -> int: if peak == 0: print("RESULT: dead stream (only zeros) - check the mic's source") return 2 - if mic.capture_risk is not None: + + # The audio is already in hand, so judging it costs nothing extra -- and it is + # the only check that sees through a device that advertises well and resamples + # underneath, which is how a Linux bluez/PipeWire HFP source presents. + spectrum_risk = capture_quality.assess_capture_spectrum( + np.concatenate(captured), TARGET_SR + ) + if spectrum_risk is not None: + print(f"WARNING: {spectrum_risk.message}") + + risk = mic.capture_risk or spectrum_risk + if risk is not None: # The frames are structurally perfect and the content is still suspect, # so a bare "correct for the pipeline" here would read as a pass on a mic # that garbles every sentence. Non-zero exit, and worded as the - # classifier's verdict rather than as ground truth: the hands-free match - # identifies the device, but the rate check reads an advertised figure that - # is conclusive only on WASAPI (see capture_quality.assess_input_device). + # classifier's verdict rather than as ground truth, because the three + # signals differ in strength: the hands-free name identifies the device + # outright, the advertised rate is conclusive only on WASAPI, and the + # spectral cliff is measured from the audio but can be masked by a noisy + # resampler (see the two assess_* docstrings in capture_quality). print( "RESULT: live frame stream, but this device is flagged as unsuitable " - f"for continuous speech ({mic.capture_risk.kind}) - see the warning " - "above" + f"for continuous speech ({risk.kind}) - see the warning above" ) return 3 print("RESULT: live frame stream - shape and dtype correct for the pipeline") diff --git a/capture_quality.py b/capture_quality.py index 1d18e1e..9a7e878 100644 --- a/capture_quality.py +++ b/capture_quality.py @@ -9,15 +9,28 @@ therefore looks healthy right up to the transcript, which is why this warns up front. README's "Choosing a microphone" is the operator-facing version. +Two ways to judge a device, in increasing order of certainty: + +- `assess_input_device` reads what the device advertises -- its name and native + rate. Free, works before any audio is captured, and is what the `--listen` + startup warning uses. It only sees what the OS chooses to report. +- `assess_capture_spectrum` reads the captured audio itself. It cannot run until + something has recorded, so `--test-mic` (which already captures a few seconds) + is its home. It catches the cases the advertised properties hide: a bluez HFP + source on Linux exposes no hands-free name and PipeWire can report a healthy + 16 kHz while resampling 8 kHz audio underneath. + Hardware-free by design. audio.py is the one OS-specific seam (it imports -sounddevice, which loads native PortAudio), but judging a device's advertised -properties is plain logic, so it lives here and is testable on a box with no -PortAudio and no microphone. +sounddevice, which loads native PortAudio), but neither judging advertised +properties nor analyzing an array of samples needs a device, so both live here +and are testable on a box with no PortAudio and no microphone. """ from __future__ import annotations -from typing import NamedTuple +from typing import NamedTuple, Sequence + +import numpy as np # The pipeline consumes 16 kHz frames (audio.TARGET_SR). A device whose own # native rate is below that cannot supply them: something upstream is @@ -35,9 +48,10 @@ # # Platform asymmetry worth knowing, because production runs on the Pi: these are # Windows names. A bluez/PipeWire HFP source on Linux typically carries no -# hands-free substring, so this check never fires there and the native-rate check -# below is the only guard -- and that one misses wideband mSBC at 16 kHz. On -# Linux, treat a clean result as "nothing detected", not "device is fine". +# hands-free substring, so this check never fires there, and PipeWire's resampling +# can hide the low rate from the native-rate check too. On Linux, treat a clean +# result here as "nothing detected", not "device is fine", and run `--test-mic`, +# where assess_capture_spectrum judges the audio instead of the advertisement. _HFP_NAME_MARKERS = ("hands-free", "hands free", "handsfree", "hfp") @@ -45,8 +59,9 @@ class CaptureRisk(NamedTuple): """A reason a device is unsuitable for continuous-speech capture. `kind` is the stable machine-readable tag ("hands-free-profile", - "narrowband"); `message` is the human line a CLI prints. Callers key logic - off `kind` so the wording can change without breaking them. + "narrowband", "upsampled-narrowband"); `message` is the human line a CLI + prints. Callers key logic off `kind` so the wording can change without + breaking them. """ kind: str @@ -66,8 +81,8 @@ def assess_input_device( Returns None when nothing is detectably wrong, which is weaker than "this device is good". Wideband HFP negotiates 16 kHz, so it clears the rate check, and an OS that reports it without a hands-free marker leaves nothing to match - on. Catching that needs the audio itself (a spectral check for an empty top - octave), which costs a capture at startup. + on. Catching that needs the audio itself, which is what + assess_capture_spectrum does once something has recorded. An unknown or unparseable `native_sr` is treated as no information rather than a fault, so a device that advertises no rate is not warned about. @@ -83,8 +98,9 @@ def assess_input_device( Opening a stream at `target_sr` is not the counter-evidence it looks like: WASAPI `auto_convert` grants that request by resampling, which is the whole reason this consults the advertised rate instead of `Microphone._open_sr`. - Distinguishing an advertised default from a hard limit needs the audio itself, - which costs a capture at startup. + Distinguishing an advertised default from a hard limit needs the audio itself + -- assess_capture_spectrum, which `--test-mic` runs on what it already + captured. """ lowered = (name or "").lower() if any(marker in lowered for marker in _HFP_NAME_MARKERS): @@ -111,3 +127,119 @@ def assess_input_device( ) return None + + +# Width of the two comparison bands straddling the half-Nyquist edge, in Hz. Wide +# enough to average over formant structure, narrow enough that both bands sit in +# comparable spectral territory. +_CLIFF_BAND_HZ = 1000 + +# Below this, the energy just above the edge has effectively vanished relative to +# the energy just below it, which is a resampler's signature rather than a voice. +# +# Measured on synthesized signals (the numbers the test pins): genuine full-band +# material runs from 0.34 (a very dull, heavily tilted voice) through 0.63 +# (speech-like) to 1.01 (white noise), while 8 kHz-sourced audio runs from 0.000 +# (an ideal resampler) up to 0.025 (a sloppy one with a -40 dB noise floor). +# This sits near the geometric middle of that gap: 3.4x under the genuine floor +# and 3.9x over the narrowband ceiling. Picking it by eye put it at 0.05, which is +# a hair under 2x over that ceiling -- close enough that the margin test caught it. +_CLIFF_RATIO_THRESHOLD = 0.10 + +# Shorter than this and the spectrum is too coarse to trust. +_MIN_ANALYSIS_SECONDS = 0.5 + + +def _cliff_ratio( + samples: Sequence[float] | np.ndarray, + sample_rate: int, + target_sr: int, +) -> float | None: + """Energy just above the half-Nyquist edge over energy just below it. + + Separate from the verdict so the measurement can be tested on its own: what + protects this check is the size of the gap between genuine and narrowband + audio, and a test can only assert on that if it can read the number. + + None means there was nothing to measure, which is not the same as 0.0 (a + real, total cliff). + """ + audio = np.asarray(samples, dtype=np.float64).ravel() + if audio.size < _MIN_ANALYSIS_SECONDS * sample_rate: + return None + + audio = audio - audio.mean() + if not np.any(audio): + return None + + edge_hz = target_sr / 4 + if sample_rate / 2 <= edge_hz: + # The capture cannot represent the band we would inspect, so there is + # nothing to measure. The advertised-rate check already covers this. + return None + + power = np.abs(np.fft.rfft(audio)) ** 2 + freqs = np.fft.rfftfreq(audio.size, 1.0 / sample_rate) + + def band_mean(low: float, high: float) -> float: + selected = (freqs >= low) & (freqs < high) + return float(power[selected].mean()) if selected.any() else 0.0 + + below = band_mean(edge_hz - _CLIFF_BAND_HZ, edge_hz) + if below <= 0: + return None + return band_mean(edge_hz, edge_hz + _CLIFF_BAND_HZ) / below + + +def assess_capture_spectrum( + samples: Sequence[float] | np.ndarray, + sample_rate: int, + target_sr: int = DEFAULT_TARGET_SR, +) -> CaptureRisk | None: + """Return a CaptureRisk if captured audio looks upsampled from a narrower band. + + This is the check that survives a lying device. `assess_input_device` can only + read what the OS advertises, and on Linux a bluez/PipeWire HFP source reports + neither a hands-free name nor a low rate -- PipeWire resamples to 16 kHz and + reports 16 kHz, so both signals look clean while the audio underneath came + from an 8 kHz transport. The evidence of that is in the audio: upsampling + cannot invent content it never received, so the band above the original + Nyquist is empty. + + It measures the *cliff* at half the target's Nyquist rather than the total + energy above it, because speech has a steep spectral tilt and a raw high-band + fraction mostly measures the tilt. On the test's own signals that metric does + not merely separate the two classes badly, it puts them in the wrong order: a + dull voice carries 0.00007% of its energy above 4 kHz while a sloppily + resampled 8 kHz signal carries 0.005%, so the genuine case looks *more* + narrowband than the narrowband one and no threshold exists to fit. Comparing + the 1 kHz band just above the edge against the 1 kHz band just below + normalizes the tilt away and puts the same cases an order of magnitude apart, + in the right order. + + Returns None when there is nothing to judge (too short, or digital silence) + rather than guessing. Note that near-silence is safe rather than a false + positive: room tone is broadband, so it produces a healthy ratio. + + Known limit: a resampler whose noise floor is high enough to fill the empty + band (around -30 dB, loud enough to hear as hiss) reads as genuine. Real + resamplers sit at -60 dB or better. + + Args: + samples: mono PCM, any amplitude scale -- the measure is a ratio. + sample_rate: the rate `samples` was captured at. + target_sr: the rate the pipeline needs, which sets the edge to inspect. + """ + edge_hz = target_sr / 4 + ratio = _cliff_ratio(samples, sample_rate, target_sr) + if ratio is None or ratio >= _CLIFF_RATIO_THRESHOLD: + return None + + return CaptureRisk( + "upsampled-narrowband", + f"the captured audio carries almost nothing above {int(edge_hz)} Hz " + f"(band ratio {ratio:.4f}), so it was upsampled from a narrower source " + f"even though it arrived at {sample_rate} Hz. That is what a Bluetooth " + "hands-free link does, and transcription comes back garbled. Use this " + "mic over USB, or another wired mic.", + ) diff --git a/test_capture_quality.py b/test_capture_quality.py index e7d4bda..b65db8e 100644 --- a/test_capture_quality.py +++ b/test_capture_quality.py @@ -1,14 +1,20 @@ #!/usr/bin/env python3 """Checks for the capture-device suitability warning (issue #34). -Two halves: the classifier's verdicts, and the fact that a verdict reaches the -operator at `--listen` startup. The second half is what makes the feature real -- -a correct classifier nobody prints is worth nothing. +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. -Hardware-free by construction. capture_quality takes the fields a device -advertises rather than reading 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. +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. @@ -21,9 +27,15 @@ import sys import types +import numpy as np + import capture_quality import pipeline -from capture_quality import CaptureRisk, assess_input_device +from capture_quality import ( + CaptureRisk, + assess_capture_spectrum, + assess_input_device, +) PASS, FAIL = "PASS", "FAIL" results: list[bool] = [] @@ -246,6 +258,150 @@ def test_startup_is_quiet_on_a_good_device() -> None: ) +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: + risk = assess_capture_spectrum(signal, SR) + 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: + risk = assess_capture_spectrum(signal, SR) + check( + f"silent for {label}", + risk is None, + "no warning" if risk is None else f"unexpected {risk.kind}", + ) + + +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: + risk = assess_capture_spectrum(signal, SR) + check( + f"no verdict for {label}", + risk is None, + "returned None" if risk is None else f"unexpected {risk.kind}", + ) + + +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) + 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() @@ -257,6 +413,11 @@ def main() -> int: 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_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 From 7b5c3f8cdedc0394203f66da68f2a86f9b2972e8 Mon Sep 17 00:00:00 2001 From: Joe Amditis <6799804+jamditis@users.noreply.github.com> Date: Mon, 27 Jul 2026 22:00:17 -0400 Subject: [PATCH 4/5] Keep capture checks honest when evidence is incomplete --- CHANGELOG.md | 17 ++-- CLAUDE.md | 2 +- README.md | 37 +++++--- audio.py | 68 +++++++------- capture_quality.py | 118 ++++++++++++++--------- test_capture_quality.py | 201 +++++++++++++++++++++++++++++++++++++--- 6 files changed, 331 insertions(+), 112 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d10b6e6..073a327 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,14 +12,15 @@ All notable changes to computah are recorded here. The format follows letting it surface as a garbled transcript. A Bluetooth hands-free (HFP) mic on Windows downsamples to narrowband and drops frames, so faster-whisper returns garbled text while wake detection keeps firing at 0.998 -- the loop looks - healthy right up to the transcript. Two signals: a hands-free marker in the - device name, and a native rate below the pipeline's 16 kHz. The check reads the - device's advertised rate rather than the opened rate, because WASAPI - `auto_convert` grants a 16 kHz open on a narrowband device and hides the - problem. `audio.py --list` tags unsuitable inputs, `--test-mic` exits 3 on one - (rather than reporting a clean pass, as it used to), and the module is - hardware-free so `test_capture_quality.py` runs with no PortAudio. Known blind - spot: wideband HFP at 16 kHz with no name marker, which needs a spectral check. + healthy right up to the transcript. Two property signals are a hands-free + marker in the device name and, on WASAPI only, a default rate below the + pipeline's 16 kHz. Other host APIs expose a preferred default rather than a + bandwidth limit. `audio.py --list` tags unsuitable inputs, and `--test-mic` + also flags the 4 kHz cliff left by an upsampled narrowband stream. It exits 3 + for a flagged device and 4 when the audio is unanalyzable or the narrowband + cliff is absent, since that check cannot rule out wideband HFP recompression + and frame drops. The module is hardware-free so `test_capture_quality.py` runs + with no PortAudio. - `prep_wake_samples.py --clean` (#8): removes leftover clips of a take the run just re-recorded (a `_NNN.wav` clip whose stem this run wrote) from `--output`, so re-recording with fewer utterances no longer strands orphaned diff --git a/CLAUDE.md b/CLAUDE.md index 052af68..94c1ae9 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -179,7 +179,7 @@ bug. - `pipeline.py` — stages, chain, CLI, and the live-streaming turn (`stream_detect_wake`, `capture_request`, `run_turn`). `run_turn` exposes `on_wake` (fires at the wake->capture boundary, for the chime) and `on_capture` (fires after capture, for half-duplex mic pause); `run_loop` is the desktop live loop and wires both. - `live_driver.py` — always-on live loop: real mic (arecord on stdin) -> wake -> chime -> STT -> brain -> spoken reply, re-arming after each turn. Gates the transcript through `pipeline.guard_transcript` before dispatch, the same mishear guard as `run_turn`. -- `capture_quality.py` — whether a capture device can carry continuous speech (issue #34): flags a Bluetooth hands-free (HFP) endpoint by name and any device whose native rate is below the pipeline's 16 kHz. Hardware-free (no sounddevice) so it is testable anywhere; `audio.py` feeds it what `sd.query_devices()` reports and stores the verdict on `Microphone.capture_risk`, which `run_loop` prints at startup and `--list`/`--test-mic` surface. Judges the device's *native* rate, not the opened one — WASAPI `auto_convert` grants a 16 kHz open on a narrowband device. +- `capture_quality.py` — whether a capture device can carry continuous speech (issue #34): flags a Bluetooth hands-free (HFP) endpoint by name, a low WASAPI shared-mode capture format, and the 4 kHz cliff left by upsampled narrowband audio. Hardware-free (no sounddevice) so it is testable anywhere; `audio.py` stores the advertised-property verdict on `Microphone.capture_risk`, which `run_loop` prints at startup and `--list`/`--test-mic` surface. A spectral result without that cliff stays inconclusive because it cannot rule out wideband HFP codec damage or frame drops. - `chime.py` — the wake-acknowledgment cue (issue #41): a pure-DSP generator for the two-tone rising chime played the instant the wake fires, before capture. Backend-free (no sounddevice/aplay) so both live loops can render the cue and play it through their own output. Gated by the `wake_chime` config key (opt-in, default off — it regresses the no-pause case on a half-duplex device); half-duplex handling keeps the cue out of the captured request when it is on. - `brain_bridge.py` — bridge plus transports. - `sim_persona.py` — test stand-in for the assistant. diff --git a/README.md b/README.md index 8e520ca..2bfcb22 100644 --- a/README.md +++ b/README.md @@ -163,32 +163,39 @@ To check a device before relying on it: .venv/bin/python audio.py --test-mic "powerconf" # see the exit codes below ``` -`--list` tags any input device it can tell is unsuitable and prints why, judging -the name and rate the device advertises. The live loop prints the same warning at -startup and keeps running, since the device is still good enough for wake -detection. +`--list` tags any input device it can tell is unsuitable and prints why. A +hands-free marker in the name works on every host. A low default rate is used only +for WASAPI, where it represents the shared-mode capture format; ALSA and CoreAudio +defaults are preferences and do not prove the device is narrowband. The live loop +prints the same warning at startup and keeps running, since the device is still +good enough for wake detection. `--test-mic` goes further, because it has audio to look at: after capturing, it inspects the spectrum for the cliff that upsampling leaves behind. Audio that arrived at 16 kHz but started at 8 kHz carries almost nothing above 4 kHz, and no -amount of resampling puts it back. That is the check that catches a device whose -advertisement looks clean, which is the usual Linux case. +amount of resampling puts it back. That catches the common Linux case where +PipeWire hides an 8 kHz transport behind a 16 kHz stream. + +The spectral check has a strict boundary: energy above 4 kHz rules out that one +upsampling signature, but it does not rule out wideband HFP. A wideband HFP codec +can carry energy above 4 kHz while recompression and frame drops still garble +speech. A capture without the narrowband cliff is therefore reported as +inconclusive, as is a capture that is too short, silent, or constant. `--test-mic` exit codes, which point at different fixes: | Code | Meaning | | --- | --- | -| 0 | Usable: frames arrived and nothing looks wrong with the device. | | 2 | Nothing arrived (only zeros). The mic is muted, unplugged, or the wrong source. | | 3 | Frames arrived, but the device is unsuitable for speech. It works; pick a different one. | - -The two layers fail differently, so a clean result means less from `--list` than -from `--test-mic`. `--list` sees only what the device advertises: wideband HFP -negotiates 16 kHz and passes the rate test, and on Linux the hands-free marker is -usually absent from the name, leaving nothing to match on. `--test-mic` reads the -audio instead, which is why it catches those -- but it too has a blind spot, a -resampler noisy enough to fill the empty band (around -30 dB, loud enough to hear -as hiss). A clean `--list` is not proof a Bluetooth link is fine. Prefer USB. +| 4 | Frames arrived, but the available checks cannot establish suitability. Prefer USB or verify with transcription. | + +The two layers fail differently. `--list` sees only what the device advertises: +wideband HFP negotiates 16 kHz and passes the rate test, and on Linux the +hands-free marker is usually absent from the name. `--test-mic` can flag an +upsampled narrowband stream from the audio, while leaving wideband HFP and a +resampler noisy enough to fill the empty band unresolved. A clean `--list` is not +proof a Bluetooth link is fine. Prefer USB. ## Configuration diff --git a/audio.py b/audio.py index d37e3e9..ca94f43 100644 --- a/audio.py +++ b/audio.py @@ -148,12 +148,12 @@ def __init__( self._open_sr = None self._open_ch = None self.device_label = None - # What the device itself advertises, kept separate from _open_sr. On - # Windows, _open() asks for 16 kHz first and WASAPI auto_convert grants - # it even for a narrowband device, so _open_sr reads a healthy 16000 - # while the audio is already degraded. native_sr is the field that shows - # the truth, and capture_risk is its verdict (issue #34). - self.native_sr = None + # PortAudio's default is kept separate from _open_sr because it is the + # WASAPI shared-mode capture format, while on other host APIs it is only + # a preference. capture_quality receives the host API so it can make that + # distinction instead of treating the default as native bandwidth. + self.device_default_sr = None + self.host_api = None self.capture_risk = None def _callback(self, indata, frames, time_info, status): @@ -163,17 +163,17 @@ def _callback(self, indata, frames, time_info, status): def _open(self, idx: int, host: str): """Open the input stream, preferring 16 kHz mono via the OS engine, with - native-rate fallbacks. Returns (stream, open_sr, open_channels).""" + default-rate fallbacks. Returns (stream, open_sr, open_channels).""" extra = _wasapi_autoconvert(host) info = sd.query_devices(idx) - native_sr = int(info["default_samplerate"]) - native_ch = max(1, int(info["max_input_channels"])) - # Order matters: clean engine-converted path first, then native fallbacks. + default_sr = int(info["default_samplerate"]) + max_ch = max(1, int(info["max_input_channels"])) + # Order matters: clean engine-converted path first, then device defaults. attempts = [ (self.target_sr, 1, extra), - (native_sr, 1, extra), - (native_sr, native_ch, extra), - (native_sr, native_ch, None), + (default_sr, 1, extra), + (default_sr, max_ch, extra), + (default_sr, max_ch, None), ] last = None for sr, ch, ex in attempts: @@ -195,9 +195,13 @@ def _open(self, idx: int, host: str): def __enter__(self): idx, info, host = find_device(self.name, "input") self.device_label = f"{info['name']} ({host})" - self.native_sr = int(info["default_samplerate"]) + self.device_default_sr = int(info["default_samplerate"]) + self.host_api = host self.capture_risk = capture_quality.assess_input_device( - info["name"], self.native_sr, self.target_sr + info["name"], + self.device_default_sr, + self.target_sr, + host_api=self.host_api, ) self._stream, self._open_sr, self._open_ch = self._open(idx, host) self._stream.start() @@ -309,7 +313,10 @@ def list_devices() -> None: risk = None if d["max_input_channels"] > 0: risk = capture_quality.assess_input_device( - d["name"], d["default_samplerate"], TARGET_SR + d["name"], + d["default_samplerate"], + TARGET_SR, + host_api=host, ) tag = f" <-- {risk.kind}, not for speech" if risk else "" if risk: @@ -333,11 +340,8 @@ def _test_mic(name: str, seconds: float) -> int: captured = [] target = int(seconds * TARGET_SR / FRAME_SIZE) with Microphone(subs) as mic: - # native_sr alongside open_sr on purpose: when they differ, something is - # converting, and open_sr=16000 on its own has fooled this check before - # (issue #34). print( - f"device: {mic.device_label} native_sr={mic.native_sr} " + f"device: {mic.device_label} default_sr={mic.device_default_sr} " f"open_sr={mic._open_sr} open_ch={mic._open_ch}" ) if mic.capture_risk is not None: @@ -362,16 +366,16 @@ def _test_mic(name: str, seconds: float) -> int: print("RESULT: dead stream (only zeros) - check the mic's source") return 2 - # The audio is already in hand, so judging it costs nothing extra -- and it is - # the only check that sees through a device that advertises well and resamples - # underneath, which is how a Linux bluez/PipeWire HFP source presents. - spectrum_risk = capture_quality.assess_capture_spectrum( + # The audio is already in hand, so judging it costs nothing extra. The cliff + # check sees an 8 kHz source that PipeWire advertises and emits at 16 kHz, + # while leaving wideband HFP explicitly inconclusive. + spectrum = capture_quality.assess_capture_spectrum( np.concatenate(captured), TARGET_SR ) - if spectrum_risk is not None: - print(f"WARNING: {spectrum_risk.message}") + if spectrum.risk is not None: + print(f"WARNING: {spectrum.risk.message}") - risk = mic.capture_risk or spectrum_risk + risk = mic.capture_risk or spectrum.risk if risk is not None: # The frames are structurally perfect and the content is still suspect, # so a bare "correct for the pipeline" here would read as a pass on a mic @@ -386,8 +390,8 @@ def _test_mic(name: str, seconds: float) -> int: f"for continuous speech ({risk.kind}) - see the warning above" ) return 3 - print("RESULT: live frame stream - shape and dtype correct for the pipeline") - return 0 + print(f"RESULT: inconclusive - {spectrum.message}") + return 4 def _cli() -> int: @@ -398,9 +402,9 @@ def _cli() -> int: metavar="NAME", nargs="?", const="", - help="capture from a mic (name substring; empty = default). exits 0 if " - "usable, 2 if nothing arrived, 3 if the device works but is flagged as " - "unsuitable for continuous speech", + help="capture from a mic (name substring; empty = default). exits 2 if " + "nothing arrived, 3 if flagged as unsuitable for continuous speech, or " + "4 if suitability remains inconclusive", ) p.add_argument("--play", metavar="WAV", help="play a WAV through an output device") p.add_argument( diff --git a/capture_quality.py b/capture_quality.py index 9a7e878..c3d09bf 100644 --- a/capture_quality.py +++ b/capture_quality.py @@ -9,16 +9,18 @@ therefore looks healthy right up to the transcript, which is why this warns up front. README's "Choosing a microphone" is the operator-facing version. -Two ways to judge a device, in increasing order of certainty: +Two evidence sources can identify a problem: -- `assess_input_device` reads what the device advertises -- its name and native - rate. Free, works before any audio is captured, and is what the `--listen` - startup warning uses. It only sees what the OS chooses to report. +- `assess_input_device` reads what the device advertises -- its name, host API, + and default rate. Free, works before any audio is captured, and is what the + `--listen` startup warning uses. It only sees what the OS chooses to report. - `assess_capture_spectrum` reads the captured audio itself. It cannot run until something has recorded, so `--test-mic` (which already captures a few seconds) is its home. It catches the cases the advertised properties hide: a bluez HFP source on Linux exposes no hands-free name and PipeWire can report a healthy - 16 kHz while resampling 8 kHz audio underneath. + 16 kHz while resampling 8 kHz audio underneath. Its 4 kHz cliff test cannot + rule out wideband HFP codec damage, so a capture without that cliff remains + inconclusive. Hardware-free by design. audio.py is the one OS-specific seam (it imports sounddevice, which loads native PortAudio), but neither judging advertised @@ -28,14 +30,12 @@ from __future__ import annotations -from typing import NamedTuple, Sequence +from typing import Literal, NamedTuple, Sequence import numpy as np -# The pipeline consumes 16 kHz frames (audio.TARGET_SR). A device whose own -# native rate is below that cannot supply them: something upstream is -# upsampling, and the detail is already gone. Callers pass their own target so -# this module stays free of audio.py and its native dependency. +# The pipeline consumes 16 kHz frames (audio.TARGET_SR). Callers pass their own +# target so this module stays free of audio.py and its native dependency. DEFAULT_TARGET_SR = 16000 # Substrings that identify a Bluetooth hands-free endpoint by name. Windows @@ -49,11 +49,18 @@ # Platform asymmetry worth knowing, because production runs on the Pi: these are # Windows names. A bluez/PipeWire HFP source on Linux typically carries no # hands-free substring, so this check never fires there, and PipeWire's resampling -# can hide the low rate from the native-rate check too. On Linux, treat a clean -# result here as "nothing detected", not "device is fine", and run `--test-mic`, -# where assess_capture_spectrum judges the audio instead of the advertisement. +# can hide the transport rate from the advertised-property check too. On Linux, +# treat a clean result here as "nothing detected", not "device is fine", and run +# `--test-mic`, where assess_capture_spectrum judges the audio instead of the +# advertisement. _HFP_NAME_MARKERS = ("hands-free", "hands free", "handsfree", "hfp") +# PortAudio's `default_samplerate` is generally a preference, not a bandwidth +# limit. WASAPI shared mode is the exception used here: its default is the mix +# format through which capture runs. Do not grow this list without verifying the +# host API's field has that meaning. +_CAPTURE_FORMAT_HOST_MARKERS = ("wasapi",) + class CaptureRisk(NamedTuple): """A reason a device is unsuitable for continuous-speech capture. @@ -68,10 +75,25 @@ class CaptureRisk(NamedTuple): message: str +class SpectrumAssessment(NamedTuple): + """The result of inspecting captured audio for a narrowband cliff. + + `status` is explicit because `None` used to mean both "no cliff" and "there + was nothing to analyze", and the CLI accidentally promoted both to success. + A non-flagged spectrum stays inconclusive: this measurement cannot detect + wideband HFP recompression or frame drops. + """ + + status: Literal["flagged", "inconclusive"] + risk: CaptureRisk | None + message: str + + def assess_input_device( name: str | None, - native_sr: float | int | None, + default_sr: float | int | None, target_sr: int = DEFAULT_TARGET_SR, + host_api: str | None = None, ) -> CaptureRisk | None: """Return a CaptureRisk if this input device will garble continuous speech. @@ -84,23 +106,19 @@ def assess_input_device( on. Catching that needs the audio itself, which is what assess_capture_spectrum does once something has recorded. - An unknown or unparseable `native_sr` is treated as no information rather + An unknown or unparseable `default_sr` is treated as no information rather than a fault, so a device that advertises no rate is not warned about. - `native_sr` is what the device advertises, not a measured capability, and how - much that is worth depends on the host API. Under WASAPI it is the shared-mode - mix format, which is the rate capture actually happens at, so a low value - there is the fault itself. Under ALSA or CoreAudio it can be a default the - device is able to exceed, which makes the rate branch advisory rather than - conclusive -- hence the hedged wording, and why this returns a risk to report - rather than a verdict to act on. + `default_sr` is not a measured capability or maximum supported rate. The rate + branch is therefore limited to host APIs where that field represents the + active capture format. WASAPI shared mode has that meaning; ALSA and CoreAudio + defaults do not, so a low value from either is ignored instead of warning + about a stream that may open at `target_sr`. Opening a stream at `target_sr` is not the counter-evidence it looks like: WASAPI `auto_convert` grants that request by resampling, which is the whole - reason this consults the advertised rate instead of `Microphone._open_sr`. - Distinguishing an advertised default from a hard limit needs the audio itself - -- assess_capture_spectrum, which `--test-mic` runs on what it already - captured. + reason the WASAPI check consults the shared-mode format instead of + `Microphone._open_sr`. """ lowered = (name or "").lower() if any(marker in lowered for marker in _HFP_NAME_MARKERS): @@ -112,18 +130,19 @@ def assess_input_device( ) try: - rate = float(native_sr) if native_sr is not None else 0.0 + rate = float(default_sr) if default_sr is not None else 0.0 except (TypeError, ValueError): rate = 0.0 - if 0.0 < rate < target_sr: + host_has_capture_format = any( + marker in (host_api or "").lower() for marker in _CAPTURE_FORMAT_HOST_MARKERS + ) + if host_has_capture_format and 0.0 < rate < target_sr: return CaptureRisk( "narrowband", - f"{name!r} advertises {int(rate)} Hz, below the {target_sr} Hz the " - "pipeline needs, so transcription is likely to come back garbled. On " - "Windows that figure is the shared-mode capture format and the limit " - "is real; on other hosts it can be a default the device exceeds. If " - "this is a Bluetooth mic, use it over USB; otherwise check the rate " - "in the OS sound settings.", + f"{name!r} uses a {int(rate)} Hz WASAPI shared-mode capture format, " + f"below the {target_sr} Hz the pipeline needs, so transcription is " + "likely to come back garbled. If this is a Bluetooth mic, use it over " + "USB; otherwise raise the rate in the Windows sound settings.", ) return None @@ -195,8 +214,8 @@ def assess_capture_spectrum( samples: Sequence[float] | np.ndarray, sample_rate: int, target_sr: int = DEFAULT_TARGET_SR, -) -> CaptureRisk | None: - """Return a CaptureRisk if captured audio looks upsampled from a narrower band. +) -> SpectrumAssessment: + """Assess whether captured audio looks upsampled from a narrower band. This is the check that survives a lying device. `assess_input_device` can only read what the OS advertises, and on Linux a bluez/PipeWire HFP source reports @@ -217,9 +236,12 @@ def assess_capture_spectrum( normalizes the tilt away and puts the same cases an order of magnitude apart, in the right order. - Returns None when there is nothing to judge (too short, or digital silence) - rather than guessing. Note that near-silence is safe rather than a false - positive: room tone is broadband, so it produces a healthy ratio. + Returns an explicit inconclusive result when there is nothing to judge (too + short, or digital silence) rather than guessing. A ratio above the threshold + is also inconclusive: it rules out an 8 kHz upsampling cliff, but wideband HFP + genuinely has energy above 4 kHz while its codec and frame drops can still + garble continuous speech. Note that near-silence is safe from a false + positive: room tone is broadband, so it produces a ratio above the threshold. Known limit: a resampler whose noise floor is high enough to fill the empty band (around -30 dB, loud enough to hear as hiss) reads as genuine. Real @@ -232,10 +254,21 @@ def assess_capture_spectrum( """ edge_hz = target_sr / 4 ratio = _cliff_ratio(samples, sample_rate, target_sr) - if ratio is None or ratio >= _CLIFF_RATIO_THRESHOLD: - return None + if ratio is None: + message = ( + "the capture did not contain enough varying audio for the spectral " + "check, so microphone suitability is inconclusive" + ) + return SpectrumAssessment("inconclusive", None, message) + if ratio >= _CLIFF_RATIO_THRESHOLD: + message = ( + f"no {int(edge_hz)} Hz upsampling cliff was detected, but that check " + "cannot detect wideband HFP codec damage or frame drops, so microphone " + "suitability is inconclusive" + ) + return SpectrumAssessment("inconclusive", None, message) - return CaptureRisk( + risk = CaptureRisk( "upsampled-narrowband", f"the captured audio carries almost nothing above {int(edge_hz)} Hz " f"(band ratio {ratio:.4f}), so it was upsampled from a narrower source " @@ -243,3 +276,4 @@ def assess_capture_spectrum( "hands-free link does, and transcription comes back garbled. Use this " "mic over USB, or another wired mic.", ) + return SpectrumAssessment("flagged", risk, risk.message) diff --git a/test_capture_quality.py b/test_capture_quality.py index b65db8e..43b1153 100644 --- a/test_capture_quality.py +++ b/test_capture_quality.py @@ -37,6 +37,19 @@ 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] = [] @@ -74,8 +87,8 @@ def test_hands_free_wins_over_rate() -> None: def test_flags_narrowband_rate() -> None: - """A device whose own rate is below the pipeline's 16 kHz.""" - risk = assess_input_device("Some Telephony Mic", 8000) + """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", @@ -88,6 +101,64 @@ def test_flags_narrowband_rate() -> None: ) +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_good_devices_are_silent() -> None: """No warning for the devices the project runs on. @@ -118,7 +189,7 @@ def test_unknown_rate_is_not_a_fault() -> None: for sr in (None, 0, 0.0, "", "not-a-number"): risk = assess_input_device("Mystery Device", sr) check( - f"silent for native_sr={sr!r}", + f"silent for default_sr={sr!r}", risk is None, "no warning" if risk is None else f"unexpected {risk.kind}", ) @@ -139,7 +210,9 @@ def test_target_rate_is_honored() -> None: 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) + 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", @@ -311,7 +384,8 @@ def test_spectrum_flags_upsampled_narrowband() -> None: ("CVSD narrowband (3.4 kHz)", _bandlimited(voice, 3400, floor_db=-50)), ] for label, signal in cases: - risk = assess_capture_spectrum(signal, SR) + assessment = assess_capture_spectrum(signal, SR) + risk = assessment.risk check( f"flags {label}", risk is not None and risk.kind == "upsampled-narrowband", @@ -334,11 +408,20 @@ def test_spectrum_is_silent_on_genuine_full_band_audio() -> None: ("mild 7 kHz lowpass", _bandlimited(_speechlike(), 7000, floor_db=-50)), ] for label, signal in cases: - risk = assess_capture_spectrum(signal, SR) + assessment = assess_capture_spectrum(signal, SR) check( - f"silent for {label}", - risk is None, - "no warning" if risk is None else f"unexpected {risk.kind}", + 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}", ) @@ -351,14 +434,99 @@ def test_spectrum_reports_no_information_rather_than_guessing() -> None: ("empty buffer", np.zeros(0)), ] for label, signal in cases: - risk = assess_capture_spectrum(signal, SR) + assessment = assess_capture_spectrum(signal, SR) check( - f"no verdict for {label}", - risk is None, - "returned None" if risk is None else f"unexpected {risk.kind}", + 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. @@ -394,7 +562,7 @@ def test_spectrum_threshold_keeps_its_measured_margin() -> None: 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 = assess_capture_spectrum(frames, SR).risk check( "int16 input is judged the same as float", risk is not None and risk.kind == "upsampled-narrowband", @@ -407,6 +575,8 @@ def main() -> int: 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_good_devices_are_silent() test_unknown_rate_is_not_a_fault() test_survives_a_missing_name() @@ -416,6 +586,9 @@ def main() -> int: 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) From e58b398ddfd2245fc46451ab15a7ceac8cb09f81 Mon Sep 17 00:00:00 2001 From: Joe Amditis <6799804+jamditis@users.noreply.github.com> Date: Mon, 27 Jul 2026 22:21:25 -0400 Subject: [PATCH 5/5] Warn on actual low-rate capture fallbacks --- audio.py | 8 ++++++++ test_capture_quality.py | 39 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 47 insertions(+) diff --git a/audio.py b/audio.py index ca94f43..1ea1e74 100644 --- a/audio.py +++ b/audio.py @@ -204,6 +204,14 @@ def __enter__(self): host_api=self.host_api, ) self._stream, self._open_sr, self._open_ch = self._open(idx, host) + if self.capture_risk is None and 0 < self._open_sr < self.target_sr: + self.capture_risk = capture_quality.CaptureRisk( + "narrowband", + f"{self.device_label} opened at {self._open_sr} Hz after the " + f"{self.target_sr} Hz attempt failed. The pipeline can resample " + "that stream, but it cannot restore the missing speech bandwidth; " + "choose an input or profile that opens at the target rate.", + ) self._stream.start() return self diff --git a/test_capture_quality.py b/test_capture_quality.py index 43b1153..8648f4b 100644 --- a/test_capture_quality.py +++ b/test_capture_quality.py @@ -159,6 +159,44 @@ def close(self): 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. @@ -577,6 +615,7 @@ def main() -> int: 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()