Trixy7/tests/test_wakeword.py

# -*- coding: utf-8 -*-
"""
Tests für Wakeword Detection Module.
"""

import pytest
from datetime import datetime, timedelta
from unittest.mock import MagicMock, AsyncMock, patch

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

from trixy_core.wakeword.detector import (
    WakewordDetector,
    WakewordDetection,
    DetectorConfig,
    WakewordType,
)
from trixy_core.wakeword.audio_buffer import (
    AudioBuffer,
    AudioChunk,
    BufferConfig,
    AudioBufferPool,
)
from trixy_core.wakeword.vad import (
    VoiceActivityDetector,
    VADConfig,
    VADState,
    SpeechSegment,
)


class TestDetectorConfig:
    """Tests für DetectorConfig."""

    def test_default_config(self):
        """Testet Standard-Konfiguration."""
        config = DetectorConfig()
        assert config.threshold == 0.5
        assert config.sample_rate == 16000
        assert config.frame_length_ms == 80
        assert config.use_onnx is True

    def test_frame_length_samples(self):
        """Testet Frame-Länge in Samples."""
        config = DetectorConfig(sample_rate=16000, frame_length_ms=80)
        assert config.frame_length_samples == 1280  # 16000 * 80 / 1000

    def test_threshold_per_model(self):
        """Testet model-spezifische Thresholds."""
        config = DetectorConfig(
            threshold=0.5,
            threshold_per_model={"custom": 0.7},
        )
        assert config.get_threshold("custom") == 0.7
        assert config.get_threshold("other") == 0.5


class TestWakewordDetection:
    """Tests für WakewordDetection."""

    def test_detection_creation(self):
        """Testet Detection-Erstellung."""
        detection = WakewordDetection(
            wakeword_type=WakewordType.CUSTOM,
            model_name="hey_jarvis",
            confidence=0.95,
            timestamp=datetime.now(),
            audio_level=0.5,
        )
        assert detection.model_name == "hey_jarvis"
        assert detection.confidence == 0.95
        assert detection.wakeword_type == WakewordType.CUSTOM

    def test_to_dict(self):
        """Testet Dictionary-Konvertierung."""
        detection = WakewordDetection(
            wakeword_type=WakewordType.SYSTEM_COMMAND,
            model_name="test",
            confidence=0.8,
            timestamp=datetime.now(),
            audio_level=0.3,
        )
        d = detection.to_dict()
        assert d["wakeword_type"] == "system_command"
        assert d["confidence"] == 0.8


class TestWakewordType:
    """Tests für WakewordType."""

    def test_types(self):
        """Testet Wakeword-Typen."""
        assert WakewordType.CUSTOM.value == "custom"
        assert WakewordType.SYSTEM_COMMAND.value == "system_command"
        assert WakewordType.UNKNOWN.value == "unknown"


class TestWakewordDetector:
    """Tests für WakewordDetector (Mock-basiert)."""

    @pytest.fixture
    def config(self):
        """Erstellt Test-Konfiguration."""
        return DetectorConfig(
            model_directory="models/wakeword",
            models=["custom"],
            threshold=0.5,
        )

    @pytest.fixture
    def detector(self, config):
        """Erstellt Detector."""
        return WakewordDetector(config)

    def test_detector_creation(self, detector):
        """Testet Detector-Erstellung."""
        assert detector.config is not None
        assert not detector.is_loaded
        assert detector.loaded_models == []

    def test_stats(self, detector):
        """Testet Statistiken."""
        stats = detector.stats
        assert stats["frames_processed"] == 0
        assert stats["detections"] == 0
        assert stats["is_loaded"] is False

    def test_callback_registration(self, detector):
        """Testet Callback-Registrierung."""
        callback = MagicMock()
        detector.on_detection(callback)
        assert callback in detector._on_detection_callbacks

        assert detector.remove_callback(callback)
        assert callback not in detector._on_detection_callbacks

    def test_reset(self, detector):
        """Testet Reset."""
        detector._chunk_buffer = [b"test"]
        detector._last_detection_time = datetime.now()

        detector.reset()

        assert detector._chunk_buffer == []
        assert detector._last_detection_time is None


class TestBufferConfig:
    """Tests für BufferConfig."""

    def test_default_config(self):
        """Testet Standard-Konfiguration."""
        config = BufferConfig()
        assert config.max_duration_seconds == 60.0
        assert config.sample_rate == 16000
        assert config.bit_depth == 16

    def test_bytes_per_second(self):
        """Testet Bytes pro Sekunde."""
        config = BufferConfig(sample_rate=16000, channels=1, bit_depth=16)
        assert config.bytes_per_second == 32000  # 16000 * 1 * 2

    def test_max_buffer_size(self):
        """Testet maximale Buffer-Größe."""
        config = BufferConfig(max_duration_seconds=60.0, sample_rate=16000)
        assert config.max_buffer_size == 60 * 32000


class TestAudioBuffer:
    """Tests für AudioBuffer."""

    @pytest.fixture
    def config(self):
        """Erstellt Test-Konfiguration."""
        return BufferConfig(
            max_duration_seconds=10.0,
            pre_buffer_seconds=0.5,
        )

    @pytest.fixture
    def buffer(self, config):
        """Erstellt AudioBuffer."""
        return AudioBuffer(config)

    def test_initial_state(self, buffer):
        """Testet initialen Zustand."""
        assert not buffer.is_recording
        assert buffer.size_bytes == 0
        assert buffer.chunk_count == 0

    def test_pre_buffer(self, buffer):
        """Testet Pre-Buffer."""
        audio = b"\x00" * 1000
        buffer.add_to_pre_buffer(audio)

        stats = buffer.get_stats()
        assert stats["pre_buffer_bytes"] == 1000

    def test_start_recording(self, buffer):
        """Testet Aufnahme-Start."""
        buffer.add_to_pre_buffer(b"\x00" * 500)
        buffer.start_recording(include_pre_buffer=True)

        assert buffer.is_recording
        assert buffer.size_bytes == 500

    def test_add_chunk(self, buffer):
        """Testet Chunk-Hinzufügen."""
        buffer.start_recording()
        audio = b"\x00" * 1000

        assert buffer.add_chunk(audio)
        assert buffer.size_bytes == 1000
        assert buffer.chunk_count == 1

    def test_stop_recording(self, buffer):
        """Testet Aufnahme-Stop."""
        buffer.start_recording()
        buffer.add_chunk(b"\x00" * 1000)
        buffer.add_chunk(b"\x01" * 500)

        data = buffer.stop_recording()
        assert len(data) == 1500
        assert not buffer.is_recording

    def test_max_size_limit(self, buffer):
        """Testet Größen-Limit."""
        buffer.start_recording()

        # Fülle Buffer bis zum Limit
        chunk_size = buffer.config.max_buffer_size + 1
        result = buffer.add_chunk(b"\x00" * chunk_size)

        assert not result  # Sollte fehlschlagen

    def test_clear(self, buffer):
        """Testet Löschen."""
        buffer.start_recording()
        buffer.add_chunk(b"\x00" * 1000)

        buffer.clear()

        assert buffer.size_bytes == 0
        assert not buffer.is_recording


class TestAudioBufferPool:
    """Tests für AudioBufferPool."""

    @pytest.fixture
    def pool(self):
        """Erstellt Buffer-Pool."""
        return AudioBufferPool(pool_size=3)

    def test_acquire_release(self, pool):
        """Testet Acquire/Release."""
        buffer = pool.acquire("test-1")
        assert buffer is not None
        assert pool.active_count == 1

        data = pool.release("test-1")
        assert pool.active_count == 0
        assert pool.available_count >= 3

    def test_multiple_acquire(self, pool):
        """Testet mehrfaches Acquire."""
        b1 = pool.acquire("test-1")
        b2 = pool.acquire("test-2")

        assert b1 is not b2
        assert pool.active_count == 2

    def test_get_same_buffer(self, pool):
        """Testet Wiederverwendung bei gleicher ID."""
        b1 = pool.acquire("test-1")
        b2 = pool.acquire("test-1")

        assert b1 is b2


class TestVADConfig:
    """Tests für VADConfig."""

    def test_default_config(self):
        """Testet Standard-Konfiguration."""
        config = VADConfig()
        assert config.speech_threshold == 0.02
        assert config.silence_duration_ms == 3000
        assert config.max_duration_ms == 60000


class TestVADState:
    """Tests für VADState."""

    def test_states(self):
        """Testet VAD-Zustände."""
        assert VADState.WAITING.value == "waiting"
        assert VADState.SPEECH.value == "speech"
        assert VADState.SILENCE.value == "silence"
        assert VADState.TIMEOUT.value == "timeout"


@pytest.mark.skipif(not HAS_NUMPY, reason="numpy nicht installiert")
class TestVoiceActivityDetector:
    """Tests für VoiceActivityDetector."""

    @pytest.fixture
    def config(self):
        """Erstellt Test-Konfiguration."""
        return VADConfig(
            speech_threshold=0.02,
            silence_threshold=0.01,
            silence_duration_ms=500,  # Kurz für Tests
            max_duration_ms=5000,
            min_speech_ms=100,
        )

    @pytest.fixture
    def vad(self, config):
        """Erstellt VAD."""
        return VoiceActivityDetector(config)

    def test_initial_state(self, vad):
        """Testet initialen Zustand."""
        assert vad.state == VADState.WAITING
        assert not vad.is_active
        assert not vad.has_speech

    def test_start(self, vad):
        """Testet Start."""
        vad.start()
        assert vad.is_active
        assert vad.state == VADState.WAITING

    def test_stop(self, vad):
        """Testet Stop."""
        vad.start()
        vad.stop()
        assert not vad.is_active

    def test_speech_detection(self, vad):
        """Testet Sprach-Erkennung."""
        vad.start()

        # Simuliere lautes Audio (Sprache)
        loud_audio = np.array([10000] * 1280, dtype=np.int16)
        state = vad.process_frame(loud_audio.tobytes())

        assert state == VADState.SPEECH

    def test_silence_detection(self, vad):
        """Testet Stille-Erkennung."""
        vad.start()

        # Erst Sprache (mehrere Frames für min_speech_ms)
        loud = np.array([10000] * 1280, dtype=np.int16)
        for _ in range(5):  # 400ms Sprache
            vad.process_frame(loud.tobytes())

        # Dann Stille - muss unter silence_threshold sein
        quiet = np.array([1] * 1280, dtype=np.int16)  # Sehr leise
        state = vad.process_frame(quiet.tobytes())

        # Nach nur einem Frame Stille sind wir noch in SPEECH→SILENCE Übergang
        assert state in (VADState.SPEECH, VADState.SILENCE)

    def test_callbacks(self, vad):
        """Testet Callbacks."""
        speech_started = []

        vad.on_speech_start(lambda: speech_started.append(True))

        vad.start()

        # Sprache simulieren - lauter als speech_threshold (0.02)
        # RMS von 15000/32768 = 0.46, deutlich über 0.02
        loud = np.array([15000] * 1280, dtype=np.int16)

        # Genug Frames für min_speech_ms (100ms = 2 Frames bei 80ms)
        for _ in range(10):  # 800ms sollte reichen
            state = vad.process_frame(loud.tobytes())

        # Sollte im SPEECH-Zustand sein
        assert state == VADState.SPEECH
        # _speech_start_time sollte gesetzt sein
        assert vad._speech_start_time is not None

    def test_stats(self, vad):
        """Testet Statistiken."""
        vad.start()

        audio = np.array([5000] * 1280, dtype=np.int16)
        vad.process_frame(audio.tobytes())

        stats = vad.get_stats()
        assert stats["is_active"] is True
        assert stats["frame_count"] == 1

    def test_reset(self, vad):
        """Testet Reset."""
        vad.start()
        vad._speech_detected = True

        vad.reset()

        assert vad.state == VADState.WAITING
        assert not vad.has_speech


class TestSpeechSegment:
    """Tests für SpeechSegment."""

    def test_segment_creation(self):
        """Testet Segment-Erstellung."""
        segment = SpeechSegment(
            start_time=datetime.now(),
            peak_level=0.5,
            avg_level=0.3,
        )
        assert segment.peak_level == 0.5
        assert not segment.is_complete

    def test_duration(self):
        """Testet Dauer-Berechnung."""
        start = datetime.now() - timedelta(seconds=2)
        segment = SpeechSegment(start_time=start)

        assert segment.duration_ms >= 2000

    def test_complete_segment(self):
        """Testet abgeschlossenes Segment."""
        segment = SpeechSegment(
            start_time=datetime.now() - timedelta(seconds=1),
            end_time=datetime.now(),
        )
        assert segment.is_complete