"""Verify replay timing assumptions against Pipecat's actual buffer logic."""

import asyncio
import time
import unittest
from types import SimpleNamespace
from unittest.mock import patch

import numpy as np
from pipecat.audio.turn.base_turn_analyzer import EndOfTurnState
from pipecat.audio.turn.smart_turn.base_smart_turn import BaseSmartTurn, SmartTurnParams

import evaluate_work_data as evaluation


class RecordingTurn(BaseSmartTurn):
    prediction = 0

    def _predict_endpoint(self, audio):
        self.last_audio = audio.copy()
        return {"prediction": self.prediction, "probability": 0.9 if self.prediction else 0.1}


class ReplayTests(unittest.TestCase):
    def setUp(self):
        self.clock = evaluation.MediaClock()
        self.time_patch = patch.object(evaluation.base_module, "time", SimpleNamespace(
            monotonic=self.clock.monotonic, perf_counter=time.perf_counter))
        self.time_patch.start()
        self.turn = RecordingTurn(sample_rate=8000, params=SmartTurnParams())
        self.turn.set_sample_rate(8000)
        self.turn.update_vad_start_secs(0.2)
        self.chunk = np.full(256, 1234, dtype=np.int16).tobytes()

    def tearDown(self):
        if self.turn._executor:
            self.turn._executor.shutdown(wait=True)
        self.time_patch.stop()

    def feed(self, count, speaking):
        for _ in range(count):
            self.clock.seconds += 0.032
            state = self.turn.append_audio(self.chunk, speaking)
        return state

    def test_incomplete_keeps_first_start_and_resumed_speech_resets_silence(self):
        self.feed(10, False)
        self.feed(20, True)
        first_start = self.turn._speech_start_time
        self.feed(60, False)  # 1.92 s pause, below the timeout.
        state, _ = asyncio.run(self.turn.analyze_end_of_turn())
        self.assertEqual(state, EndOfTurnState.INCOMPLETE)
        preserved_count = len(self.turn._audio_buffer)
        self.feed(1, True)
        self.assertEqual(self.turn._speech_start_time, first_start)
        self.assertEqual(len(self.turn._audio_buffer), preserved_count + 1)
        self.assertEqual(self.turn._silence_ms, 0)

    def test_model_only_gets_last_eight_seconds_and_probe_does_not_clear(self):
        self.feed(320, True)  # 10.24 s accumulated speech.
        before_start = self.turn._speech_start_time
        before_count = len(self.turn._audio_buffer)
        self.turn.prediction = 1
        state, _ = self.turn._process_speech_segment(self.turn._audio_buffer)
        self.assertEqual(state, EndOfTurnState.COMPLETE)
        self.assertEqual(len(self.turn.last_audio), 8 * 8000)
        self.assertEqual(self.turn._speech_start_time, before_start)
        self.assertEqual(len(self.turn._audio_buffer), before_count)

    def test_three_second_silence_timeout_clears_old_origin(self):
        self.feed(20, True)
        state = self.feed(94, False)  # 3.008 s at 32 ms frame granularity.
        self.assertEqual(state, EndOfTurnState.COMPLETE)
        self.assertFalse(self.turn.speech_triggered)
        self.assertEqual(self.turn._speech_start_time, 0)


if __name__ == "__main__":
    unittest.main()