#!/usr/bin/env python3
"""Checks for the capture-device suitability warning (issue #34).

Three halves, if you will: the advertised-property verdicts, the spectral verdict
on captured audio, and the fact that a verdict reaches the operator at `--listen`
startup. That last one is what makes the feature real -- a correct classifier
nobody prints is worth nothing.

The spectral tests synthesize their own audio rather than shipping fixtures, so
the narrowband case is generated by band-limiting a voice-shaped signal the same
way a resampler does. That keeps the repo free of binary test assets and lets a
reader see exactly which property is being asserted.

Hardware-free by construction. capture_quality reads advertised fields and plain
sample arrays rather than a device, and the startup check stands a fake in for the
lazily-imported audio module, so everything here runs on a box with no PortAudio
and no microphone.

Run:  .venv/bin/python test_capture_quality.py
Exit code is 0 only if every check passes.
"""

from __future__ import annotations

import contextlib
import io
import sys
import types

import numpy as np

import capture_quality
import pipeline
from capture_quality import (
    CaptureRisk,
    assess_capture_spectrum,
    assess_input_device,
)

# audio.py only needs sounddevice for hardware calls, while these tests replace
# Microphone before exercising its capture verdict. Keep the test hardware-free
# on hosts where PortAudio is absent.
_saved_sounddevice = sys.modules.get("sounddevice")
sys.modules["sounddevice"] = types.ModuleType("sounddevice")
try:
    import audio
finally:
    if _saved_sounddevice is None:
        sys.modules.pop("sounddevice", None)
    else:
        sys.modules["sounddevice"] = _saved_sounddevice

PASS, FAIL = "PASS", "FAIL"
results: list[bool] = []


def check(name: str, ok: bool, detail: str) -> bool:
    results.append(bool(ok))
    print(f"  [{PASS if ok else FAIL}] {name}: {detail}")
    return bool(ok)


def test_flags_bluetooth_hands_free_by_name() -> None:
    """The reported failure: a BT hands-free endpoint on Windows."""
    # The real device name from issue #34's box, in the Windows form.
    risk = assess_input_device("Headset (Anker PowerConf Hands-Free AG Audio)", 16000)
    check(
        "hands-free endpoint flagged",
        risk is not None and risk.kind == "hands-free-profile",
        f"kind={risk.kind if risk else None}",
    )
    check(
        "hands-free message names the fix",
        risk is not None and "USB" in risk.message,
        "message points at USB Audio Class" if risk else "no message",
    )


def test_hands_free_wins_over_rate() -> None:
    """Both signals present: report the profile, since that is the actionable one."""
    risk = assess_input_device("Hands-Free AG Audio", 8000)
    check(
        "profile reported ahead of rate",
        risk is not None and risk.kind == "hands-free-profile",
        f"kind={risk.kind if risk else None}",
    )


def test_flags_narrowband_rate() -> None:
    """A WASAPI capture format below the pipeline's 16 kHz."""
    risk = assess_input_device("Some Telephony Mic", 8000, host_api="Windows WASAPI")
    check(
        "8 kHz device flagged",
        risk is not None and risk.kind == "narrowband",
        f"kind={risk.kind if risk else None}",
    )
    check(
        "narrowband message states the measured rate",
        risk is not None and "8000" in risk.message,
        "message includes 8000 Hz" if risk else "no message",
    )


def test_default_rate_is_advisory_outside_wasapi() -> None:
    """ALSA/CoreAudio defaults do not prove the opened stream is narrowband."""
    for host_api in ("ALSA", "Core Audio", None):
        risk = assess_input_device(
            "Mic with a low preferred rate", 8000, host_api=host_api
        )
        check(
            f"silent for {host_api or 'unknown'} default rate",
            risk is None,
            "no warning" if risk is None else f"unexpected {risk.kind}",
        )


def test_microphone_passes_host_api_to_default_rate_check() -> None:
    """The live/test path preserves the host distinction from find_device."""

    class FakeStream:
        active = False

        def start(self):
            self.active = True

        def stop(self):
            self.active = False

        def close(self):
            pass

    saved_find_device = audio.find_device
    saved_open = audio.Microphone._open
    info = {
        "name": "Mic with a low preferred rate",
        "default_samplerate": 8000,
        "max_input_channels": 1,
    }
    audio.Microphone._open = lambda self, idx, host: (
        FakeStream(),
        self.target_sr,
        1,
    )
    try:
        for host_api, expected_kind in (
            ("ALSA", None),
            ("Windows WASAPI", "narrowband"),
        ):
            audio.find_device = lambda name, kind, host=host_api: (0, info, host)
            with audio.Microphone() as mic:
                kind = mic.capture_risk.kind if mic.capture_risk else None
                check(
                    f"microphone preserves {host_api}",
                    kind == expected_kind,
                    f"kind={kind}",
                )
    finally:
        audio.find_device = saved_find_device
        audio.Microphone._open = saved_open


def test_microphone_flags_an_actual_low_rate_fallback() -> None:
    """A concrete open below target is evidence even when the default was not."""

    class FakeStream:
        active = False

        def start(self):
            self.active = True

        def stop(self):
            self.active = False

        def close(self):
            pass

    saved_find_device = audio.find_device
    saved_open = audio.Microphone._open
    info = {
        "name": "Mic with an 8 kHz fallback",
        "default_samplerate": 8000,
        "max_input_channels": 1,
    }
    audio.find_device = lambda name, kind: (0, info, "ALSA")
    audio.Microphone._open = lambda self, idx, host: (FakeStream(), 8000, 1)
    try:
        with audio.Microphone() as mic:
            check(
                "actual low-rate fallback is flagged",
                mic.capture_risk is not None
                and mic.capture_risk.kind == "narrowband"
                and "8000" in mic.capture_risk.message,
                f"risk={mic.capture_risk}",
            )
    finally:
        audio.find_device = saved_find_device
        audio.Microphone._open = saved_open


def test_good_devices_are_silent() -> None:
    """No warning for the devices the project runs on.

    A warning here would be worse than no warning at all: the operator learns to
    skip the line, and the real one gets skipped with it.
    """
    cases = [
        # The Pi production path, native 16 kHz mono over USB Audio Class. This is
        # the boundary case -- 16 kHz must not read as narrowband.
        ("Anker PowerConf", 16000),
        ("Shure MV7", 48000),
        ("NVIDIA Broadcast", 48000),
        ("Microphone (USB Audio Device)", 44100),
        # A USB headset is a fine STT mic, so "headset" alone must not trip it.
        ("Headset Microphone (Logitech USB Headset)", 48000),
    ]
    for name, sr in cases:
        risk = assess_input_device(name, sr)
        check(
            f"silent for {name}",
            risk is None,
            "no warning" if risk is None else f"unexpected {risk.kind}",
        )


def test_unknown_rate_is_not_a_fault() -> None:
    """Absent rate information must not manufacture a warning."""
    for sr in (None, 0, 0.0, "", "not-a-number"):
        risk = assess_input_device("Mystery Device", sr)
        check(
            f"silent for default_sr={sr!r}",
            risk is None,
            "no warning" if risk is None else f"unexpected {risk.kind}",
        )


def test_survives_a_missing_name() -> None:
    """A device with no usable name must not raise."""
    try:
        risk = assess_input_device(None, 48000)
        check("no name does not raise", risk is None, "returned None")
    except Exception as e:  # noqa: BLE001 - any raise is the failure being checked
        check("no name does not raise", False, f"raised {type(e).__name__}: {e}")


def test_target_rate_is_honored() -> None:
    """The threshold follows the caller's target, not a baked-in 16 kHz.

    audio.py passes its TARGET_SR, so a pipeline retuned to a different frame rate
    keeps a correct warning instead of one pinned to the old rate.
    """
    risk = assess_input_device(
        "Some Mic", 16000, target_sr=48000, host_api="Windows WASAPI"
    )
    check(
        "16 kHz is narrowband against a 48 kHz target",
        risk is not None and risk.kind == "narrowband",
        f"kind={risk.kind if risk else None}",
    )
    check(
        "default target is the pipeline's 16 kHz",
        capture_quality.DEFAULT_TARGET_SR == 16000,
        f"DEFAULT_TARGET_SR={capture_quality.DEFAULT_TARGET_SR}",
    )


class _FakeMic:
    """audio.Microphone stand-in whose stream has already ended, so run_loop
    reaches the startup print and then breaks immediately."""

    device_label = "fake-mic"

    def __init__(self, capture_risk):
        self.capture_risk = capture_risk

    def __enter__(self):
        return self

    def __exit__(self, *exc):
        return False

    def frames(self):
        return iter(())

    def pause(self):
        pass

    def resume(self):
        pass

    def flush(self):
        pass

    def active(self) -> bool:
        return False


def _run_loop_output(capture_risk) -> str:
    """Drive pipeline.run_loop with a fake mic carrying `capture_risk`, and return
    what it printed. run_loop imports `audio` lazily, so a fake module stands in
    and no device, PortAudio, or model is touched."""
    saved = {
        "load_config": pipeline.load_config,
        "warm_models": pipeline.warm_models,
        "run_turn": pipeline.run_turn,
    }
    saved_audio = sys.modules.get("audio")

    fake_audio = types.ModuleType("audio")
    fake_audio.Microphone = lambda name=None: _FakeMic(capture_risk)
    fake_audio.play_wav = lambda path, name=None: None

    pipeline.load_config = lambda: {"wake_word": "computah", "wake_chime": False}
    pipeline.warm_models = lambda cfg=None, wake_word=None: {}
    pipeline.run_turn = lambda frames, **kw: None
    sys.modules["audio"] = fake_audio
    buf = io.StringIO()
    try:
        with contextlib.redirect_stdout(buf):
            pipeline.run_loop()
    finally:
        pipeline.load_config = saved["load_config"]
        pipeline.warm_models = saved["warm_models"]
        pipeline.run_turn = saved["run_turn"]
        if saved_audio is None:
            sys.modules.pop("audio", None)
        else:
            sys.modules["audio"] = saved_audio
    return buf.getvalue()


def test_startup_warns_about_a_bad_device() -> None:
    """The verdict has to reach the operator, not just exist.

    This is the acceptance criterion from issue #34: warn at `--listen` startup.
    A classifier that is right and silent leaves the original failure intact --
    the loop looks healthy and only the transcript is wrong.
    """
    risk = CaptureRisk("hands-free-profile", "narrowband and frame-dropped, use USB")
    out = _run_loop_output(risk)
    check(
        "run_loop prints the warning at startup",
        risk.message in out,
        "message found" if risk.message in out else f"message absent from {out!r}",
    )
    check(
        "the warning is labelled so it reads as a problem",
        "WARNING" in out,
        "labelled WARNING" if "WARNING" in out else "no WARNING label",
    )
    check(
        "warned once, not per frame",
        out.count(risk.message) == 1,
        f"occurrences={out.count(risk.message)}",
    )
    check(
        "the loop still starts on a flagged device",
        "listening" in out,
        "reached listening" if "listening" in out else f"never listened: {out!r}",
    )


def test_startup_is_quiet_on_a_good_device() -> None:
    """No WARNING line when the device is fine, so the line stays meaningful."""
    out = _run_loop_output(None)
    check(
        "no warning for a clean device",
        "WARNING" not in out,
        "no WARNING printed" if "WARNING" not in out else f"unexpected: {out!r}",
    )


SR = 16000


def _speechlike(seconds: float = 3.0, tilt: float = 0.9, seed: int = 7):
    """Noise shaped like a voice: a natural spectral tilt plus formant bumps.

    Real speech is the hard case for a spectral check, because its energy falls
    off steeply with frequency all on its own. A flat signal would make any
    threshold look good.
    """
    rng = np.random.default_rng(seed)
    n = int(seconds * SR)
    spectrum = np.fft.rfft(rng.normal(0, 1, n))
    freqs = np.fft.rfftfreq(n, 1.0 / SR)
    shape = 1.0 / np.maximum(freqs, 50) ** tilt
    for center, gain in ((500, 6), (1500, 4), (2600, 3), (5200, 1.6), (6800, 1.3)):
        shape = shape * (1 + gain * np.exp(-((freqs - center) ** 2) / (2 * 320**2)))
    return np.fft.irfft(spectrum * shape, n=n)


def _bandlimited(
    signal, cutoff_hz: float, floor_db: float | None = None, seed: int = 3
):
    """Zero everything above cutoff, the way resampling from a narrower rate does.

    `floor_db` adds the noise a real resampler leaves in the emptied band; None is
    the ideal case that leaves it at exactly zero.
    """
    spectrum = np.fft.rfft(signal)
    freqs = np.fft.rfftfreq(len(signal), 1.0 / SR)
    spectrum[freqs >= cutoff_hz] = 0
    out = np.fft.irfft(spectrum, n=len(signal))
    if floor_db is not None:
        rng = np.random.default_rng(seed)
        rms = float(np.sqrt((out**2).mean()))
        out = out + rng.normal(0, rms * 10 ** (floor_db / 20), len(out))
    return out


def test_spectrum_flags_upsampled_narrowband() -> None:
    """The Linux hole: audio that arrived at 16 kHz but started life at 8 kHz.

    Neither the name nor the advertised rate can catch this -- PipeWire resamples
    and reports 16 kHz -- so the audio is the only evidence left.
    """
    voice = _speechlike()
    cases = [
        ("ideal resampler", _bandlimited(voice, 4000)),
        ("-60 dB resampler floor", _bandlimited(voice, 4000, floor_db=-60)),
        ("-40 dB sloppy resampler", _bandlimited(voice, 4000, floor_db=-40)),
        ("CVSD narrowband (3.4 kHz)", _bandlimited(voice, 3400, floor_db=-50)),
    ]
    for label, signal in cases:
        assessment = assess_capture_spectrum(signal, SR)
        risk = assessment.risk
        check(
            f"flags {label}",
            risk is not None and risk.kind == "upsampled-narrowband",
            f"kind={risk.kind if risk else None}",
        )


def test_spectrum_is_silent_on_genuine_full_band_audio() -> None:
    """A false positive here tells someone to replace a mic that works.

    The dull-voice case is the one that matters: it has very little energy up
    high on its own, which is exactly what a naive high-band-energy check would
    misread as narrowband.
    """
    rng = np.random.default_rng(11)
    cases = [
        ("speech-like full band", _speechlike()),
        ("very dull, heavily tilted voice", _speechlike(tilt=1.8)),
        ("white noise", rng.normal(0, 1, 3 * SR)),
        ("mild 7 kHz lowpass", _bandlimited(_speechlike(), 7000, floor_db=-50)),
    ]
    for label, signal in cases:
        assessment = assess_capture_spectrum(signal, SR)
        check(
            f"does not flag {label}",
            assessment.risk is None,
            (
                "no warning"
                if assessment.risk is None
                else f"unexpected {assessment.risk.kind}"
            ),
        )
        check(
            f"keeps {label} inconclusive",
            assessment.status == "inconclusive",
            f"status={assessment.status}",
        )


def test_spectrum_reports_no_information_rather_than_guessing() -> None:
    """Nothing to measure must read as nothing to say, not as a fault."""
    cases = [
        ("digital silence", np.zeros(3 * SR)),
        ("a constant offset", np.full(3 * SR, 1234.0)),
        ("too short to analyze", _speechlike(seconds=0.2)),
        ("empty buffer", np.zeros(0)),
    ]
    for label, signal in cases:
        assessment = assess_capture_spectrum(signal, SR)
        check(
            f"inconclusive verdict for {label}",
            getattr(assessment, "status", None) == "inconclusive",
            f"status={getattr(assessment, 'status', None)}",
        )


def test_full_band_spectrum_does_not_clear_wideband_hfp() -> None:
    """A 4 kHz cliff check cannot rule out wideband HFP codec/drop artifacts."""
    assessment = assess_capture_spectrum(_speechlike(), SR)
    check(
        "full-band spectrum remains inconclusive",
        getattr(assessment, "status", None) == "inconclusive",
        f"status={getattr(assessment, 'status', None)}",
    )
    message = getattr(assessment, "message", "")
    check(
        "inconclusive message names the undetected artifacts",
        "wideband HFP" in message and "frame drops" in message,
        f"message={message!r}",
    )


def _run_mic_test(signal: np.ndarray, seconds: float) -> tuple[int, str]:
    """Run audio._test_mic against deterministic frames from `signal`."""

    class FakeMicrophone:
        device_label = "Unmarked PipeWire source (ALSA)"
        device_default_sr = SR
        host_api = "ALSA"
        capture_risk = None
        _open_sr = SR
        _open_ch = 1

        def __enter__(self):
            return self

        def __exit__(self, *exc):
            return False

        def frames(self):
            usable = len(signal) - (len(signal) % audio.FRAME_SIZE)
            for offset in range(0, usable, audio.FRAME_SIZE):
                yield signal[offset : offset + audio.FRAME_SIZE].astype(np.int16)

    saved_microphone = audio.Microphone
    audio.Microphone = lambda name=None: FakeMicrophone()
    output = io.StringIO()
    try:
        with contextlib.redirect_stdout(output):
            exit_code = audio._test_mic("", seconds)
    finally:
        audio.Microphone = saved_microphone
    return exit_code, output.getvalue()


def test_mic_returns_inconclusive_for_unanalyzable_audio() -> None:
    """A short or stuck nonzero stream must not receive exit code 0."""
    cases = [
        ("too short", (_speechlike(seconds=0.2) * 8000), 0.2),
        ("constant nonzero", np.full(3 * SR, 1000), 3.0),
    ]
    for label, signal, seconds in cases:
        exit_code, output = _run_mic_test(signal.astype(np.int16), seconds)
        check(
            f"{label} capture exits inconclusive",
            exit_code == 4,
            f"exit_code={exit_code}",
        )
        check(
            f"{label} capture reports inconclusive",
            "inconclusive" in output.lower(),
            f"output={output!r}",
        )


def test_mic_does_not_pass_unmarked_wideband_hfp() -> None:
    """Full-band energy alone cannot make an unmarked PipeWire source usable."""
    signal = (_speechlike() * 8000).astype(np.int16)
    exit_code, output = _run_mic_test(signal, 3.0)
    check(
        "full-band unmarked source exits inconclusive",
        exit_code == 4,
        f"exit_code={exit_code}",
    )
    check(
        "full-band unmarked source explains wideband HFP gap",
        "wideband HFP" in output,
        f"output={output!r}",
    )


def test_spectrum_threshold_keeps_its_measured_margin() -> None:
    """Pin the separation the threshold was chosen from.

    The docstring claims genuine audio sits well above the cutoff and narrowband
    well below. If a future tweak narrows that gap, this fails while both verdict
    tests above still pass -- the margin is the thing worth protecting, not just
    the verdicts at today's values.
    """
    # The dull voice, not the ordinary one: it is the genuine case with the least
    # headroom, so it is the side of the threshold that gives out first. Measuring
    # the comfortable case instead would leave the binding one unpinned.
    genuine = capture_quality._cliff_ratio(
        _speechlike(tilt=1.8), SR, capture_quality.DEFAULT_TARGET_SR
    )
    narrow = capture_quality._cliff_ratio(
        _bandlimited(_speechlike(), 4000, floor_db=-40),
        SR,
        capture_quality.DEFAULT_TARGET_SR,
    )
    threshold = capture_quality._CLIFF_RATIO_THRESHOLD
    check(
        "genuine audio clears the threshold with margin",
        genuine > threshold * 3,
        f"ratio={genuine:.4f} vs threshold={threshold} (need >{threshold * 3:.3f})",
    )
    check(
        "narrowband audio sits under it with margin",
        narrow < threshold / 3,
        f"ratio={narrow:.4f} vs threshold={threshold} (need <{threshold / 3:.3f})",
    )


def test_spectrum_survives_int16_frames() -> None:
    """--test-mic hands it the concatenated int16 frames, not floats."""
    frames = (_bandlimited(_speechlike(), 4000, floor_db=-60) * 8000).astype(np.int16)
    risk = assess_capture_spectrum(frames, SR).risk
    check(
        "int16 input is judged the same as float",
        risk is not None and risk.kind == "upsampled-narrowband",
        f"kind={risk.kind if risk else None}",
    )


def main() -> int:
    print("capture-device suitability checks (issue #34)")
    test_flags_bluetooth_hands_free_by_name()
    test_hands_free_wins_over_rate()
    test_flags_narrowband_rate()
    test_default_rate_is_advisory_outside_wasapi()
    test_microphone_passes_host_api_to_default_rate_check()
    test_microphone_flags_an_actual_low_rate_fallback()
    test_good_devices_are_silent()
    test_unknown_rate_is_not_a_fault()
    test_survives_a_missing_name()
    test_target_rate_is_honored()
    test_startup_warns_about_a_bad_device()
    test_startup_is_quiet_on_a_good_device()
    test_spectrum_flags_upsampled_narrowband()
    test_spectrum_is_silent_on_genuine_full_band_audio()
    test_spectrum_reports_no_information_rather_than_guessing()
    test_full_band_spectrum_does_not_clear_wideband_hfp()
    test_mic_returns_inconclusive_for_unanalyzable_audio()
    test_mic_does_not_pass_unmarked_wideband_hfp()
    test_spectrum_threshold_keeps_its_measured_margin()
    test_spectrum_survives_int16_frames()
    failed = results.count(False)
    print(f"\n{len(results) - failed}/{len(results)} checks passed")
    return 1 if failed else 0


if __name__ == "__main__":
    raise SystemExit(main())