Wake Word Detection

Wake Word Detection Skill

1. Overview

Risk Level: MEDIUM - Continuous audio monitoring, privacy implications, resource constraints

You are an expert in wake word detection with deep expertise in openWakeWord, keyword spotting, and always-listening systems.

Primary Use Cases:

• JARVIS activation phrase detection ("Hey JARVIS")
• Always-listening with minimal resource usage
• Offline wake word detection (no cloud dependency)

---

2. Core Principles

• TDD First - Write tests before implementation code
• Performance Aware - Optimize for CPU, memory, and latency
• Privacy Preserving - Never store audio, minimize buffers
• Accuracy Focused - Minimize false positives/negatives
• Resource Efficient - Target <5% CPU, <100MB memory

---

3. Core Responsibilities

3.1 Privacy-First Monitoring

• Process locally - Never send audio to external services
• Buffer minimally - Only keep audio needed for detection
• Discard non-wake - Immediately discard non-wake audio
• User control - Easy disable/pause functionality

3.2 Efficiency Requirements

• Minimal CPU usage (<5% average)
• Low memory footprint (<100MB)
• Low latency detection (<500ms)
• Low false positive rate (<1 per hour)

---

4. Technical Foundation

# requirements.txt
openwakeword>=0.6.0
numpy>=1.24.0
sounddevice>=0.4.6
onnxruntime>=1.16.0

---

5. Implementation Workflow (TDD)

Step 1: Write Failing Test First

# tests/test_wake_word.py
import pytest
import numpy as np
from unittest.mock import Mock, patch

class TestWakeWordDetector:
    """TDD tests for wake word detection."""

    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"

Step 2: Implement Minimum to Pass

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

Step 3: Run Full Verification

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

---

6. Implementation Patterns

Pattern 1: Secure Wake Word Detector

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()

Pattern 2: False Positive Reduction

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

---

7. Performance Patterns

Pattern 1: Model Quantization

# Good - Use quantized ONNX model
import onnxruntime as ort

class QuantizedDetector:
    def __init__(self, model_path: str):
        sess_options = ort.SessionOptions()
        sess_options.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL
        self.session = ort.InferenceSession(model_path, sess_options, providers=['CPUExecutionProvider'])

# Bad - Full precision model
class SlowDetector:
    def __init__(self, model_path: str):
        self.session = ort.InferenceSession(model_path)  # No optimization

Pattern 2: Efficient Audio Buffering

# Good - Pre-allocated numpy buffer with circular indexing
class EfficientBuffer:
    def __init__(self, size: int):
        self.buffer = np.zeros(size, dtype=np.float32)
        self.write_idx = 0
        self.size = size

    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

# Bad - Individual appends
class SlowBuffer:
    def append(self, audio: np.ndarray):
        for sample in audio:  # Slow!
            self.buffer.append(sample)

Pattern 3: VAD Preprocessing

# Good - Skip inference on silence
import webrtcvad

class VADOptimizedDetector:
    def __init__(self):
        self.vad = webrtcvad.Vad(2)
        self.detector = SecureWakeWordDetector()

    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()

# Bad - Always run inference
class WastefulDetector:
    def process(self, audio: np.ndarray):
        return self.detector._process_audio()  # Even on silence

Pattern 4: Batch Inference

# Good - Process multiple windows in single inference
class BatchDetector:
    def __init__(self, batch_size: int = 4):
        self.batch_size = batch_size
        self.pending_windows = []

    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

Pattern 5: Memory-Mapped Models

# Good - Memory-map large model files
import mmap

class MmapModelLoader:
    def __init__(self, model_path: str):
        self.file = open(model_path, 'rb')
        self.mmap = mmap.mmap(self.file.fileno(), 0, access=mmap.ACCESS_READ)

# Bad - Load entire model into memory
class EagerModelLoader:
    def __init__(self, model_path: str):
        with open(model_path, 'rb') as f:
            self.model_data = f.read()  # Entire model in RAM

---

8. Security Standards

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

# Data minimization
MAX_BUFFER_SECONDS = 2.0
def on_wake_detected():
    audio_buffer.clear()  # Delete immediately

---

9. Common Mistakes

# BAD - Stores all audio
def on_audio(chunk):
    with open("audio.raw", "ab") as f:
        f.write(chunk)

# GOOD - Discard after processing
def on_audio(chunk):
    buffer.extend(chunk)
    process_buffer()

# BAD - Large buffer
buffer = deque(maxlen=sample_rate * 60)  # 1 minute!

# GOOD - Minimal buffer
buffer = deque(maxlen=sample_rate * 1.5)  # 1.5 seconds

---

10. Pre-Implementation Checklist

Phase 1: Before Writing Code

• [ ] Read TDD workflow section completely
• [ ] Set up test file with detection accuracy tests
• [ ] Define threshold and performance targets
• [ ] Identify which performance patterns apply
• [ ] Review privacy requirements

Phase 2: During Implementation

• [ ] Write failing test for each feature first
• [ ] Implement minimal code to pass test
• [ ] Apply performance patterns (VAD, quantization)
• [ ] Buffer size minimal (<2 seconds)
• [ ] Audio cleared after detection

Phase 3: Before Committing

• [ ] All tests pass: pytest tests/test_wake_word.py -v
• [ ] Coverage >80%: pytest --cov=wake_word
• [ ] False positive rate <1/hour tested
• [ ] CPU usage <5% measured
• [ ] Memory usage <100MB verified
• [ ] Audio never stored to disk

---

11. Summary

Your goal is to create wake word detection that is:

• Private: Audio processed locally, minimal retention
• Efficient: Low CPU (<5%), low memory (<100MB)
• Accurate: Low false positive rate (<1/hour)
• Test-Driven: All features have tests first

Critical Reminders: 1. Write tests before implementation 2. Never store audio to disk 3. Keep buffer minimal (<2 seconds) 4. Apply performance patterns (VAD, quantization)

Wake Word Detection Advanced Patterns

Adaptive Threshold

class AdaptiveDetector:
    """Adjust threshold based on environment."""

    def __init__(self, base_threshold: float = 0.5):
        self.base_threshold = base_threshold
        self.noise_level = 0.0
        self.false_positives = []

    def get_threshold(self) -> float:
        """Get current adaptive threshold."""
        # Increase threshold in noisy environment
        noise_factor = min(self.noise_level * 0.5, 0.3)

        # Increase after false positives
        fp_factor = len(self.false_positives) * 0.05

        return min(self.base_threshold + noise_factor + fp_factor, 0.9)

    def update_noise(self, audio: np.ndarray):
        """Update noise level estimate."""
        energy = np.mean(audio ** 2)
        self.noise_level = 0.9 * self.noise_level + 0.1 * energy

    def record_false_positive(self):
        """Record false positive for threshold adjustment."""
        self.false_positives.append(time.time())
        # Keep only recent
        hour_ago = time.time() - 3600
        self.false_positives = [t for t in self.false_positives if t > hour_ago]

Multi-Stage Detection

class TwoStageDetector:
    """Fast first stage, accurate second stage."""

    def __init__(self):
        # Fast, low accuracy model
        self.fast_model = Model(wakeword_models=["hey_jarvis_small"])
        # Slower, high accuracy model
        self.accurate_model = Model(wakeword_models=["hey_jarvis"])

    def detect(self, audio: np.ndarray) -> bool:
        # First stage: fast check
        fast_result = self.fast_model.predict(audio)
        if np.max(list(fast_result.values())[0]) < 0.3:
            return False  # Quick reject

        # Second stage: accurate check
        accurate_result = self.accurate_model.predict(audio)
        return np.max(list(accurate_result.values())[0]) > 0.6

Context-Aware Detection

class ContextAwareDetector:
    """Adjust detection based on context."""

    def __init__(self, detector: SecureWakeWordDetector):
        self.detector = detector
        self.context = "default"

    def set_context(self, context: str):
        """Set current context."""
        contexts = {
            "default": 0.5,
            "conversation": 0.7,  # Higher threshold during conversation
            "quiet": 0.4,  # Lower threshold in quiet environment
            "noisy": 0.6  # Higher threshold in noisy environment
        }
        self.detector.threshold = contexts.get(context, 0.5)
        self.context = context
        logger.info("context.updated", context=context)

Performance Monitoring

class DetectorMetrics:
    """Monitor wake word detector performance."""

    def __init__(self):
        self.detections = []
        self.cpu_usage = []
        self.latencies = []

    def record_detection(self, latency_ms: float):
        """Record detection event."""
        self.detections.append({
            "time": time.time(),
            "latency": latency_ms
        })

    def record_cpu(self, usage: float):
        """Record CPU usage."""
        self.cpu_usage.append({
            "time": time.time(),
            "usage": usage
        })

    def get_report(self) -> dict:
        """Get performance report."""
        return {
            "avg_latency_ms": np.mean([d["latency"] for d in self.detections]),
            "avg_cpu_percent": np.mean([c["usage"] for c in self.cpu_usage]),
            "detections_per_hour": len(self.detections)
        }

Embedded Optimization

class EmbeddedDetector:
    """Optimized for embedded systems."""

    def __init__(self):
        # Use quantized model
        self.model = Model(
            wakeword_models=["hey_jarvis"],
            inference_framework="onnx"  # Efficient runtime
        )

        # Reduce processing frequency
        self.process_every_n = 3
        self.frame_count = 0

    def on_audio(self, audio: np.ndarray):
        """Process with reduced frequency."""
        self.frame_count += 1

        # Only process every Nth frame
        if self.frame_count % self.process_every_n != 0:
            return

        # Quick energy check
        if np.mean(audio ** 2) < 0.0001:
            return  # Skip silence

        # Run detection
        self._detect(audio)

Wake Word Detection Security Examples

Privacy-Preserving Audio Handling

class PrivateAudioBuffer:
    """Buffer that never persists audio."""

    def __init__(self, max_seconds: float = 1.5, sample_rate: int = 16000):
        self.max_size = int(max_seconds * sample_rate)
        self._buffer = np.zeros(self.max_size, dtype=np.float32)
        self._position = 0

    def add(self, audio: np.ndarray):
        """Add audio to circular buffer."""
        n = len(audio)
        if n >= self.max_size:
            self._buffer[:] = audio[-self.max_size:]
            self._position = 0
        else:
            end_pos = self._position + n
            if end_pos <= self.max_size:
                self._buffer[self._position:end_pos] = audio
                self._position = end_pos
            else:
                first = self.max_size - self._position
                self._buffer[self._position:] = audio[:first]
                self._buffer[:n-first] = audio[first:]
                self._position = n - first

    def clear(self):
        """Securely clear buffer."""
        self._buffer.fill(0)
        self._position = 0

    def get(self) -> np.ndarray:
        """Get buffer contents."""
        return np.roll(self._buffer, -self._position)

User Consent Management

class ConsentManager:
    """Manage user consent for always-listening."""

    def __init__(self, config_path: str):
        self.config_path = Path(config_path)
        self._load_config()

    def _load_config(self):
        if self.config_path.exists():
            self.config = json.loads(self.config_path.read_text())
        else:
            self.config = {"listening_enabled": False}

    def is_listening_allowed(self) -> bool:
        return self.config.get("listening_enabled", False)

    def set_listening(self, enabled: bool):
        self.config["listening_enabled"] = enabled
        self.config["last_modified"] = datetime.now().isoformat()
        self._save_config()
        logger.info("consent.updated", enabled=enabled)

    def _save_config(self):
        self.config_path.write_text(json.dumps(self.config))

False Positive Logging

class DetectionLogger:
    """Log detections without audio content."""

    def log_detection(self, model: str, confidence: float, was_false_positive: bool = False):
        """Log detection event."""
        logger.info("wake_word.detection",
                   model=model,
                   confidence=confidence,
                   false_positive=was_false_positive,
                   # Never log audio content
                   timestamp=datetime.now().isoformat())

    def log_stats(self, period_hours: int = 24):
        """Log detection statistics."""
        # Query from metrics store
        stats = self._get_stats(period_hours)
        logger.info("wake_word.stats",
                   total_detections=stats["total"],
                   false_positives=stats["false_positives"],
                   false_positive_rate=stats["fpr"])

Security Testing

def test_audio_not_stored():
    """Verify audio is never written to disk."""
    import tempfile
    import os

    temp_dir = tempfile.mkdtemp()
    detector = SecureWakeWordDetector()

    # Run detector
    detector.start(lambda m, c: None)
    time.sleep(5)
    detector.stop()

    # Check no audio files created
    for root, dirs, files in os.walk(temp_dir):
        for f in files:
            assert not f.endswith(('.wav', '.raw', '.pcm'))

def test_buffer_cleared_on_detection():
    """Test audio buffer is cleared after detection."""
    detector = SecureWakeWordDetector()

    # Simulate audio and detection
    detector.audio_buffer.extend([0.1] * 16000)
    detector._process_audio()  # Triggers detection

    assert len(detector.audio_buffer) == 0

def test_max_buffer_size():
    """Ensure buffer never exceeds maximum."""
    detector = SecureWakeWordDetector()

    # Add more than max
    for _ in range(100):
        detector.audio_buffer.extend([0.1] * 1600)

    assert len(detector.audio_buffer) <= detector.buffer_size