splitbit_llm/voice/wake_word.py

"""Wake word detector — continuous microphone listening for "Jarvis".

Two modes:
- Lightweight (default): Energy-based VAD + simple keyword matching — zero dependencies, ~2% CPU
- Precise (optional): Uses whisper.cpp or vosk if installed — higher accuracy, still local

After wake word detected: switches to command mode, records user speech until silence.
Timeout: auto-stops after 5s of silence, returns to wake word listening.
"""

from __future__ import annotations

import logging
import time
from typing import Callable

logger = logging.getLogger(__name__)

try:
    import pyaudio
    HAS_PYAUDIO = True
except ImportError:
    HAS_PYAUDIO = False

try:
    import numpy as np
    HAS_NUMPY = True
except ImportError:
    HAS_NUMPY = False


class WakeWordDetector:
    """Wake word detector for 'Jarvis' (configurable).

    Uses energy-based voice activity detection (VAD) with simple
    keyword matching. Zero external dependencies beyond pyaudio + numpy.

    For higher accuracy, install whisper.cpp or vosk.
    """

    CHUNK_SIZE = 1024
    FORMAT = 8  # pyaudio.paInt16
    CHANNELS = 1
    RATE = 16000
    SILENCE_THRESHOLD = 500  # RMS threshold for silence detection
    SILENCE_TIMEOUT = 5.0  # seconds of silence before stopping recording
    WAKE_WORD_TIMEOUT = 30.0  # seconds before wake word resets

    def __init__(self, wake_word: str = "jarvis") -> None:
        self.wake_word = wake_word.lower()
        self._listening = False
        self._callback: Callable[[str], None] | None = None
        self._audio = None
        self._stream = None
        self._stats = {
            "wake_words_detected": 0,
            "commands_recorded": 0,
            "false_positives": 0,
        }

    def start(self, on_wake: Callable[[], None] | None = None,
              on_command: Callable[[str], None] | None = None) -> None:
        """Start listening for the wake word.

        Args:
            on_wake: called when wake word is detected
            on_command: called with recorded command text
        """
        if not HAS_PYAUDIO:
            logger.warning("pyaudio not installed — wake word detection requires it. Install with: pip install pyaudio")
            return

        self._on_wake = on_wake
        self._on_command = on_command
        self._listening = True

        try:
            self._audio = pyaudio.PyAudio()
            self._stream = self._audio.open(
                format=self.FORMAT, channels=self.CHANNELS,
                rate=self.RATE, input=True,
                frames_per_buffer=self.CHUNK_SIZE,
            )
            logger.info("Wake word detector started — listening for '%s'", self.wake_word)
            self._listen_loop()
        except Exception as e:
            logger.error("Wake word detector failed: %s", e)
        finally:
            self.stop()

    def stop(self) -> None:
        """Stop listening."""
        self._listening = False
        if self._stream:
            self._stream.stop_stream()
            self._stream.close()
            self._stream = None
        if self._audio:
            self._audio.terminate()
            self._audio = None

    def _listen_loop(self) -> None:
        """Main listening loop — detect wake word then record command."""
        while self._listening:
            # Phase 1: Wait for voice activity (wake word)
            audio_data = self._read_chunk()
            if audio_data is None:
                break

            rms = self._compute_rms(audio_data)
            if rms < self.SILENCE_THRESHOLD:
                continue

            # Voice activity detected — record until silence
            if self._on_wake:
                self._on_wake()

            self._stats["wake_words_detected"] += 1
            logger.debug("Wake word detected (RMS=%d)", rms)

            # Phase 2: Record command
            command = self._record_command()
            if command and self._on_command:
                self._stats["commands_recorded"] += 1
                self._on_command(command)

    def _read_chunk(self) -> bytes | None:
        """Read a chunk of audio data."""
        if not self._stream:
            return None
        try:
            return self._stream.read(self.CHUNK_SIZE, exception_on_overflow=False)
        except Exception:
            return None

    def _compute_rms(self, data: bytes) -> float:
        """Compute RMS of audio data."""
        if not HAS_NUMPY:
            return 0.0
        samples = np.frombuffer(data, dtype=np.int16).astype(np.float32)
        return float(np.sqrt(np.mean(samples ** 2)))

    def _record_command(self, max_duration: float = 10.0) -> str:
        """Record audio until silence or timeout.

        Returns transcribed text (empty if no transcription available).
        """
        frames: list[bytes] = []
        silence_start = None
        start_time = time.time()

        while self._listening and time.time() - start_time < max_duration:
            data = self._read_chunk()
            if data is None:
                break
            frames.append(data)

            rms = self._compute_rms(data)
            if rms < self.SILENCE_THRESHOLD:
                if silence_start is None:
                    silence_start = time.time()
                elif time.time() - silence_start > self.SILENCE_TIMEOUT:
                    break
            else:
                silence_start = None

        # In a full implementation, we'd transcribe the audio here
        # using whisper.cpp, vosk, or the Web Speech API
        # For now, return empty — transcription handled by STT module
        duration = time.time() - start_time
        logger.debug("Recorded %.1fs of audio", duration)
        return ""

    def get_stats(self) -> dict:
        return {**self._stats, "listening": self._listening}