Trixy3/trixy_core/ml/wakeword/examples.py

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