""" Example usage and testing utilities for the wakeword detection system. This module provides comprehensive examples, testing utilities, and demonstration code for the Trixy wakeword detection system. """ import numpy as np import torch import time import json from pathlib import Path from typing import Optional, List, Dict, Any import logging import threading from .config import ( WakewordConfig, create_default_config, create_lightweight_config, create_high_performance_config ) from .detector import WakewordDetector, create_wakeword_detector from .detection_engine import DetectionResult, WakewordType from .audio_buffer import AudioChunk from .optimization import PerformanceMonitor, ModelOptimizer class MockEventHandler: """Mock event handler for testing purposes.""" def __init__(self): self.triggered_events = [] self.event_callbacks = {} def trigger(self, event_name: str, event_data: Any): """Trigger an event.""" self.triggered_events.append({ 'event_name': event_name, 'event_data': event_data, 'timestamp': time.time() }) # Call callbacks if registered if event_name in self.event_callbacks: for callback in self.event_callbacks[event_name]: try: callback(event_name, event_data) except Exception as e: print(f"Error in event callback: {e}") def register_callback(self, event_name: str, callback): """Register event callback.""" if event_name not in self.event_callbacks: self.event_callbacks[event_name] = [] self.event_callbacks[event_name].append(callback) def get_triggered_events(self, event_name: Optional[str] = None) -> List[Dict]: """Get triggered events.""" if event_name: return [e for e in self.triggered_events if e['event_name'] == event_name] return self.triggered_events def clear_events(self): """Clear event history.""" self.triggered_events.clear() class AudioSimulator: """Simulate audio input for testing wakeword detection.""" def __init__(self, sample_rate: int = 16000): self.sample_rate = sample_rate def generate_silence(self, duration_seconds: float) -> np.ndarray: """Generate silence audio.""" samples = int(duration_seconds * self.sample_rate) return np.zeros(samples, dtype=np.float32) def generate_white_noise(self, duration_seconds: float, amplitude: float = 0.1) -> np.ndarray: """Generate white noise audio.""" samples = int(duration_seconds * self.sample_rate) return np.random.normal(0, amplitude, samples).astype(np.float32) def generate_sine_wave(self, duration_seconds: float, frequency: float = 440.0, amplitude: float = 0.5) -> np.ndarray: """Generate sine wave audio.""" samples = int(duration_seconds * self.sample_rate) t = np.linspace(0, duration_seconds, samples, False) return (amplitude * np.sin(2 * np.pi * frequency * t)).astype(np.float32) def generate_synthetic_wakeword(self, duration_seconds: float = 1.0) -> np.ndarray: """Generate synthetic wakeword-like audio.""" # Combine multiple frequency components to simulate speech samples = int(duration_seconds * self.sample_rate) t = np.linspace(0, duration_seconds, samples, False) # Fundamental frequency (voice pitch) f0 = 150 # Hz audio = 0.3 * np.sin(2 * np.pi * f0 * t) # Add harmonics for harmonic in [2, 3, 4, 5]: amplitude = 0.1 / harmonic audio += amplitude * np.sin(2 * np.pi * f0 * harmonic * t) # Add formants (speech resonances) formants = [800, 1200, 2600] # Hz for formant in formants: amplitude = 0.05 * np.exp(-0.5 * ((t - duration_seconds/2) / (duration_seconds/6))**2) audio += amplitude * np.sin(2 * np.pi * formant * t) # Apply envelope (attack, sustain, decay) envelope = np.ones_like(t) attack_samples = int(0.1 * samples) decay_samples = int(0.2 * samples) # Attack envelope[:attack_samples] = np.linspace(0, 1, attack_samples) # Decay envelope[-decay_samples:] = np.linspace(1, 0, decay_samples) audio *= envelope # Add slight noise audio += 0.02 * np.random.normal(0, 1, samples) return audio.astype(np.float32) def load_test_audio(self, file_path: str) -> Optional[np.ndarray]: """Load test audio file.""" try: import librosa audio, sr = librosa.load(file_path, sr=self.sample_rate, mono=True) return audio.astype(np.float32) except ImportError: print("librosa not available, cannot load audio file") return None except Exception as e: print(f"Error loading audio file: {e}") return None def create_test_sequence(self) -> List[np.ndarray]: """Create a sequence of test audio chunks.""" return [ self.generate_silence(0.5), self.generate_white_noise(0.3, 0.05), self.generate_synthetic_wakeword(1.0), self.generate_silence(0.5), self.generate_sine_wave(0.2, 1000, 0.1), self.generate_synthetic_wakeword(1.2), self.generate_silence(1.0) ] class WakewordTester: """Comprehensive testing utility for wakeword detection.""" def __init__(self, detector: WakewordDetector, audio_simulator: AudioSimulator): self.detector = detector self.audio_simulator = audio_simulator self.test_results = [] def run_basic_test(self) -> Dict[str, Any]: """Run basic functionality test.""" print("Running basic wakeword detection test...") # Start detector self.detector.start() # Generate test audio test_audio = self.audio_simulator.generate_synthetic_wakeword(1.0) # Process audio start_time = time.time() results = self.detector.process_audio(test_audio) processing_time = time.time() - start_time # Stop detector self.detector.stop() test_result = { 'test_name': 'basic_test', 'audio_duration': len(test_audio) / self.audio_simulator.sample_rate, 'processing_time': processing_time, 'results_count': len(results), 'success': True } self.test_results.append(test_result) print(f"Basic test completed: {len(results)} results in {processing_time:.3f}s") return test_result def run_performance_test(self, duration_seconds: float = 10.0) -> Dict[str, Any]: """Run performance stress test.""" print(f"Running performance test for {duration_seconds} seconds...") # Set up performance monitoring perf_monitor = PerformanceMonitor() perf_monitor.start_monitoring() # Start detector self.detector.start() start_time = time.time() total_chunks = 0 total_detections = 0 try: while time.time() - start_time < duration_seconds: # Generate random audio chunk chunk_type = np.random.choice(['silence', 'noise', 'wakeword'], p=[0.6, 0.3, 0.1]) if chunk_type == 'silence': audio = self.audio_simulator.generate_silence(0.5) elif chunk_type == 'noise': audio = self.audio_simulator.generate_white_noise(0.5, 0.1) else: # wakeword audio = self.audio_simulator.generate_synthetic_wakeword(1.0) # Process audio results = self.detector.process_audio(audio) total_chunks += 1 # Count detections for result in results: if hasattr(result, 'is_wakeword_detected') and result.is_wakeword_detected: total_detections += 1 # Small delay to avoid overwhelming the system time.sleep(0.01) except KeyboardInterrupt: print("Performance test interrupted") # Stop systems self.detector.stop() perf_monitor.stop_monitoring() # Get performance stats detector_stats = self.detector.get_performance_stats() perf_summary = perf_monitor.get_performance_summary() test_result = { 'test_name': 'performance_test', 'duration': time.time() - start_time, 'total_chunks': total_chunks, 'total_detections': total_detections, 'chunks_per_second': total_chunks / (time.time() - start_time), 'detector_stats': detector_stats, 'performance_summary': perf_summary } self.test_results.append(test_result) print(f"Performance test completed: {total_chunks} chunks, {total_detections} detections") return test_result def run_accuracy_test(self, test_cases: List[Dict[str, Any]]) -> Dict[str, Any]: """Run accuracy test with labeled test cases.""" print(f"Running accuracy test with {len(test_cases)} test cases...") self.detector.start() correct_predictions = 0 total_predictions = 0 confusion_matrix = { 'true_positives': 0, 'false_positives': 0, 'true_negatives': 0, 'false_negatives': 0 } for i, test_case in enumerate(test_cases): audio_type = test_case['type'] # 'wakeword', 'negative' expected_detection = test_case['should_detect'] # Generate or load audio if 'audio_data' in test_case: audio = test_case['audio_data'] elif audio_type == 'wakeword': audio = self.audio_simulator.generate_synthetic_wakeword() else: audio = self.audio_simulator.generate_white_noise(1.0, 0.1) # Process audio results = self.detector.process_audio(audio) # Check prediction detected = any(hasattr(r, 'is_wakeword_detected') and r.is_wakeword_detected for r in results) if detected == expected_detection: correct_predictions += 1 # Update confusion matrix if expected_detection and detected: confusion_matrix['true_positives'] += 1 elif expected_detection and not detected: confusion_matrix['false_negatives'] += 1 elif not expected_detection and detected: confusion_matrix['false_positives'] += 1 else: confusion_matrix['true_negatives'] += 1 total_predictions += 1 if i % 10 == 0: print(f"Processed {i+1}/{len(test_cases)} test cases") self.detector.stop() # Calculate metrics accuracy = correct_predictions / total_predictions if total_predictions > 0 else 0 tp = confusion_matrix['true_positives'] fp = confusion_matrix['false_positives'] fn = confusion_matrix['false_negatives'] precision = tp / (tp + fp) if (tp + fp) > 0 else 0 recall = tp / (tp + fn) if (tp + fn) > 0 else 0 f1_score = 2 * (precision * recall) / (precision + recall) if (precision + recall) > 0 else 0 test_result = { 'test_name': 'accuracy_test', 'total_cases': total_predictions, 'correct_predictions': correct_predictions, 'accuracy': accuracy, 'precision': precision, 'recall': recall, 'f1_score': f1_score, 'confusion_matrix': confusion_matrix } self.test_results.append(test_result) print(f"Accuracy test completed: {accuracy:.3f} accuracy, {f1_score:.3f} F1-score") return test_result def run_latency_test(self, num_iterations: int = 100) -> Dict[str, Any]: """Run latency measurement test.""" print(f"Running latency test with {num_iterations} iterations...") self.detector.start() latencies = [] for i in range(num_iterations): # Generate test audio audio = self.audio_simulator.generate_synthetic_wakeword(1.0) # Measure processing latency start_time = time.time() results = self.detector.process_audio(audio) end_time = time.time() latency_ms = (end_time - start_time) * 1000 latencies.append(latency_ms) if i % 20 == 0: print(f"Completed {i+1}/{num_iterations} latency measurements") self.detector.stop() test_result = { 'test_name': 'latency_test', 'iterations': num_iterations, 'mean_latency_ms': np.mean(latencies), 'std_latency_ms': np.std(latencies), 'min_latency_ms': min(latencies), 'max_latency_ms': max(latencies), 'p95_latency_ms': np.percentile(latencies, 95), 'p99_latency_ms': np.percentile(latencies, 99) } self.test_results.append(test_result) print(f"Latency test completed: {test_result['mean_latency_ms']:.1f}ms average") return test_result def get_test_report(self) -> Dict[str, Any]: """Get comprehensive test report.""" return { 'test_results': self.test_results, 'summary': { 'total_tests': len(self.test_results), 'successful_tests': sum(1 for t in self.test_results if t.get('success', True)), 'test_names': [t['test_name'] for t in self.test_results] } } def save_test_report(self, file_path: str): """Save test report to file.""" report = self.get_test_report() with open(file_path, 'w') as f: json.dump(report, f, indent=2, default=str) print(f"Test report saved to: {file_path}") def example_basic_usage(): """Basic usage example of the wakeword detection system.""" print("=== Basic Usage Example ===") # Create configuration config = create_default_config() # Note: In real usage, you would set these to actual values config.model_config.model_path = "path/to/your/model.pth" config.model_config.model_password = "your_password" config.satellite_id = "example_satellite" # Create mock event handler for this example event_handler = MockEventHandler() # Register event callback def on_wakeword_detected(event_name, event_data): print(f"Wakeword detected: {event_data.wakeword_type} " f"(confidence: {event_data.confidence:.3f})") event_handler.register_callback("wakeword_received", on_wakeword_detected) try: # Create detector (this would fail without a real model file) # detector = WakewordDetector(config, event_handler) print("Note: This example requires a real model file to run") print("Configuration created successfully:") print(f"- Model path: {config.model_config.model_path}") print(f"- Device: {config.model_config.device}") print(f"- Sample rate: {config.audio_config.sample_rate} Hz") print(f"- Custom threshold: {config.detection_config.custom_threshold}") print(f"- System command threshold: {config.detection_config.system_command_threshold}") except Exception as e: print(f"Cannot create detector without valid model: {e}") def example_performance_optimization(): """Example of performance optimization techniques.""" print("=== Performance Optimization Example ===") # Create lightweight configuration for resource-constrained devices lightweight_config = create_lightweight_config() print("Lightweight configuration:") print(f"- Half precision: {lightweight_config.model_config.use_half_precision}") print(f"- Memory limit: {lightweight_config.performance_config.max_memory_usage_mb} MB") print(f"- Worker threads: {lightweight_config.performance_config.max_worker_threads}") # Create high-performance configuration high_perf_config = create_high_performance_config() print("\nHigh-performance configuration:") print(f"- Model compilation: {high_perf_config.model_config.enable_model_compilation}") print(f"- Batch inference: {high_perf_config.detection_config.batch_inference}") print(f"- Memory limit: {high_perf_config.performance_config.max_memory_usage_mb} MB") # Example of auto-tuning from .config import auto_tune_config tuned_config = auto_tune_config( lightweight_config, target_latency_ms=50.0, available_memory_mb=256.0 ) print(f"\nAuto-tuned configuration for 50ms latency and 256MB memory:") print(f"- Chunk duration: {tuned_config.audio_config.chunk_duration}s") print(f"- Temporal smoothing: {tuned_config.detection_config.use_temporal_smoothing}") def example_testing_framework(): """Example of using the testing framework.""" print("=== Testing Framework Example ===") # Create audio simulator audio_sim = AudioSimulator(sample_rate=16000) # Generate test audio print("Generating test audio samples...") silence = audio_sim.generate_silence(1.0) noise = audio_sim.generate_white_noise(1.0, 0.1) synthetic_wakeword = audio_sim.generate_synthetic_wakeword(1.0) print(f"Generated silence: {len(silence)} samples") print(f"Generated noise: {len(noise)} samples") print(f"Generated synthetic wakeword: {len(synthetic_wakeword)} samples") # Create test sequence test_sequence = audio_sim.create_test_sequence() total_duration = sum(len(chunk) for chunk in test_sequence) / audio_sim.sample_rate print(f"Created test sequence: {len(test_sequence)} chunks, {total_duration:.1f}s total") # Mock event handler event_handler = MockEventHandler() # Note: Would create detector with real model for actual testing print("\nNote: Testing requires a real model file to run") print("Test cases would include:") print("- Basic functionality test") print("- Performance stress test") print("- Accuracy evaluation") print("- Latency measurement") def example_streaming_audio(): """Example of streaming audio processing.""" print("=== Streaming Audio Example ===") def simulate_audio_stream(detector, duration_seconds=5.0): """Simulate real-time audio streaming.""" audio_sim = AudioSimulator() chunk_duration = 0.1 # 100ms chunks sample_rate = detector.config.audio_config.sample_rate chunk_samples = int(chunk_duration * sample_rate) print(f"Simulating {duration_seconds}s of audio streaming...") start_time = time.time() chunk_count = 0 while time.time() - start_time < duration_seconds: # Generate audio chunk (mix of silence and occasional synthetic speech) if np.random.random() < 0.1: # 10% chance of "wakeword" audio_chunk = audio_sim.generate_synthetic_wakeword(chunk_duration) elif np.random.random() < 0.3: # 30% chance of noise audio_chunk = audio_sim.generate_white_noise(chunk_duration, 0.05) else: # 60% silence audio_chunk = audio_sim.generate_silence(chunk_duration) # Process chunk results = detector.process_audio(audio_chunk) chunk_count += 1 # Print detection results for result in results: if hasattr(result, 'is_wakeword_detected') and result.is_wakeword_detected: print(f"Detection: {result.wakeword_type.value} " f"(confidence: {result.confidence:.3f})") # Simulate real-time processing time.sleep(chunk_duration * 0.8) # Slightly faster than real-time print(f"Processed {chunk_count} audio chunks") print("Note: Streaming example requires a real detector instance") print("Would simulate real-time audio processing with periodic detections") def run_all_examples(): """Run all examples.""" print("Running all Trixy Wakeword Detection examples...\n") try: example_basic_usage() print("\n" + "="*60 + "\n") example_performance_optimization() print("\n" + "="*60 + "\n") example_testing_framework() print("\n" + "="*60 + "\n") example_streaming_audio() print("\n" + "="*60 + "\n") print("All examples completed successfully!") except Exception as e: print(f"Error running examples: {e}") import traceback traceback.print_exc() if __name__ == "__main__": # Set up logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) run_all_examples()