Trixy3/trixy_core/ml/voice_recognition

Trixy Voice Recognition System

A comprehensive voice recognition system for the Trixy voice assistant client, providing real-time speaker identification and verification capabilities with production-ready performance optimization.

Features

Core Capabilities

• Real-time Speaker Identification: Identify speakers from audio with low latency (<100ms)
• Speaker Verification: Verify if audio belongs to a specific speaker
• Dynamic Speaker Management: Add/remove speakers without retraining models
• Voice Activity Detection (VAD): WebRTC VAD and energy-based detection
• Multiple Model Architectures: ECAPA-TDNN, TitaNet-S, SpeakerNet-M
• Secure Model Loading: Password-protected .pth files with metadata
• Event System Integration: Seamless integration with Trixy event system

Audio Processing

• Sample Rate: 16kHz, 16-bit PCM mono audio
• Feature Extraction: Log-Mel-Spectrogram or MFCC (40 filter banks, 25ms window, 10ms shift)
• Data Augmentation: Noise, reverberation, loudness, mic simulation
• Audio Length: Configurable (default 3 seconds)

Performance Optimization

• Real-time Processing: Optimized for live audio streams
• Memory Management: Efficient memory usage with cleanup
• Device Support: CPU, CUDA, MPS (Apple Silicon)
• Model Optimization: TorchScript, quantization, half precision
• Batch Processing: Support for batch inference

Architecture

System Components

┌─────────────────────────────────────────────────────────────┐
│                 Voice Recognition System                     │
├─────────────────────────────────────────────────────────────┤
│  VoiceRecognitionDetector (Main Interface)                  │
│  ├── Audio Features (Log-Mel/MFCC)                         │
│  ├── VAD (WebRTC/Energy-based)                             │
│  ├── Model Architectures (ECAPA-TDNN/TitaNet-S/SpeakerNet) │
│  ├── Speaker Embeddings & Comparison                       │
│  ├── Speaker Manager (Dynamic Add/Remove)                  │
│  ├── Recognition Engine (Confidence Scoring)               │
│  └── Performance Optimization                              │
├─────────────────────────────────────────────────────────────┤
│  Integration Layer                                          │
│  ├── Event System Integration                              │
│  ├── Wakeword Enhancement                                  │
│  └── Continuous Recognition                                │
└─────────────────────────────────────────────────────────────┘

Model Architectures

1. ECAPA-TDNN - Enhanced version with channel attention and residual connections
2. TitaNet-S - Lightweight architecture optimized for real-time inference
   
3. SpeakerNet-M - LSTM-based with attention mechanism

Supported Configurations

• Speakers: 4-90+ speakers (dynamic)
• Embedding Dimensions: 192, 256, 512
• Feature Types: Log-Mel-Spectrogram, MFCC
• Similarity Metrics: Cosine similarity, Euclidean distance, Dot product
• Loss Functions: ArcFace, GE2E, Triplet Loss (for training)

Installation

Dependencies

# Core dependencies
pip install torch torchaudio numpy scipy scikit-learn

# Audio processing
pip install webrtcvad librosa

# Security (for encrypted models)
pip install cryptography

# Performance monitoring
pip install psutil

# Optional: for visualization
pip install matplotlib seaborn

Setup

from trixy_core.ml.voice_recognition import setup_complete_voice_recognition_system

# Create complete system
integration = setup_complete_voice_recognition_system(
    model_path="/path/to/trained_model.pth",
    model_password="your_secret_password",
    event_handler=event_handler  # Optional
)

# Start the system
integration.start()

Usage Examples

Basic Speaker Identification

from trixy_core.ml.voice_recognition import (
    create_default_config, 
    VoiceRecognitionDetector
)
import numpy as np

# Create configuration
config = create_default_config(
    model_path="/path/to/model.pth",
    model_password="secret123"
)

# Initialize detector
detector = VoiceRecognitionDetector(config)
detector.start()

# Process audio (16kHz, 16-bit PCM)
audio_data = np.random.randn(48000).astype(np.float32)  # 3 seconds
result = detector.identify_speaker(audio_data)

print(f"Speaker: {result.speaker_name}")
print(f"Confidence: {result.confidence:.3f}")
print(f"Processing time: {result.processing_time_ms:.1f}ms")

Speaker Verification

# Verify specific speaker
verification_result = detector.verify_speaker(audio_data, "expected_speaker_id")

print(f"Verification: {verification_result.verification_success}")
print(f"Expected: {verification_result.expected_speaker_id}")
print(f"Actual: {verification_result.speaker_id}")
print(f"Confidence: {verification_result.confidence:.3f}")

Dynamic Speaker Management

# Add new speaker with multiple audio samples
audio_samples = [
    np.random.randn(48000).astype(np.float32) for _ in range(5)
]

success = detector.add_speaker("New Speaker", audio_samples)
print(f"Enrollment success: {success}")

# List all speakers
speakers = detector.get_speaker_list()
for speaker in speakers:
    print(f"{speaker.speaker_name}: {speaker.embedding_count} embeddings")

# Remove speaker
removed = detector.remove_speaker("speaker_id_to_remove")
print(f"Removal success: {removed}")

Event System Integration

from trixy_core.ml.voice_recognition import create_voice_recognition_integration
from trixy_core.events import create_default_event_handler

# Create event handler
event_handler = create_default_event_handler()

# Create integration
integration = create_voice_recognition_integration(
    voice_detector=detector,
    event_handler=event_handler,
    enable_wakeword_enhancement=True
)

# Start integration
integration.start()

# Process wakeword detection with enhancement
enhanced_info = integration.process_wakeword_detection(
    audio_data=wakeword_audio,
    wakeword_id="hey_trixy"
)

print(f"Enhanced speaker info: {enhanced_info}")

Performance Monitoring

from trixy_core.ml.voice_recognition import create_vr_performance_monitor

# Create performance monitor
monitor = create_vr_performance_monitor(
    window_size=100,
    enable_profiling=True
)

# Start monitoring
monitor.start_monitoring()

# Record inference performance
monitor.record_inference(
    inference_time_ms=25.3,
    confidence=0.85,
    success=True
)

# Get performance metrics
metrics = monitor.get_metrics()
print(f"Average inference time: {metrics.avg_inference_time_ms:.1f}ms")
print(f"Frames per second: {metrics.frames_per_second:.1f}")
print(f"Memory usage: {metrics.memory_usage_mb:.1f}MB")

Continuous Recognition

# Enable continuous recognition mode
integration = create_voice_recognition_integration(
    voice_detector=detector,
    event_handler=event_handler,
    enable_continuous_recognition=True
)

integration.start()

# Process streaming audio
def audio_callback(audio_chunk):
    result = integration.process_audio_stream(audio_chunk)
    if result and result.speaker_id:
        print(f"Continuous recognition: {result.speaker_name}")

# The integration will continuously process audio and trigger callbacks

Configuration

Audio Configuration

from trixy_core.ml.voice_recognition import AudioConfig, FeatureType, VADType

audio_config = AudioConfig(
    sample_rate=16000,
    audio_length=3.0,
    feature_type=FeatureType.LOG_MEL,
    n_mels=40,
    n_fft=1024,
    hop_length=160,
    win_length=400,
    use_vad=True,
    vad_type=VADType.WEBRTC,
    vad_aggressiveness=2
)

Model Configuration

from trixy_core.ml.voice_recognition import ModelConfig, ModelArchitectureType

model_config = ModelConfig(
    model_path="/path/to/model.pth",
    model_password="secret123",
    architecture_type=ModelArchitectureType.ECAPA_TDNN,
    embedding_dim=192,
    num_speakers=10,
    device="auto"  # auto-detect best device
)

Recognition Configuration

from trixy_core.ml.voice_recognition import RecognitionConfig, SimilarityMetric

recognition_config = RecognitionConfig(
    identification_threshold=0.7,
    verification_threshold=0.8,
    similarity_metric=SimilarityMetric.COSINE,
    use_temporal_smoothing=True,
    smoothing_window_size=5,
    confidence_decay_factor=0.9
)

Performance Configuration

from trixy_core.ml.voice_recognition import PerformanceConfig

performance_config = PerformanceConfig(
    device="cuda",  # or "cpu", "mps", "auto"
    use_half_precision=True,
    use_torch_compile=True,
    batch_size=1,
    num_threads=4,
    target_latency_ms=100
)

Model Training Integration

The system is designed to work with models trained using the ML framework. Model files should include:

Model Metadata

• Speaker mapping (ID to name)
• Architecture configuration
• Training statistics
• Audio processing parameters
• Compatibility information

Secure Model Format

model.pth (ZIP archive)
├── model.pth (PyTorch state dict)
├── metadata.json (Model metadata)
└── README.txt (Documentation)

Example Model Loading

from trixy_core.ml.voice_recognition import (
    SecureVRModelLoader, 
    validate_vr_model_file
)

# Validate model file
validation = validate_vr_model_file(
    model_path="/path/to/model.pth",
    password="secret123"
)

if validation['is_valid']:
    print("Model is valid")
    print(f"Speakers: {len(validation['metadata']['speaker_mapping'])}")
else:
    print(f"Model validation failed: {validation['errors']}")

# Load model securely
loader = SecureVRModelLoader("cuda")
model, metadata = loader.load_model(
    model_path="/path/to/model.pth",
    password="secret123"
)

Performance Optimization

Real-time Optimization

from trixy_core.ml.voice_recognition import optimize_vr_for_deployment

# Optimize model for deployment
optimization_results = optimize_vr_for_deployment(
    model=model,
    target_device="cuda",
    target_latency_ms=50,
    enable_quantization=True
)

print(f"Model size reduced by {optimization_results['model_size_reduction']:.1f}%")
optimized_model = optimization_results['optimized_model']

Memory Management

from trixy_core.ml.voice_recognition import VRMemoryManager

# Create memory manager
memory_manager = VRMemoryManager(max_memory_mb=512)

# Check memory usage
if not memory_manager.check_memory_usage():
    memory_manager.force_cleanup()

Latency Optimization

from trixy_core.ml.voice_recognition import VRLatencyOptimizer

# Create latency optimizer
latency_optimizer = VRLatencyOptimizer(target_latency_ms=100)

# Record latencies
latency_optimizer.record_latency(85.3)  # ms

# Get optimization suggestions
suggestions = latency_optimizer.get_optimization_suggestions()
for suggestion in suggestions:
    print(f"Suggestion: {suggestion}")

Testing

Unit Testing

from trixy_core.ml.voice_recognition import VoiceRecognitionTester

# Create tester
tester = VoiceRecognitionTester(config)

# Setup test speakers
speakers = {
    'alice': {'f0': 180, 'formants': [900, 1300, 2800]},
    'bob': {'f0': 120, 'formants': [700, 1100, 2300]}
}
tester.setup_test_speakers(speakers)

# Run identification test
results = tester.run_identification_test(
    test_speakers=['alice', 'bob'],
    samples_per_speaker=10
)

print(f"Accuracy: {results['accuracy']:.2%}")
print(f"Average confidence: {results['avg_confidence']:.3f}")

Performance Testing

# Run performance stress test
perf_results = tester.run_performance_test(
    duration_minutes=5.0,
    target_fps=33.3
)

print(f"Achieved FPS: {perf_results['achieved_fps']:.1f}")
print(f"Average processing time: {perf_results['avg_processing_time_ms']:.1f}ms")

Integration Testing

# Test complete integration
from trixy_core.ml.voice_recognition import run_all_vr_examples

# Run all example scenarios
results = run_all_vr_examples()
print("All tests completed successfully!")

Event System Integration

Event Types

The system generates the following events:

• voice_recognition_started - System initialization
• voice_recognition_stopped - System shutdown
• speaker_identified - Speaker identification completed
• speaker_verified - Speaker verification completed
• speaker_enrolled - New speaker added
• speaker_removed - Speaker removed
• wakeword_enhanced - Wakeword detection enhanced with speaker info

Event Data Structures

# Speaker identification event
{
    'event_type': 'speaker_identified',
    'speaker_id': 'spk_001',
    'speaker_name': 'Alice',
    'confidence': 0.89,
    'similarity_score': 0.89,
    'processing_time_ms': 23.5,
    'audio_quality': 0.92,
    'vad_confidence': 0.95,
    'timestamp': '2024-01-15T10:30:45.123Z'
}

Custom Event Handlers

from trixy_core.events import TrixyEvent

class VoiceRecognitionHandler:
    @TrixyEvent(["speaker_identified", "speaker_enrolled"])
    def handle_voice_events(self, event_name, event_data):
        if event_name == "speaker_identified":
            print(f"Speaker identified: {event_data.speaker_name}")
        elif event_name == "speaker_enrolled":
            print(f"New speaker enrolled: {event_data.speaker_name}")

# Register handler
handler = VoiceRecognitionHandler()
event_handler.register_handler_object(handler)

Best Practices

Production Deployment

1. Model Security: Always use password-protected models in production
2. Performance Monitoring: Enable performance monitoring for latency tracking
3. Memory Management: Set appropriate memory limits and enable cleanup
4. Error Handling: Implement robust error handling and fallback mechanisms
5. Logging: Enable appropriate logging levels for debugging

Speaker Management

1. Enrollment Quality: Use multiple high-quality audio samples for enrollment
2. Speaker Diversity: Ensure diverse speaker representations in training data
3. Regular Updates: Periodically update speaker embeddings with new samples
4. Verification Thresholds: Tune thresholds based on security requirements

Performance Optimization

1. Device Selection: Use CUDA when available for better performance
2. Model Optimization: Enable quantization and compilation for deployment
3. Batch Processing: Use batch processing for multiple simultaneous requests
4. Caching: Cache speaker embeddings for faster comparison

Troubleshooting

Common Issues

1. Model Loading Fails

   • Check model file path and password
   • Verify model file integrity
   • Ensure sufficient memory

2. Poor Recognition Accuracy

   • Check audio quality and sample rate
   • Verify speaker enrollment quality
   • Adjust confidence thresholds

3. High Latency

   • Enable model optimization
   • Check device utilization
   • Reduce audio processing length

4. Memory Issues

   • Enable memory management
   • Reduce batch size
   • Clear speaker embedding cache

Debug Mode

# Enable debug mode for detailed logging
config.debug_mode = True
detector = VoiceRecognitionDetector(config)

# Save debug information
detector.save_debug_info("/tmp/vr_debug.json")

API Reference

For detailed API documentation, see the individual module docstrings:

• VoiceRecognitionDetector - Main detection interface
• VoiceRecognitionConfig - Configuration management
• SpeakerManager - Dynamic speaker management
• VRPerformanceMonitor - Performance monitoring
• VoiceRecognitionIntegration - Event system integration

License

This voice recognition system is part of the Trixy voice assistant project.