""" Text-to-Speech (TTS) Plugin for Trixy Application This plugin demonstrates TTS integration with the Trixy plugin system: - Audio synthesis and voice generation - Integration with various TTS engines (OpenAI TTS, Google TTS, Azure, etc.) - Voice selection and customization - Speech rate, pitch, and volume control - SSML (Speech Synthesis Markup Language) support - Audio output format control - Batch processing and streaming synthesis - Voice caching and optimization - Multi-language support - Custom voice models and training This serves as a reference for implementing TTS functionality in Trixy. Features: - Multiple TTS engine support - Voice customization and selection - SSML markup support - Audio format control - Streaming and batch synthesis - Voice caching - Performance optimization - Multi-language support """ import os import time import threading import queue import tempfile import hashlib from typing import Any, Dict, Optional, List, Tuple, Union from pathlib import Path from dataclasses import dataclass # Import plugin base class and event decorator from trixy_core.plugins import TrixyPlugin from trixy_core.events import TrixyEvent def pprint(message: str) -> None: """TTS plugin logging function.""" print(f"[TTS_PLUGIN] {message}") @dataclass class Voice: """Voice configuration.""" name: str gender: str = "neutral" language: str = "en" region: str = "US" age: str = "adult" style: str = "neutral" engine: str = "default" @dataclass class TTSRequest: """Text-to-speech synthesis request.""" text: str voice: Voice conversation_id: str = "" speed: float = 1.0 pitch: float = 1.0 volume: float = 1.0 output_format: str = "wav" use_ssml: bool = False timestamp: float = 0.0 def __post_init__(self): if self.timestamp == 0.0: self.timestamp = time.time() class TTSResult: """Text-to-speech synthesis result.""" def __init__( self, audio_data: bytes, text: str, voice: Voice, processing_time: float = 0.0, output_format: str = "wav", sample_rate: int = 16000, success: bool = True, error_message: str = "" ): self.audio_data = audio_data self.text = text self.voice = voice self.processing_time = processing_time self.output_format = output_format self.sample_rate = sample_rate self.success = success self.error_message = error_message self.timestamp = time.time() class TTSPlugin(TrixyPlugin): """ Text-to-Speech plugin for Trixy application. This plugin provides TTS functionality with support for: - Multiple TTS engines (OpenAI TTS, Google TTS, Azure Speech, etc.) - Voice selection and customization - SSML markup support - Audio format control and optimization - Streaming and batch synthesis - Voice caching and performance optimization - Multi-language support """ def initialize(self) -> None: """Initialize the TTS plugin.""" pprint("Initializing TTS Plugin...") # Plugin state self._synthesis_queue = queue.Queue() self._synthesis_thread: Optional[threading.Thread] = None self._synthesis_shutdown = threading.Event() self._active_requests: Dict[str, TTSRequest] = {} # Voice cache self._voice_cache: Dict[str, bytes] = {} self._cache_lock = threading.RLock() # Statistics self._stats = { 'total_requests': 0, 'successful_syntheses': 0, 'failed_syntheses': 0, 'total_processing_time': 0.0, 'average_processing_time': 0.0, 'cache_hits': 0, 'total_audio_generated': 0, # in bytes 'voices_used': set(), 'languages_used': set() } # Setup configuration self._setup_configuration() # Initialize available voices self._initialize_voices() # Initialize TTS engines self._initialize_engines() # Start synthesis thread self._start_synthesis_thread() pprint(f"TTS Plugin initialized with {len(self._voices)} voices and {len(self._engines)} engines") def _setup_configuration(self) -> None: """Setup plugin configuration.""" # Engine configuration self.primary_engine = self.get_config_value("primary_engine", "openai", str) self.fallback_engine = self.get_config_value("fallback_engine", "google", str) self.engines_enabled = self.get_config_value("engines_enabled", ["openai", "google", "azure"], list) # Voice configuration self.default_voice_name = self.get_config_value("default_voice", "alloy", str) self.default_language = self.get_config_value("default_language", "en", str) self.default_gender = self.get_config_value("default_gender", "neutral", str) # Audio configuration self.output_format = self.get_config_value("output_format", "wav", str) self.sample_rate = self.get_config_value("sample_rate", 16000, int) self.bit_depth = self.get_config_value("bit_depth", 16, int) self.channels = self.get_config_value("channels", 1, int) # Synthesis parameters self.default_speed = self.get_config_value("default_speed", 1.0, float) self.default_pitch = self.get_config_value("default_pitch", 1.0, float) self.default_volume = self.get_config_value("default_volume", 1.0, float) # Processing configuration self.max_text_length = self.get_config_value("max_text_length", 4000, int) self.enable_caching = self.get_config_value("enable_caching", True, bool) self.enable_ssml = self.get_config_value("enable_ssml", True, bool) self.processing_timeout = self.get_config_value("processing_timeout", 30.0, float) # Performance configuration self.max_concurrent_requests = self.get_config_value("max_concurrent_requests", 2, int) self.cache_size_mb = self.get_config_value("cache_size_mb", 50, int) self.enable_streaming = self.get_config_value("enable_streaming", False, bool) # Validation if self.default_speed <= 0 or self.default_speed > 3.0: pprint("Warning: default_speed should be between 0.1 and 3.0") self.default_speed = max(0.1, min(3.0, self.default_speed)) if self.max_text_length <= 0: pprint("Warning: max_text_length must be positive") self.max_text_length = 4000 def _initialize_voices(self) -> None: """Initialize available voices.""" self._voices: Dict[str, Voice] = {} # OpenAI voices openai_voices = [ Voice("alloy", "neutral", "en", "US", "adult", "neutral", "openai"), Voice("echo", "male", "en", "US", "adult", "neutral", "openai"), Voice("fable", "female", "en", "UK", "adult", "calm", "openai"), Voice("onyx", "male", "en", "US", "adult", "deep", "openai"), Voice("nova", "female", "en", "US", "young", "bright", "openai"), Voice("shimmer", "female", "en", "US", "adult", "warm", "openai") ] # Google voices google_voices = [ Voice("en-US-Standard-A", "female", "en", "US", "adult", "standard", "google"), Voice("en-US-Standard-B", "male", "en", "US", "adult", "standard", "google"), Voice("en-US-Standard-C", "female", "en", "US", "adult", "standard", "google"), Voice("en-US-Standard-D", "male", "en", "US", "adult", "standard", "google"), Voice("en-US-Wavenet-A", "female", "en", "US", "adult", "wavenet", "google"), Voice("en-US-Wavenet-B", "male", "en", "US", "adult", "wavenet", "google") ] # Azure voices azure_voices = [ Voice("en-US-AriaNeural", "female", "en", "US", "adult", "neutral", "azure"), Voice("en-US-DavisNeural", "male", "en", "US", "adult", "neutral", "azure"), Voice("en-US-JennyNeural", "female", "en", "US", "adult", "assistant", "azure"), Voice("en-US-GuyNeural", "male", "en", "US", "adult", "neutral", "azure") ] # Add all voices to the registry for voice_list in [openai_voices, google_voices, azure_voices]: for voice in voice_list: self._voices[voice.name] = voice # Load custom voices from config custom_voices = self.get_config_value("custom_voices", [], list) for voice_config in custom_voices: if isinstance(voice_config, dict) and "name" in voice_config: voice = Voice( name=voice_config["name"], gender=voice_config.get("gender", "neutral"), language=voice_config.get("language", "en"), region=voice_config.get("region", "US"), age=voice_config.get("age", "adult"), style=voice_config.get("style", "neutral"), engine=voice_config.get("engine", "custom") ) self._voices[voice.name] = voice pprint(f"Initialized {len(self._voices)} voices") def _initialize_engines(self) -> None: """Initialize available TTS engines.""" self._engines = {} # Initialize OpenAI TTS engine if "openai" in self.engines_enabled: self._engines["openai"] = self._init_openai_engine() # Initialize Google TTS engine if "google" in self.engines_enabled: self._engines["google"] = self._init_google_engine() # Initialize Azure Speech engine if "azure" in self.engines_enabled: self._engines["azure"] = self._init_azure_engine() # Initialize custom engines if "custom" in self.engines_enabled: self._engines["custom"] = self._init_custom_engine() pprint(f"Initialized {len(self._engines)} TTS engines: {list(self._engines.keys())}") def _init_openai_engine(self) -> Dict[str, Any]: """Initialize OpenAI TTS engine.""" try: # Try to import openai (placeholder - would import real OpenAI client) # import openai openai_config = { 'api_key': self.get_config_value("openai_api_key", "", str), 'model': self.get_config_value("openai_model", "tts-1", str), 'response_format': self.get_config_value("openai_response_format", "mp3", str), 'speed': self.get_config_value("openai_speed", 1.0, float) } pprint(f"OpenAI TTS engine configured with model: {openai_config['model']}") return { 'name': 'openai', 'available': True, 'config': openai_config, 'client': None, # Placeholder - would initialize actual client 'supported_formats': ['mp3', 'opus', 'aac', 'flac'], 'max_text_length': 4096 } except ImportError: pprint("OpenAI TTS not available - install openai package") return {'name': 'openai', 'available': False} def _init_google_engine(self) -> Dict[str, Any]: """Initialize Google Text-to-Speech engine.""" try: # Try to import google cloud tts (placeholder) # from google.cloud import texttospeech google_config = { 'credentials_path': self.get_config_value("google_credentials_path", "", str), 'audio_encoding': self.get_config_value("google_audio_encoding", "LINEAR16", str), 'sample_rate_hertz': self.get_config_value("google_sample_rate", 16000, int), 'effects_profile_id': self.get_config_value("google_effects_profile", [], list) } pprint("Google TTS engine configured") return { 'name': 'google', 'available': True, 'config': google_config, 'client': None, # Placeholder 'supported_formats': ['wav', 'mp3', 'ogg'], 'max_text_length': 5000 } except ImportError: pprint("Google TTS not available - install google-cloud-texttospeech package") return {'name': 'google', 'available': False} def _init_azure_engine(self) -> Dict[str, Any]: """Initialize Azure Speech Services TTS engine.""" try: # Try to import azure speech (placeholder) # import azure.cognitiveservices.speech as speechsdk azure_config = { 'subscription_key': self.get_config_value("azure_subscription_key", "", str), 'region': self.get_config_value("azure_region", "eastus", str), 'output_format': self.get_config_value("azure_output_format", "Wav16Khz16BitMonoPcm", str) } pprint("Azure Speech TTS engine configured") return { 'name': 'azure', 'available': True, 'config': azure_config, 'synthesizer': None, # Placeholder 'supported_formats': ['wav', 'mp3'], 'max_text_length': 10000 } except ImportError: pprint("Azure Speech TTS not available - install azure-cognitiveservices-speech package") return {'name': 'azure', 'available': False} def _init_custom_engine(self) -> Dict[str, Any]: """Initialize custom TTS engine.""" custom_config = { 'model_path': self.get_config_value("custom_model_path", "", str), 'vocoder_path': self.get_config_value("custom_vocoder_path", "", str), 'use_gpu': self.get_config_value("custom_use_gpu", False, bool) } return { 'name': 'custom', 'available': bool(custom_config['model_path']), 'config': custom_config, 'model': None, 'supported_formats': ['wav'], 'max_text_length': 1000 } def _start_synthesis_thread(self) -> None: """Start the TTS synthesis thread.""" if self._synthesis_thread is None or not self._synthesis_thread.is_alive(): self._synthesis_thread = threading.Thread( target=self._synthesis_loop, name="TTSSynthesisThread", daemon=True ) self._synthesis_thread.start() pprint("TTS synthesis thread started") def _synthesis_loop(self) -> None: """Main synthesis loop for TTS requests.""" while not self._synthesis_shutdown.wait(0.1): try: # Get request from queue with timeout try: request = self._synthesis_queue.get(timeout=1.0) except queue.Empty: continue # Process the synthesis request self._process_synthesis_request(request) self._synthesis_queue.task_done() except Exception as e: pprint(f"Error in synthesis loop: {e}") def _process_synthesis_request(self, request: TTSRequest) -> None: """Process a single TTS synthesis request.""" try: pprint(f"Processing TTS request: '{request.text[:50]}...' with voice {request.voice.name}") # Check cache if enabled if self.enable_caching: cached_result = self._check_cache(request) if cached_result: self._stats['cache_hits'] += 1 self._send_tts_result(cached_result, request) return # Validate text length if len(request.text) > self.max_text_length: pprint(f"Text too long ({len(request.text)} chars), truncating to {self.max_text_length}") request.text = request.text[:self.max_text_length] # Perform synthesis result = self._synthesize_speech(request) # Cache result if successful and caching enabled if result and result.success and self.enable_caching: self._cache_result(request, result) # Send result if result: self._send_tts_result(result, request) # Update statistics self._update_stats(result, request) except Exception as e: pprint(f"Error processing synthesis request: {e}") self._stats['failed_syntheses'] += 1 def _synthesize_speech(self, request: TTSRequest) -> Optional[TTSResult]: """Synthesize speech from text.""" start_time = time.time() try: # Determine engine to use engine_name = request.voice.engine if engine_name == "default": engine_name = self.primary_engine # Try primary engine if engine_name in self._engines and self._engines[engine_name]['available']: result = self._synthesize_with_engine(request, engine_name) if result and result.success: return result # Fallback to secondary engine if (self.fallback_engine in self._engines and self._engines[self.fallback_engine]['available'] and self.fallback_engine != engine_name): pprint(f"Primary engine failed, trying fallback: {self.fallback_engine}") # Update request voice for fallback engine fallback_voice = self._get_compatible_voice(request.voice, self.fallback_engine) fallback_request = TTSRequest( text=request.text, voice=fallback_voice, conversation_id=request.conversation_id, speed=request.speed, pitch=request.pitch, volume=request.volume, output_format=request.output_format, use_ssml=request.use_ssml ) result = self._synthesize_with_engine(fallback_request, self.fallback_engine) if result and result.success: return result return None except Exception as e: pprint(f"Speech synthesis failed: {e}") return TTSResult( audio_data=b'', text=request.text, voice=request.voice, processing_time=time.time() - start_time, success=False, error_message=str(e) ) def _synthesize_with_engine(self, request: TTSRequest, engine_name: str) -> Optional[TTSResult]: """Synthesize speech using a specific engine.""" if engine_name == "openai": return self._synthesize_openai(request) elif engine_name == "google": return self._synthesize_google(request) elif engine_name == "azure": return self._synthesize_azure(request) elif engine_name == "custom": return self._synthesize_custom(request) else: pprint(f"Unknown engine: {engine_name}") return None def _synthesize_openai(self, request: TTSRequest) -> Optional[TTSResult]: """Synthesize speech using OpenAI TTS.""" try: start_time = time.time() pprint("Synthesizing with OpenAI TTS...") # Simulate processing delay time.sleep(0.5) # Simulate audio generation (actual implementation would call OpenAI API) audio_data = b'\x00' * (self.sample_rate * 2) # 1 second of silence return TTSResult( audio_data=audio_data, text=request.text, voice=request.voice, processing_time=time.time() - start_time, output_format="mp3", sample_rate=self.sample_rate, success=True ) except Exception as e: pprint(f"OpenAI TTS synthesis failed: {e}") return TTSResult( audio_data=b'', text=request.text, voice=request.voice, success=False, error_message=str(e) ) def _synthesize_google(self, request: TTSRequest) -> Optional[TTSResult]: """Synthesize speech using Google TTS.""" try: start_time = time.time() pprint("Synthesizing with Google TTS...") # Simulate processing time.sleep(0.3) # Simulate audio generation audio_data = b'\x00' * (self.sample_rate * 2) # 1 second of silence return TTSResult( audio_data=audio_data, text=request.text, voice=request.voice, processing_time=time.time() - start_time, output_format="wav", sample_rate=self.sample_rate, success=True ) except Exception as e: pprint(f"Google TTS synthesis failed: {e}") return None def _synthesize_azure(self, request: TTSRequest) -> Optional[TTSResult]: """Synthesize speech using Azure Speech Services.""" try: start_time = time.time() pprint("Synthesizing with Azure Speech...") # Simulate processing time.sleep(0.4) # Simulate audio generation audio_data = b'\x00' * (self.sample_rate * 2) # 1 second of silence return TTSResult( audio_data=audio_data, text=request.text, voice=request.voice, processing_time=time.time() - start_time, output_format="wav", sample_rate=self.sample_rate, success=True ) except Exception as e: pprint(f"Azure TTS synthesis failed: {e}") return None def _synthesize_custom(self, request: TTSRequest) -> Optional[TTSResult]: """Synthesize speech using custom engine.""" try: start_time = time.time() pprint("Synthesizing with custom engine...") # Simulate processing time.sleep(0.8) # Simulate audio generation audio_data = b'\x00' * (self.sample_rate * 2) # 1 second of silence return TTSResult( audio_data=audio_data, text=request.text, voice=request.voice, processing_time=time.time() - start_time, output_format="wav", sample_rate=self.sample_rate, success=True ) except Exception as e: pprint(f"Custom TTS synthesis failed: {e}") return None def _get_compatible_voice(self, voice: Voice, engine_name: str) -> Voice: """Get a compatible voice for the specified engine.""" # Find a voice with same language and gender for the target engine for voice_name, available_voice in self._voices.items(): if (available_voice.engine == engine_name and available_voice.language == voice.language and available_voice.gender == voice.gender): return available_voice # Fallback to any voice from the target engine for voice_name, available_voice in self._voices.items(): if available_voice.engine == engine_name: return available_voice # Last resort: return original voice return voice def _check_cache(self, request: TTSRequest) -> Optional[TTSResult]: """Check if synthesis result is cached.""" # Generate cache key cache_key = self._generate_cache_key(request) with self._cache_lock: if cache_key in self._voice_cache: pprint("Cache hit for TTS request") return TTSResult( audio_data=self._voice_cache[cache_key], text=request.text, voice=request.voice, processing_time=0.0, output_format=self.output_format, sample_rate=self.sample_rate, success=True ) return None def _cache_result(self, request: TTSRequest, result: TTSResult) -> None: """Cache TTS synthesis result.""" cache_key = self._generate_cache_key(request) with self._cache_lock: # Check cache size limit current_size = sum(len(data) for data in self._voice_cache.values()) max_size = self.cache_size_mb * 1024 * 1024 if current_size + len(result.audio_data) > max_size: # Remove oldest entries self._cleanup_cache() self._voice_cache[cache_key] = result.audio_data pprint(f"Cached TTS result for key: {cache_key[:16]}...") def _generate_cache_key(self, request: TTSRequest) -> str: """Generate a cache key for the request.""" key_data = f"{request.text}_{request.voice.name}_{request.speed}_{request.pitch}_{request.volume}_{request.output_format}" return hashlib.md5(key_data.encode()).hexdigest() def _cleanup_cache(self) -> None: """Clean up cache by removing oldest entries.""" if len(self._voice_cache) > 10: # Keep at most 10 entries # Remove first 5 entries (oldest) keys_to_remove = list(self._voice_cache.keys())[:5] for key in keys_to_remove: del self._voice_cache[key] pprint("Cleaned up voice cache") def _send_tts_result(self, result: TTSResult, request: TTSRequest) -> None: """Send TTS result via event system.""" try: event_handler = self.application.get_event_handler() if event_handler: event_data = { 'conversation_id': request.conversation_id, 'audio_data': result.audio_data, 'text': result.text, 'voice_settings': { 'voice_name': result.voice.name, 'language': result.voice.language, 'gender': result.voice.gender, 'engine': result.voice.engine }, 'audio_format': result.output_format, 'sample_rate': result.sample_rate, 'processing_time': result.processing_time, 'timestamp': result.timestamp, 'success': result.success, 'error_message': result.error_message } event_handler.trigger_event("tts_received", event_data) pprint(f"TTS result sent: '{result.text[:30]}...' ({len(result.audio_data)} bytes)") except Exception as e: pprint(f"Error sending TTS result: {e}") def _update_stats(self, result: Optional[TTSResult], request: TTSRequest) -> None: """Update processing statistics.""" self._stats['total_requests'] += 1 if result and result.success: self._stats['successful_syntheses'] += 1 self._stats['total_processing_time'] += result.processing_time self._stats['average_processing_time'] = ( self._stats['total_processing_time'] / self._stats['successful_syntheses'] ) self._stats['total_audio_generated'] += len(result.audio_data) self._stats['voices_used'].add(result.voice.name) self._stats['languages_used'].add(result.voice.language) else: self._stats['failed_syntheses'] += 1 # Event Handlers @TrixyEvent("intent_received") def on_intent_received(self, event_name: str, event_data: Any) -> None: """Handle intent events that need TTS response.""" if not self.is_enabled(): return # This would typically be handled by an intent/response plugin # But we can provide basic TTS functionality here try: conversation_id = event_data.get('conversation_id', 'unknown') intent = event_data.get('intent', '') entities = event_data.get('entities', {}) # Example: Handle basic TTS requests if intent == "tts_request": text = entities.get('text', 'Hello, this is a test message') voice_name = entities.get('voice', self.default_voice_name) self.synthesize_text(text, voice_name, conversation_id) except Exception as e: pprint(f"Error handling intent: {e}") @TrixyEvent("system_startup") def on_system_startup(self, event_name: str, event_data: Any) -> None: """Handle system startup.""" if not self.is_enabled(): return pprint("TTS Plugin system startup") # Verify engines available_engines = [name for name, engine in self._engines.items() if engine['available']] if not available_engines: pprint("WARNING: No TTS engines available!") else: pprint(f"Available TTS engines: {available_engines}") # Public Methods def synthesize_text( self, text: str, voice_name: Optional[str] = None, conversation_id: str = "", speed: float = None, pitch: float = None, volume: float = None, use_ssml: bool = False ) -> bool: """ Synthesize text to speech. Args: text: Text to synthesize voice_name: Name of voice to use conversation_id: Conversation ID for response routing speed: Speech speed multiplier pitch: Pitch adjustment volume: Volume adjustment use_ssml: Whether text contains SSML markup Returns: bool: True if request was queued successfully """ try: # Get voice if voice_name and voice_name in self._voices: voice = self._voices[voice_name] else: voice = self._voices.get(self.default_voice_name) if not voice: voice = next(iter(self._voices.values())) # Create request request = TTSRequest( text=text, voice=voice, conversation_id=conversation_id, speed=speed or self.default_speed, pitch=pitch or self.default_pitch, volume=volume or self.default_volume, output_format=self.output_format, use_ssml=use_ssml and self.enable_ssml ) # Queue for processing self._synthesis_queue.put(request) pprint(f"TTS request queued: '{text[:30]}...' with voice {voice.name}") return True except Exception as e: pprint(f"Error queueing TTS request: {e}") return False def get_available_voices(self, language: Optional[str] = None, gender: Optional[str] = None) -> List[Dict[str, str]]: """Get list of available voices with optional filtering.""" voices = [] for voice_name, voice in self._voices.items(): if language and voice.language != language: continue if gender and voice.gender != gender: continue voices.append({ 'name': voice.name, 'language': voice.language, 'gender': voice.gender, 'region': voice.region, 'age': voice.age, 'style': voice.style, 'engine': voice.engine }) return voices def test_voice(self, voice_name: str, test_text: str = "This is a test message") -> Dict[str, Any]: """Test a specific voice.""" if voice_name not in self._voices: return {'success': False, 'error': f'Voice {voice_name} not found'} try: voice = self._voices[voice_name] request = TTSRequest( text=test_text, voice=voice, conversation_id="test", speed=self.default_speed, pitch=self.default_pitch, volume=self.default_volume, output_format=self.output_format ) result = self._synthesize_speech(request) return { 'success': result.success if result else False, 'voice': voice_name, 'text': test_text, 'processing_time': result.processing_time if result else 0, 'audio_size': len(result.audio_data) if result and result.success else 0, 'error': result.error_message if result and not result.success else None } except Exception as e: return {'success': False, 'error': str(e)} # Lifecycle Hooks def on_enable(self) -> None: """Called when plugin is enabled.""" pprint("TTS Plugin enabled") self._start_synthesis_thread() def on_disable(self) -> None: """Called when plugin is disabled.""" pprint("TTS Plugin disabled") # Clear synthesis queue while not self._synthesis_queue.empty(): try: self._synthesis_queue.get_nowait() except queue.Empty: break def cleanup(self) -> None: """Clean up TTS plugin resources.""" pprint("TTS Plugin cleanup") # Stop synthesis thread self._synthesis_shutdown.set() if self._synthesis_thread and self._synthesis_thread.is_alive(): self._synthesis_thread.join(timeout=5.0) # Clear queues and cache while not self._synthesis_queue.empty(): try: self._synthesis_queue.get_nowait() except queue.Empty: break with self._cache_lock: self._voice_cache.clear() # Save final statistics stats = self._stats.copy() stats['voices_used'] = list(stats['voices_used']) stats['languages_used'] = list(stats['languages_used']) self.set_config_value("final_stats", stats) def get_tts_stats(self) -> Dict[str, Any]: """Get TTS processing statistics.""" stats = self._stats.copy() # Convert sets to lists for JSON serialization stats['voices_used'] = list(stats['voices_used']) stats['languages_used'] = list(stats['languages_used']) # Add current status stats.update({ 'queue_size': self._synthesis_queue.qsize(), 'cache_size': len(self._voice_cache), 'cache_memory_mb': sum(len(data) for data in self._voice_cache.values()) / (1024 * 1024), 'available_voices': len(self._voices), 'available_engines': [name for name, engine in self._engines.items() if engine['available']], 'primary_engine': self.primary_engine, 'fallback_engine': self.fallback_engine }) return stats