"""
Silero VAD provider.

Acoustic voice activity detection backed by the Silero ONNX model. Wraps
:class:`~getpatter.providers.silero_onnx.OnnxModel` with a streaming state
machine that buffers incoming PCM frames, runs inference on fixed-size
windows (256 samples at 8 kHz, 512 at 16 kHz), applies an exponential
probability filter, and emits :class:`~getpatter.providers.base.VADEvent`
transitions (``speech_start`` / ``speech_end`` / ``silence``).

Input is raw PCM ``bytes`` (int16, little-endian, mono) via
``process_frame(pcm_chunk, sample_rate)``. ``onnxruntime`` runs inline in a
``loop.run_in_executor`` worker thread to keep the event loop responsive.
"""

from __future__ import annotations

import asyncio
import logging
import time
from dataclasses import dataclass
from enum import IntEnum, StrEnum
from pathlib import Path
from typing import TYPE_CHECKING, Literal

import numpy as np

from getpatter.providers.base import VADEvent, VADProvider
from getpatter.providers.silero_onnx import (
    SUPPORTED_SAMPLE_RATES,
    OnnxModel,
    new_inference_session,
)

if TYPE_CHECKING:
    import onnxruntime  # type: ignore

logger = logging.getLogger(__name__)

SLOW_INFERENCE_THRESHOLD = 0.2  # late by 200ms


class SileroSampleRate(IntEnum):
    """Sample rates supported by the bundled Silero VAD ONNX model."""

    HZ_8000 = 8000
    HZ_16000 = 16000


class SileroVADEventType(StrEnum):
    """VAD transition types emitted by :class:`SileroVAD`."""

    SPEECH_START = "speech_start"
    SPEECH_END = "speech_end"
    SILENCE = "silence"


class SileroVADProviderTag(StrEnum):
    """Provider/model identifier strings exposed via the public properties."""

    MODEL = "silero"
    PROVIDER = "ONNX"


@dataclass
class _VADOptions:
    min_speech_duration: float
    min_silence_duration: float
    prefix_padding_duration: float
    activation_threshold: float
    deactivation_threshold: float
    sample_rate: int


class _ExpFilter:
    """Exponential smoothing filter.

    Returns a smoothed value using ``a = alpha**exp`` as the smoothing
    coefficient so that callers can adapt per-sample weight based on how
    stale the filter state is relative to the incoming sample.
    """

    def __init__(self, alpha: float) -> None:
        if not 0 < alpha <= 1:
            raise ValueError("alpha must be in (0, 1].")
        self._alpha = alpha
        self._filtered: float | None = None

    def apply(self, exp: float, sample: float) -> float:
        """Apply the exponential filter to *sample* with a per-call exponent."""
        if self._filtered is None:
            self._filtered = sample
        else:
            a = self._alpha**exp
            self._filtered = a * self._filtered + (1 - a) * sample
        return self._filtered

    def reset(self) -> None:
        """Drop the buffered filter state."""
        self._filtered = None


class SileroVAD(VADProvider):
    """Silero-based :class:`~getpatter.providers.base.VADProvider`.

    Construct via :meth:`load` to initialise the ONNX session and options::

        vad = SileroVAD.load(activation_threshold=0.5, sample_rate=16000)
        event = await vad.process_frame(pcm_bytes, sample_rate=16000)
        if event and event.type == "speech_start":
            ...
        await vad.close()

    ``process_frame`` accepts raw mono int16 little-endian PCM and returns an
    event only when the internal state transitions; it returns ``None`` while
    buffering or emitting continuous silence/speech frames that do not change
    state. Inference runs in an executor to avoid blocking the event loop.
    """

    @classmethod
    def load(
        cls,
        *,
        min_speech_duration: float = 0.25,
        # Bumped 0.1 → 0.4s after round 10f confirmed VAD speech_end fired on
        # natural inter-sentence pauses < 250ms, causing double-talk dispatch.
        # 0.4s is the industry-standard default for telephony agents — enough
        # to bridge natural inter-sentence pauses without delaying single-
        # sentence turns excessively.
        min_silence_duration: float = 0.4,
        prefix_padding_duration: float = 0.03,
        activation_threshold: float = 0.5,
        sample_rate: SileroSampleRate
        | Literal[8000, 16000] = SileroSampleRate.HZ_16000,
        force_cpu: bool = True,
        onnx_file_path: Path | str | None = None,
        deactivation_threshold: float | None = None,
    ) -> "SileroVAD":
        """Load the Silero VAD model and return a ready-to-use provider.

        Args:
            min_speech_duration: Minimum continuous speech duration (seconds)
                before a ``speech_start`` event is emitted.
            min_silence_duration: Minimum continuous silence duration (seconds)
                before a ``speech_end`` event is emitted after speech.
            prefix_padding_duration: Padding added to the start of detected
                speech. Reserved for future use; callers currently own the
                upstream PCM buffer so re-emission isn't required.
            activation_threshold: Probability threshold above which a frame is
                considered speech.
            sample_rate: Inference sample rate (8000 or 16000 Hz).
            force_cpu: Restrict ONNX Runtime to the CPU execution provider.
            onnx_file_path: Optional override for the bundled model file.
            deactivation_threshold: Exit threshold while speaking. Defaults to
                ``max(activation_threshold - 0.15, 0.01)``.
        """
        if sample_rate not in SUPPORTED_SAMPLE_RATES:
            raise ValueError("Silero VAD only supports 8KHz and 16KHz sample rates")

        if deactivation_threshold is not None and deactivation_threshold <= 0:
            raise ValueError("deactivation_threshold must be greater than 0")

        session = new_inference_session(force_cpu, onnx_file_path=onnx_file_path)
        opts = _VADOptions(
            min_speech_duration=min_speech_duration,
            min_silence_duration=min_silence_duration,
            prefix_padding_duration=prefix_padding_duration,
            activation_threshold=activation_threshold,
            deactivation_threshold=(
                deactivation_threshold
                if deactivation_threshold is not None
                else max(activation_threshold - 0.15, 0.01)
            ),
            sample_rate=sample_rate,
        )
        return cls(session=session, opts=opts)

    @classmethod
    def for_phone_call(cls, **overrides) -> "SileroVAD":
        """Convenience factory for telephony pipelines.

        Identical to :meth:`load` but pins ``sample_rate`` to 16000 Hz
        — the only sample rate Patter's pipeline-mode audio bus uses
        (8 kHz mulaw from Twilio is upsampled to 16 kHz PCM before
        reaching the VAD). Every other parameter mirrors the upstream
        Silero VAD defaults from ``snakers4/silero-vad``
        (``get_speech_timestamps`` / ``VADIterator``):

          - ``activation_threshold = 0.5`` — upstream ``threshold``
          - ``deactivation_threshold = 0.35`` — upstream
            ``neg_threshold = threshold - 0.15``
          - ``min_speech_duration = 0.25`` — upstream
            ``min_speech_duration_ms = 250``
          - ``min_silence_duration = 0.1`` — upstream
            ``min_silence_duration_ms = 100``
          - ``prefix_padding_duration = 0.03`` — upstream
            ``speech_pad_ms = 30``

        Override any field via keyword arguments. Deployments that
        experience truncation on natural pauses can raise
        ``min_silence_duration`` (e.g. 0.5–1.0 s) per call site rather
        than as a global default.

        Example::

            vad = await asyncio.to_thread(SileroVAD.for_phone_call)
            # or, if natural-pause truncation is observed:
            vad = await asyncio.to_thread(
                SileroVAD.for_phone_call, min_silence_duration=0.5
            )
        """
        defaults: dict = {
            "sample_rate": SileroSampleRate.HZ_16000,
            # Telephony bumps the activation threshold from the upstream
            # 0.5 → 0.8 (with deactivation 0.65) so background voices
            # and low-volume audio in the caller's room don't trip
            # barge-in. Near-mic speech typically scores 0.85-0.98 on
            # Silero — above 0.8 — while a distant second speaker
            # through a phone's noise-suppression pipeline lands around
            # 0.4-0.6 and is now correctly ignored. Bumped twice during
            # 2026-05-20 acceptance: first 0.5 → 0.7 (still triggered on
            # quiet voices), then 0.7 → 0.8. Trade-off: a whispered
            # legitimate input may not trigger; typical phone-call
            # speakers are unaffected. Pass an explicit
            # ``activation_threshold`` to override per call site.
            "activation_threshold": 0.8,
            "deactivation_threshold": 0.65,
        }
        defaults.update(overrides)
        return cls.load(**defaults)

    def __init__(
        self,
        *,
        session: "onnxruntime.InferenceSession",
        opts: _VADOptions,
    ) -> None:
        self._opts = opts
        self._onnx_session = session
        self._model = OnnxModel(onnx_session=session, sample_rate=opts.sample_rate)
        self._exp_filter = _ExpFilter(alpha=0.35)

        # Streaming state
        self._pending = np.zeros(0, dtype=np.float32)
        self._pub_speaking = False
        self._speech_threshold_duration = 0.0
        self._silence_threshold_duration = 0.0
        self._pub_current_sample = 0
        # Transitions beyond the first in a single process_frame call
        # (drained one per call so none are lost).
        self._pending_transitions: list = []
        self._pub_timestamp = 0.0
        self._closed = False

    @property
    def model(self) -> str:
        """Identifier of the underlying VAD model (``silero``)."""
        return SileroVADProviderTag.MODEL.value

    @property
    def provider(self) -> str:
        """Identifier of the provider tag (``silero``)."""
        return SileroVADProviderTag.PROVIDER.value

    @property
    def sample_rate(self) -> int:
        """Configured inference sample rate (8000 or 16000 Hz)."""
        return self._opts.sample_rate

    async def process_frame(
        self, pcm_chunk: bytes, sample_rate: int
    ) -> VADEvent | None:
        """Process a raw mono int16 little-endian PCM chunk.

        Returns a :class:`~getpatter.providers.base.VADEvent` only on state
        transitions (``speech_start`` / ``speech_end``). Between transitions
        this method returns ``None`` while buffering audio and running
        inference; callers that need per-frame probabilities should subclass
        and observe the internal state.
        """
        if self._closed:
            raise RuntimeError("SileroVAD is closed")

        if sample_rate != self._opts.sample_rate:
            raise ValueError(
                f"input sample_rate {sample_rate} does not match model "
                f"sample_rate {self._opts.sample_rate}; resampling is not "
                "implemented in the Patter port"
            )

        if not pcm_chunk:
            return None

        # int16 LE PCM -> float32 in [-1.0, 1.0]
        samples_i16 = np.frombuffer(pcm_chunk, dtype=np.int16)
        if samples_i16.size == 0:
            return None
        samples_f32 = samples_i16.astype(np.float32) / float(np.iinfo(np.int16).max)

        # Append to pending buffer
        self._pending = np.concatenate([self._pending, samples_f32])

        window_size = self._model.window_size_samples
        loop = asyncio.get_running_loop()
        # Drain a transition queued by a previous multi-window chunk first.
        if self._pending_transitions:
            return self._pending_transitions.pop(0)
        event: VADEvent | None = None

        while self._pending.shape[0] >= window_size:
            window = self._pending[:window_size].copy()
            self._pending = self._pending[window_size:]

            start_time = time.perf_counter()
            raw_p = await loop.run_in_executor(
                None, _run_inference, self._model, window
            )
            p = self._exp_filter.apply(exp=1.0, sample=raw_p)

            inference_duration = time.perf_counter() - start_time
            window_duration = window_size / self._opts.sample_rate
            if inference_duration > SLOW_INFERENCE_THRESHOLD:
                logger.warning(
                    "Silero VAD inference slower than realtime",
                    extra={"inference_duration": inference_duration},
                )

            self._pub_current_sample += window_size
            self._pub_timestamp += window_duration

            transition = self._advance_state(p, window_duration)
            if transition is not None:
                if event is None:
                    event = transition
                else:
                    # Queue later transitions instead of dropping them: a
                    # chunk spanning speech_end → speech_start lost the
                    # start event entirely (the comment claimed the
                    # opposite). The next process_frame call drains the
                    # queue first. Mirrors the TS adapter's pending queue.
                    self._pending_transitions.append(transition)

        return event

    def _advance_state(self, p: float, window_duration: float) -> VADEvent | None:
        """Update internal speaking state based on filtered probability.

        Returns a VADEvent on transitions, None otherwise.
        """
        opts = self._opts
        if p >= opts.activation_threshold or (
            self._pub_speaking and p > opts.deactivation_threshold
        ):
            self._speech_threshold_duration += window_duration
            self._silence_threshold_duration = 0.0

            if not self._pub_speaking:
                if self._speech_threshold_duration >= opts.min_speech_duration:
                    self._pub_speaking = True
                    return VADEvent(
                        type=SileroVADEventType.SPEECH_START.value,
                        confidence=float(p),
                        duration_ms=self._speech_threshold_duration * 1000.0,
                    )
        else:
            self._silence_threshold_duration += window_duration
            self._speech_threshold_duration = 0.0

            if (
                self._pub_speaking
                and self._silence_threshold_duration >= opts.min_silence_duration
            ):
                self._pub_speaking = False
                return VADEvent(
                    type=SileroVADEventType.SPEECH_END.value,
                    confidence=float(p),
                    duration_ms=self._silence_threshold_duration * 1000.0,
                )

        return None

    async def close(self) -> None:
        """Release the ONNX session. Idempotent."""
        if self._closed:
            return
        self._closed = True
        # onnxruntime sessions don't expose an explicit close; drop refs
        # so the session can be garbage collected.
        self._onnx_session = None  # type: ignore[assignment]
        self._model = None  # type: ignore[assignment]

    def reset(self) -> None:
        """Reset all per-utterance state so the next ``process_frame`` starts
        from a clean SILENCE state.

        Called by the stream handler between agent turns to prevent a "stuck
        SPEECH" condition where PSTN echo / loopback kept the detector's
        probability above ``deactivation_threshold`` for the entire agent
        turn. Without this reset the next user utterance would never
        trigger a SILENCE→SPEECH transition and barge-in would feel
        "one-shot" (works once, then never again until the call ends).

        Safe to call any time including on a closed instance (no-op).
        """
        if self._closed:
            return
        self._pending = np.zeros(0, dtype=np.float32)
        self._pending_transitions.clear()
        self._pub_speaking = False
        self._speech_threshold_duration = 0.0
        self._silence_threshold_duration = 0.0
        self._exp_filter.reset()
        if self._model is not None:
            self._model.reset()


def _run_inference(model: OnnxModel, window: np.ndarray) -> float:
    """Blocking inference entrypoint executed in an executor thread."""
    return model(window)
