"""Barge-in detection, driven by a fake mic stream and a fake wake model
(no audio hardware, no ONNX runtime)."""

import sys
from pathlib import Path

import numpy as np

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from bolt_pet.audio.barge_in import BargeInDetector, WakeWordBargeIn, make_detector


class FakeStream:
    """Yields frames of a given amplitude, mimicking sounddevice's
    (data, overflowed) 2-D int16 return shape."""

    def __init__(self, amplitudes):
        self._amplitudes = list(amplitudes)

    def read(self, frames):
        amplitude = self._amplitudes.pop(0) if self._amplitudes else 0
        data = np.full((frames, 1), amplitude, dtype=np.int16)
        return data, False


def test_silence_never_interrupts():
    detector = BargeInDetector(FakeStream([0] * 20), threshold=1000, required_frames=3)
    assert not any(detector.check() for _ in range(20))


def test_sustained_speech_interrupts_after_the_required_frames():
    detector = BargeInDetector(FakeStream([2000] * 5), threshold=1000, required_frames=3)
    assert detector.check() is False
    assert detector.check() is False
    assert detector.check() is True


def test_a_single_thump_does_not_interrupt():
    # loud, quiet, loud, quiet ... never three in a row
    detector = BargeInDetector(FakeStream([2000, 0, 2000, 0, 2000, 0]), threshold=1000, required_frames=3)
    assert not any(detector.check() for _ in range(6))


def test_counter_resets_after_a_quiet_frame():
    detector = BargeInDetector(FakeStream([2000, 2000, 0, 2000, 2000, 2000]), threshold=1000, required_frames=3)
    results = [detector.check() for _ in range(6)]
    assert results == [False, False, False, False, False, True]


def test_reset_clears_progress():
    detector = BargeInDetector(FakeStream([2000] * 6), threshold=1000, required_frames=3)
    detector.check()
    detector.check()
    detector.reset()
    assert detector.check() is False
    assert detector.loud_frames == 1


def test_a_mic_error_mid_playback_is_not_fatal():
    class BrokenStream:
        def read(self, frames):
            raise OSError("device disappeared")

    detector = BargeInDetector(BrokenStream(), threshold=1000, required_frames=1)
    assert detector.check() is False


# ── wake-word mode ───────────────────────────────────────────────────────────


class FakeModel:
    """Scores frames from a canned list, mimicking openWakeWord's
    {class_name: score} return. Records reset() calls."""

    def __init__(self, scores):
        self._scores = list(scores)
        self.resets = 0

    def predict(self, frame):
        score = self._scores.pop(0) if self._scores else 0.0
        return {"thunderbolt": score}

    def reset(self):
        self.resets += 1


def _wake_detector(scores, threshold=0.5, amplitudes=None):
    stream = FakeStream(amplitudes if amplitudes is not None else [500] * len(scores))
    return WakeWordBargeIn(stream, model=FakeModel(scores), threshold=threshold), stream


def test_loud_noise_alone_does_not_interrupt_in_wake_mode():
    """The whole point of wake mode: a slammed door is deafening and scores
    nothing, so the pet keeps talking."""
    detector, _ = _wake_detector([0.01] * 6, amplitudes=[30000] * 6)
    assert not any(detector.check() for _ in range(6))


def test_the_wake_word_interrupts():
    detector, _ = _wake_detector([0.1, 0.2, 0.9])
    assert [detector.check() for _ in range(3)] == [False, False, True]


def test_a_single_frame_is_enough_when_it_clears_the_threshold():
    detector, _ = _wake_detector([0.55])
    assert detector.check() is True


def test_scores_just_under_the_threshold_do_not_fire():
    detector, _ = _wake_detector([0.49, 0.499], threshold=0.5)
    assert not any(detector.check() for _ in range(2))


def test_detecting_resets_the_model_so_the_tail_is_not_reused():
    model = FakeModel([0.9])
    detector = WakeWordBargeIn(FakeStream([500]), model=model, threshold=0.5)
    assert detector.check() is True
    assert model.resets == 1


def test_reset_clears_the_models_audio_window_not_just_predictions():
    """The regression that made the pet interrupt itself a word into every
    reply: openwakeword's reset() clears only the prediction buffer, so the
    "thunderbolt" that started the turn was still in the preprocessor's
    rolling window when playback began, and the first frame fed to the model
    re-fired on it."""

    class FakePreprocessor:
        def __init__(self):
            self.raw_data_buffer = [1, 2, 3]
            self.feature_buffer = np.ones((120, 96))
            self.melspectrogram_buffer = np.zeros((76, 32))
            self.accumulated_samples = 4096

        def _get_embeddings(self, audio):
            return np.zeros((120, 96))

    model = FakeModel([0.9])
    model.preprocessor = FakePreprocessor()

    WakeWordBargeIn(FakeStream([500]), model=model, threshold=0.5).reset()

    assert model.preprocessor.raw_data_buffer == []
    assert model.preprocessor.accumulated_samples == 0
    assert not model.preprocessor.feature_buffer.any()  # blank, not the old audio
    assert model.preprocessor.melspectrogram_buffer.all()  # restored to ones


def test_the_lazy_wrapper_exposes_its_preprocessor():
    """The pet holds _default_model — a lazy *wrapper* around openwakeword's
    Model. If the wrapper stops proxying .preprocessor, hard_reset() finds
    nothing to clear and silently degrades to the shallow reset that leaves
    the last detection in the audio window. That failure is invisible: no
    exception, no log, the pet just interrupts itself again."""
    from bolt_pet.audio.wake_word import _OpenWakeWordModel

    wrapper = _OpenWakeWordModel()
    assert hasattr(wrapper, "preprocessor")
    assert wrapper.preprocessor is None  # not loaded yet: a no-op, not a load

    class FakeInner:
        preprocessor = object()

        def reset(self):
            pass

    wrapper._model = FakeInner()
    assert wrapper.preprocessor is FakeInner.preprocessor


def test_a_model_without_a_preprocessor_still_resets():
    """Fakes in tests, and any future openwakeword whose internals moved."""
    model = FakeModel([0.0])
    WakeWordBargeIn(FakeStream([500]), model=model, threshold=0.5).reset()
    assert model.resets == 1


def test_a_callable_threshold_is_read_every_frame():
    """The tray tuner's slider has to apply mid-playback, not just mid-idle."""
    threshold = {"value": 0.9}
    detector = WakeWordBargeIn(
        FakeStream([500] * 2), model=FakeModel([0.6, 0.6]),
        threshold=lambda: threshold["value"],
    )
    assert detector.check() is False
    threshold["value"] = 0.5
    assert detector.check() is True


def test_a_model_that_blows_up_mid_playback_is_not_fatal():
    class BrokenModel:
        def predict(self, frame):
            raise RuntimeError("onnx session died")

        def reset(self):
            raise RuntimeError("still dead")

    detector = WakeWordBargeIn(FakeStream([500]), model=BrokenModel(), threshold=0.5)
    assert detector.check() is False
    detector.reset()  # must not raise either


def test_a_mic_error_is_not_fatal_in_wake_mode():
    class BrokenStream:
        def read(self, frames):
            raise OSError("device disappeared")

    detector = WakeWordBargeIn(BrokenStream(), model=FakeModel([0.9]), threshold=0.5)
    assert detector.check() is False


def test_make_detector_picks_the_mode():
    stream = FakeStream([0])
    assert isinstance(make_detector(stream, mode="wake"), WakeWordBargeIn)
    assert isinstance(make_detector(stream, mode="energy"), BargeInDetector)
    # A typo in .env shouldn't stop the pet from starting.
    assert isinstance(make_detector(stream, mode="waek"), BargeInDetector)