#!/usr/bin/env python3
"""Fast, no-model tests for the wake-acknowledgment chime (issue #41).

Two concerns, both mic-free and model-free:
  1. The cue generator (chime.py) -- a 16 kHz mono int16 two-tone rising cue with
     edge fades. Checks the format, that the edges fade to near silence (no click),
     and that the pitch actually rises (the "two-tone rising" spec).
  2. The wiring -- both live loops fire the chime at the wake->capture boundary, the
     instant the wake fires and before the request is captured, and never when no
     wake fires. The detection/capture/transcribe/brain/speak/playback stages are
     stubbed, so this runs with no microphone, model, or audio device.

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

from __future__ import annotations

import os
import sys
import tempfile

import numpy as np
import soundfile as sf

import chime
import live_driver
import pipeline
from pipeline import FRAME_SIZE, Transcript

PASS, FAIL = "PASS", "FAIL"
results: list[tuple[str, str]] = []


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


class FakeMic:
    """Records flush-to-now calls. The post-cue drop is now mic.flush() (issue #56):
    a cue that played flushes once; a failed or suppressed cue never flushes."""

    def __init__(self) -> None:
        self.flushed = 0

    def flush(self) -> None:
        self.flushed += 1


def _dom_freq(seg: np.ndarray) -> float:
    """Dominant frequency (Hz) of an int16 segment, windowed to limit edge leakage."""
    x = seg.astype(np.float64)
    if x.size < 16:
        return 0.0
    spec = np.abs(np.fft.rfft(x * np.hanning(x.size)))
    freqs = np.fft.rfftfreq(x.size, 1.0 / chime.SR)
    return float(freqs[int(np.argmax(spec))])


def test_generator() -> None:
    print("=== chime cue generator ===")
    pcm = chime.wake_cue_pcm()

    check(
        "cue is 1-D int16 PCM",
        isinstance(pcm, np.ndarray) and pcm.dtype == np.int16 and pcm.ndim == 1,
        f"dtype={pcm.dtype} ndim={pcm.ndim}",
    )

    # A short cue: long enough to be heard, short enough not to delay the turn.
    ms = pcm.size / chime.SR * 1000.0
    check(
        "cue length is in a sane 80-600 ms range",
        80.0 <= ms <= 600.0,
        f"{ms:.0f} ms ({pcm.size} samples)",
    )

    peak = int(np.max(np.abs(pcm))) if pcm.size else 0
    check("cue is audible but not clipped", 1000 < peak < 32768, f"peak={peak}")

    # Raised-cosine edge fades ramp through zero, so the waveform is continuous at
    # both ends -- no step discontinuity, hence no click.
    edge = max(abs(int(pcm[0])), abs(int(pcm[-1]))) if pcm.size else 0
    check(
        "edges fade to near silence (no click)", edge < 500, f"max edge sample={edge}"
    )

    # "Two-tone rising": the back of the cue is higher in pitch than the front.
    n = pcm.size
    lo = _dom_freq(pcm[: int(0.4 * n)])
    hi = _dom_freq(pcm[int(0.6 * n) :])
    check(
        "pitch rises (second tone higher than first)",
        hi > lo + 50,
        f"front~{lo:.0f}Hz back~{hi:.0f}Hz",
    )

    # The WAV helper writes a readable 16 kHz mono file and is idempotent.
    d = tempfile.mkdtemp()
    path = os.path.join(d, "cue.wav")
    p1 = chime.wake_cue_wav(path)
    data, sr = sf.read(p1, dtype="int16", always_2d=False)
    check(
        "wake_cue_wav writes a 16 kHz mono WAV at the given path",
        p1 == path and os.path.exists(path) and sr == chime.SR and data.ndim == 1,
        f"path={p1!r} sr={sr} ndim={data.ndim}",
    )
    mtime = os.path.getmtime(path)
    p2 = chime.wake_cue_wav(path)
    check(
        "wake_cue_wav is idempotent (same path, file intact)",
        p2 == path and os.path.getmtime(path) == mtime,
        f"p2={p2!r}",
    )


# --- wiring: pipeline.run_turn fires on_wake between detection and capture --------


def _run_turn_order(
    cfg_wake: float, fire: bool, use_hook: bool = True, hook_boundary: bool = True
):
    """Drive pipeline.run_turn with every stage stubbed, recording the call order of
    detection, the on_wake hook, and capture, plus the pre-roll capture_request saw.

    `fire` toggles whether a wake fires; `use_hook` toggles wiring the on_wake chime
    hook, so a caller can compare the pre-roll cleared (chime) vs preserved (no chime).
    `hook_boundary` is what the hook returns: True models a cue that played and flushed
    (a fresh capture boundary), False a cue that failed (no boundary, buffer kept) --
    so a caller can check the pre-roll is dropped only on the boundary case.
    Returns (order, seen_preroll)."""
    order: list[str] = []
    seen_preroll: list[list] = []
    real = (
        pipeline._resolve_wake_path,
        pipeline._get_oww_model,
        pipeline.stream_detect_wake,
        pipeline.capture_request,
        pipeline.transcribe_detailed,
        pipeline.guard_transcript,
        pipeline.brain,
        pipeline.speak,
    )

    pipeline._resolve_wake_path = lambda name: "dummy"
    pipeline._get_oww_model = lambda path: object()

    def fake_detect(frames, model, threshold, preroll=None):
        order.append("detect")
        if fire and preroll is not None:  # a real detection leaves wake frames here
            preroll.append(np.full(FRAME_SIZE, 7, dtype=np.int16))
        return 0.9 if fire else None

    def fake_capture(frames, preroll=None, vad_threshold=None, **_):
        order.append("capture")
        seen_preroll.append(list(preroll) if preroll is not None else [])
        return np.full(8 * FRAME_SIZE, 4000, dtype=np.int16)

    pipeline.stream_detect_wake = fake_detect
    pipeline.capture_request = fake_capture
    pipeline.transcribe_detailed = lambda p: Transcript("file an issue", -0.2, 0.05)
    pipeline.guard_transcript = lambda heard, cfg: (True, "")
    pipeline.brain = lambda t, **_: "ok"
    pipeline.speak = lambda text, out, **_: out

    def on_wake() -> bool:
        order.append("wake")
        return hook_boundary  # True = cue played and flushed (a fresh boundary)

    try:
        pipeline.run_turn(
            iter([]),
            model_name="x",
            threshold=cfg_wake,
            out_wav_path=os.devnull,
            on_wake=on_wake if use_hook else None,
        )
    finally:
        (
            pipeline._resolve_wake_path,
            pipeline._get_oww_model,
            pipeline.stream_detect_wake,
            pipeline.capture_request,
            pipeline.transcribe_detailed,
            pipeline.guard_transcript,
            pipeline.brain,
            pipeline.speak,
        ) = real
    return order, seen_preroll


def test_run_turn_hook() -> None:
    print("\n=== pipeline.run_turn on_wake hook ===")
    order, pr = _run_turn_order(0.5, fire=True, use_hook=True, hook_boundary=True)
    check(
        "on_wake fires once, after detection and before capture",
        order == ["detect", "wake", "capture"],
        f"order={order}",
    )
    check(
        "a cue that played (hook returns True) drops the pre-roll so the wake-word "
        "tail is not prepended",
        pr == [[]],
        f"preroll lengths seen by capture={[len(p) for p in pr]}",
    )

    order, pr = _run_turn_order(0.5, fire=True, use_hook=True, hook_boundary=False)
    check(
        "a cue that failed (hook returns falsy, no boundary) keeps the pre-roll so "
        "a no-pause command is still recovered (issue #30)",
        order == ["detect", "wake", "capture"] and len(pr) == 1 and len(pr[0]) == 1,
        f"order={order} preroll lengths={[len(p) for p in pr]}",
    )

    order, pr = _run_turn_order(0.5, fire=True, use_hook=False)
    check(
        "without the chime hook the pre-roll is preserved (issue #30 recovery)",
        order == ["detect", "capture"] and len(pr) == 1 and len(pr[0]) == 1,
        f"order={order} preroll lengths={[len(p) for p in pr]}",
    )

    order, _ = _run_turn_order(0.5, fire=False, use_hook=True)
    check(
        "on_wake does not fire when no wake fires",
        order == ["detect"] and "wake" not in order,
        f"order={order}",
    )


# --- wiring: live_driver plays the chime between wake and capture ----------------


def test_live_driver_chime() -> None:
    print("\n=== live_driver fires the chime at the wake->capture boundary ===")
    order: list[str] = []
    real = (
        live_driver.listen_for_wake,
        pipeline.capture_request,
        pipeline.transcribe_detailed,
        pipeline.brain,
        pipeline.speak,
        live_driver._play_wav,
        chime.wake_cue_wav,
    )

    def fake_listen(fr, m, t, d, preroll=None):
        if preroll is not None:  # detection leaves the wake-word tail in the pre-roll
            preroll.append(np.full(FRAME_SIZE, 7, dtype=np.int16))
        return 0.9

    live_driver.listen_for_wake = fake_listen
    chime.wake_cue_wav = lambda path=None: "CUE.wav"

    seen_preroll: list[list] = []

    def fake_capture(fr, preroll=None, vad_threshold=None, **_):
        order.append("capture")
        seen_preroll.append(list(preroll) if preroll is not None else [])
        return np.full(8 * FRAME_SIZE, 4000, dtype=np.int16)

    def fake_play(path, dev):
        order.append(f"play:{'cue' if path == 'CUE.wav' else 'reply'}")

    pipeline.capture_request = fake_capture
    pipeline.transcribe_detailed = lambda p: Transcript("file an issue", -0.2, 0.05)
    pipeline.brain = lambda t, **_: "Brain answer."
    pipeline.speak = lambda text, out, **_: out
    live_driver._play_wav = fake_play
    mic = FakeMic()

    cfg = {
        "stt_confidence_guard": True,
        "stt_min_avg_logprob": -1.0,
        "stt_max_no_speech_prob": 0.6,
        "wake_chime": True,
        "capture_vad_threshold": 0.5,
        "endpoint_silence_ms": 800,
        "max_request_ms": 8000,
    }
    try:
        ran = live_driver.run_turn(
            iter([]), mic, object(), 0.5, os.devnull, None, cfg, False
        )
    finally:
        (
            live_driver.listen_for_wake,
            pipeline.capture_request,
            pipeline.transcribe_detailed,
            pipeline.brain,
            pipeline.speak,
            live_driver._play_wav,
            chime.wake_cue_wav,
        ) = real

    check(
        "chime plays before the request is captured",
        "play:cue" in order
        and "capture" in order
        and order.index("play:cue") < order.index("capture"),
        f"ran={ran} order={order}",
    )
    check(
        "reply still plays after capture",
        "play:reply" in order and order.index("play:reply") > order.index("capture"),
        f"order={order}",
    )
    check(
        "a cue that played flushes-to-now once (drops the post-cue mic bleed)",
        mic.flushed == 1,
        f"flushed={mic.flushed}",
    )
    check(
        "the chime drops the pre-roll so the wake-word tail is not prepended",
        seen_preroll == [[]],
        f"preroll lengths seen by capture={[len(p) for p in seen_preroll]}",
    )


# --- wiring: a failed cue keeps the pre-roll (no false boundary) ------------------


def test_live_driver_chime_failure_keeps_preroll() -> None:
    print("\n=== live_driver: a failed cue keeps the pre-roll (issue #30 recovery) ===")
    order: list[str] = []
    real = (
        live_driver.listen_for_wake,
        pipeline.capture_request,
        pipeline.transcribe_detailed,
        pipeline.brain,
        pipeline.speak,
        live_driver._play_wav,
        chime.wake_cue_wav,
    )

    def fake_listen(fr, m, t, d, preroll=None):
        if preroll is not None:
            preroll.append(np.full(FRAME_SIZE, 7, dtype=np.int16))
        return 0.9

    live_driver.listen_for_wake = fake_listen
    chime.wake_cue_wav = lambda path=None: "CUE.wav"

    seen_preroll: list[list] = []

    def fake_capture(fr, preroll=None, vad_threshold=None, **_):
        order.append("capture")
        seen_preroll.append(list(preroll) if preroll is not None else [])
        return np.full(8 * FRAME_SIZE, 4000, dtype=np.int16)

    def failing_play(path, dev):
        # Both the cue and the reply route through _play_wav; the cue must fail so the
        # turn exercises the chime's failure branch. The reply's own try/except logs
        # and continues, so failing it too does not abort the turn.
        raise RuntimeError("aplay unavailable")

    pipeline.capture_request = fake_capture
    pipeline.transcribe_detailed = lambda p: Transcript("file an issue", -0.2, 0.05)
    pipeline.brain = lambda t, **_: "Brain answer."
    pipeline.speak = lambda text, out, **_: out
    live_driver._play_wav = failing_play
    mic = FakeMic()

    cfg = {
        "stt_confidence_guard": True,
        "stt_min_avg_logprob": -1.0,
        "stt_max_no_speech_prob": 0.6,
        "wake_chime": True,
        "capture_vad_threshold": 0.5,
        "endpoint_silence_ms": 800,
        "max_request_ms": 8000,
    }
    try:
        ran = live_driver.run_turn(
            iter([]), mic, object(), 0.5, os.devnull, None, cfg, False
        )
    finally:
        (
            live_driver.listen_for_wake,
            pipeline.capture_request,
            pipeline.transcribe_detailed,
            pipeline.brain,
            pipeline.speak,
            live_driver._play_wav,
            chime.wake_cue_wav,
        ) = real

    check(
        "a failed cue does not flush without a clean playback boundary",
        mic.flushed == 0,
        f"flushed={mic.flushed}",
    )
    check(
        "a failed cue keeps the pre-roll so a no-pause command is still recovered",
        len(seen_preroll) == 1 and len(seen_preroll[0]) == 1,
        f"preroll lengths seen by capture={[len(p) for p in seen_preroll]}",
    )
    check(
        "the turn still completes after a failed cue",
        ran is True and "capture" in order,
        f"ran={ran} order={order}",
    )


# --- wiring: the chime is opt-in (default off) -----------------------------------


def test_chime_opt_in_default_off() -> None:
    print("\n=== chime is opt-in: no cue when wake_chime is off or unset ===")
    # The cue regresses the no-pause case on a half-duplex device, so it ships off by
    # default (issue #41). Both an absent key and an explicit false must suppress it,
    # while the turn itself still runs normally (capture -> reply, just no cue/flush).
    for label, cfg_extra in (
        ("wake_chime unset", {}),
        ("wake_chime false", {"wake_chime": False}),
    ):
        order: list[str] = []
        real = (
            live_driver.listen_for_wake,
            pipeline.capture_request,
            pipeline.transcribe_detailed,
            pipeline.brain,
            pipeline.speak,
            live_driver._play_wav,
            chime.wake_cue_wav,
        )

        def fake_listen(fr, m, t, d, preroll=None):
            if preroll is not None:
                preroll.append(np.full(FRAME_SIZE, 7, dtype=np.int16))
            return 0.9

        live_driver.listen_for_wake = fake_listen
        chime.wake_cue_wav = lambda path=None: "CUE.wav"

        seen_preroll: list[list] = []

        def fake_capture(fr, preroll=None, vad_threshold=None, **_):
            order.append("capture")
            seen_preroll.append(list(preroll) if preroll is not None else [])
            return np.full(8 * FRAME_SIZE, 4000, dtype=np.int16)

        def fake_play(path, dev):
            order.append(f"play:{'cue' if path == 'CUE.wav' else 'reply'}")

        pipeline.capture_request = fake_capture
        pipeline.transcribe_detailed = lambda p: Transcript("file an issue", -0.2, 0.05)
        pipeline.brain = lambda t, **_: "Brain answer."
        pipeline.speak = lambda text, out, **_: out
        live_driver._play_wav = fake_play
        mic = FakeMic()

        cfg = {
            "stt_confidence_guard": True,
            "stt_min_avg_logprob": -1.0,
            "stt_max_no_speech_prob": 0.6,
            "capture_vad_threshold": 0.5,
            "endpoint_silence_ms": 800,
            "max_request_ms": 8000,
            **cfg_extra,
        }
        try:
            ran = live_driver.run_turn(
                iter([]), mic, object(), 0.5, os.devnull, None, cfg, False
            )
        finally:
            (
                live_driver.listen_for_wake,
                pipeline.capture_request,
                pipeline.transcribe_detailed,
                pipeline.brain,
                pipeline.speak,
                live_driver._play_wav,
                chime.wake_cue_wav,
            ) = real

        check(
            f"no cue and no flush when {label}",
            "play:cue" not in order and mic.flushed == 0,
            f"ran={ran} order={order} flushed={mic.flushed}",
        )
        check(
            f"the turn still runs (capture then reply) when {label}",
            "capture" in order and "play:reply" in order,
            f"order={order}",
        )
        check(
            f"the pre-roll is preserved (not chime-cleared) when {label}",
            len(seen_preroll) == 1 and len(seen_preroll[0]) == 1,
            f"preroll lengths seen by capture={[len(p) for p in seen_preroll]}",
        )


def main() -> int:
    test_generator()
    test_run_turn_hook()
    test_live_driver_chime()
    test_live_driver_chime_failure_keeps_preroll()
    test_chime_opt_in_default_off()
    n_pass = sum(1 for r in results if r[0] == PASS)
    n_total = len(results)
    print(f"\n=== {n_pass}/{n_total} checks passed ===")
    return 0 if n_pass == n_total else 1


if __name__ == "__main__":
    sys.exit(main())