undefined

---

undefined

def test_detection_accuracy_threshold(self):
    """Test that detector respects confidence threshold."""
    from wake_word import SecureWakeWordDetector

    detector = SecureWakeWordDetector(threshold=0.7)
    callback = Mock()
    test_audio = np.random.randn(16000).astype(np.float32)

    with patch.object(detector.model, 'predict') as mock_predict:
        # Below threshold - should not trigger
        mock_predict.return_value = {"hey_jarvis": np.array([0.5])}
        detector._test_process(test_audio, callback)
        callback.assert_not_called()

        # Above threshold - should trigger
        mock_predict.return_value = {"hey_jarvis": np.array([0.8])}
        detector._test_process(test_audio, callback)
        callback.assert_called_once()

def test_buffer_cleared_after_detection(self):
    """Test privacy: buffer cleared immediately after detection."""
    from wake_word import SecureWakeWordDetector

    detector = SecureWakeWordDetector()
    detector.audio_buffer.extend(np.zeros(16000))

    with patch.object(detector.model, 'predict') as mock_predict:
        mock_predict.return_value = {"hey_jarvis": np.array([0.9])}
        detector._process_audio()

    assert len(detector.audio_buffer) == 0, "Buffer must be cleared"

def test_cpu_usage_under_threshold(self):
    """Test CPU usage stays under 5%."""
    import psutil
    import time
    from wake_word import SecureWakeWordDetector

    detector = SecureWakeWordDetector()
    process = psutil.Process()
    start_time = time.time()

    while time.time() - start_time < 10:
        audio = np.random.randn(1600).astype(np.float32)
        detector.audio_buffer.extend(audio)
        if len(detector.audio_buffer) >= 16000:
            detector._process_audio()

    avg_cpu = process.cpu_percent() / psutil.cpu_count()
    assert avg_cpu < 5, f"CPU usage too high: {avg_cpu}%"

def test_memory_footprint(self):
    """Test memory usage stays under 100MB."""
    import tracemalloc
    from wake_word import SecureWakeWordDetector

    tracemalloc.start()
    detector = SecureWakeWordDetector()

    for _ in range(600):
        audio = np.random.randn(1600).astype(np.float32)
        detector.audio_buffer.extend(audio)

    current, peak = tracemalloc.get_traced_memory()
    tracemalloc.stop()

    peak_mb = peak / 1024 / 1024
    assert peak_mb < 100, f"Memory too high: {peak_mb}MB"

undefined

class SecureWakeWordDetector:
    def __init__(self, threshold=0.5):
        self.threshold = threshold
        self.model = Model(wakeword_models=["hey_jarvis"])
        self.audio_buffer = deque(maxlen=24000)

    def _test_process(self, audio, callback):
        predictions = self.model.predict(audio)
        for model_name, scores in predictions.items():
            if np.max(scores) > self.threshold:
                self.audio_buffer.clear()
                callback(model_name, np.max(scores))
                break

pytest tests/test_wake_word.py -v
pytest --cov=wake_word --cov-report=term-missing

from openwakeword.model import Model
import numpy as np
import sounddevice as sd
from collections import deque
import structlog

logger = structlog.get_logger()

class SecureWakeWordDetector:
    """Privacy-preserving wake word detection."""

    def __init__(self, model_path: str = None, threshold: float = 0.5, sample_rate: int = 16000):
        if model_path:
            self.model = Model(wakeword_models=[model_path])
        else:
            self.model = Model(wakeword_models=["hey_jarvis"])

        self.threshold = threshold
        self.sample_rate = sample_rate
        self.buffer_size = int(sample_rate * 1.5)
        self.audio_buffer = deque(maxlen=self.buffer_size)
        self.is_listening = False
        self.on_wake = None

    def start(self, callback):
        """Start listening for wake word."""
        self.on_wake = callback
        self.is_listening = True

        def audio_callback(indata, frames, time, status):
            if not self.is_listening:
                return
            audio = indata[:, 0] if len(indata.shape) > 1 else indata
            self.audio_buffer.extend(audio)
            if len(self.audio_buffer) >= self.sample_rate:
                self._process_audio()

        self.stream = sd.InputStream(
            samplerate=self.sample_rate, channels=1, dtype=np.float32,
            callback=audio_callback, blocksize=int(self.sample_rate * 0.1)
        )
        self.stream.start()

    def _process_audio(self):
        """Process audio buffer for wake word."""
        audio = np.array(list(self.audio_buffer))
        predictions = self.model.predict(audio)

        for model_name, scores in predictions.items():
            if np.max(scores) > self.threshold:
                self.audio_buffer.clear()  # Privacy: clear immediately
                if self.on_wake:
                    self.on_wake(model_name, np.max(scores))
                break

    def stop(self):
        """Stop listening."""
        self.is_listening = False
        if hasattr(self, 'stream'):
            self.stream.stop()
            self.stream.close()
        self.audio_buffer.clear()

class RobustDetector:
    """Reduce false positives with confirmation."""

    def __init__(self, detector: SecureWakeWordDetector):
        self.detector = detector
        self.detection_history = []
        self.confirmation_window = 2.0
        self.min_confirmations = 2

    def on_potential_wake(self, model: str, confidence: float):
        now = time.time()
        self.detection_history.append({"time": now, "confidence": confidence})
        self.detection_history = [d for d in self.detection_history if now - d["time"] < self.confirmation_window]

        if len(self.detection_history) >= self.min_confirmations:
            avg_confidence = np.mean([d["confidence"] for d in self.detection_history])
            if avg_confidence > 0.6:
                self.detection_history.clear()
                return True
        return False

undefined

undefined

undefined

def append(self, audio: np.ndarray):
    n = len(audio)
    end_idx = (self.write_idx + n) % self.size
    if end_idx > self.write_idx:
        self.buffer[self.write_idx:end_idx] = audio
    else:
        self.buffer[self.write_idx:] = audio[:self.size - self.write_idx]
        self.buffer[:end_idx] = audio[self.size - self.write_idx:]
    self.write_idx = end_idx

undefined

undefined

def process(self, audio: np.ndarray):
    audio_int16 = (audio * 32767).astype(np.int16)
    if not self.vad.is_speech(audio_int16.tobytes(), 16000):
        return None  # Skip expensive inference
    return self.detector._process_audio()

undefined

undefined

def add_window(self, audio: np.ndarray):
    self.pending_windows.append(audio)
    if len(self.pending_windows) >= self.batch_size:
        batch = np.stack(self.pending_windows)
        results = self.model.predict_batch(batch)
        self.pending_windows.clear()
        return results
    return None

undefined

undefined

---

class PrivacyController:
    """Ensure privacy in always-listening system."""

    def __init__(self):
        self.is_enabled = True
        self.last_activity = time.time()

    def check_privacy_mode(self) -> bool:
        if self._is_dnd_enabled():
            return False
        if time.time() - self.last_activity > 3600:
            return False
        return self.is_enabled

---

undefined

---