"""TTS provider factory for creating and managing TTS providers.""" import logging from typing import Dict, List, Optional, Type from ..base.tts_provider_base import TTSProviderBase from ...domain.exceptions import SpeechSynthesisException logger = logging.getLogger(__name__) class TTSProviderFactory: """Factory for creating and managing TTS providers.""" def __init__(self): """Initialize the TTS provider factory.""" self._providers: Dict[str, Type[TTSProviderBase]] = {} self._provider_instances: Dict[str, TTSProviderBase] = {} self._register_default_providers() def _register_default_providers(self): """Register all available TTS providers.""" # Import providers dynamically to avoid import errors if dependencies are missing # Always register dummy provider as fallback from .dummy_provider import DummyTTSProvider self._providers['dummy'] = DummyTTSProvider # Register only Chatterbox provider try: from .chatterbox_provider import ChatterboxTTSProvider self._providers['chatterbox'] = ChatterboxTTSProvider logger.info("Registered Chatterbox TTS provider") except ImportError as e: logger.info(f"Chatterbox TTS provider not available: {e}") def get_available_providers(self) -> List[str]: """Get list of available TTS providers.""" logger.info("🔍 Checking availability of TTS providers...") available = [] for name, provider_class in self._providers.items(): logger.info(f"Checking provider: {name}") try: # Create instance if not cached if name not in self._provider_instances: logger.info(f"Creating instance for {name} provider") if name == 'chatterbox': self._provider_instances[name] = provider_class() else: self._provider_instances[name] = provider_class() # Check if provider is available logger.info(f"Checking availability for {name}") is_available = self._provider_instances[name].is_available() logger.info(f"Provider {name} availability: {'✅ Available' if is_available else '❌ Not Available'}") if is_available: available.append(name) except Exception as e: logger.warning(f"❌ Failed to check availability of {name} provider: {e}") logger.info(f"Provider check error details: {type(e).__name__}: {e}") logger.info(f"📋 Available TTS providers: {available}") return available def create_provider(self, provider_name: str, **kwargs) -> TTSProviderBase: """ Create a TTS provider instance. Args: provider_name: Name of the provider to create **kwargs: Additional arguments for provider initialization Returns: TTSProviderBase: The created provider instance Raises: SpeechSynthesisException: If provider is not available or creation fails """ if provider_name not in self._providers: available = list(self._providers.keys()) raise SpeechSynthesisException( f"Unknown TTS provider: {provider_name}. Available providers: {available}" ) # Check if provider is actually available before creating available_providers = self.get_available_providers() if provider_name not in available_providers: logger.warning(f"TTS provider {provider_name} is registered but not available") raise SpeechSynthesisException(f"TTS provider {provider_name} is not available") try: provider_class = self._providers[provider_name] # Create instance with appropriate parameters if provider_name == 'chatterbox': lang_code = kwargs.get('lang_code', 'en') provider = provider_class(lang_code=lang_code) else: provider = provider_class(**kwargs) # Verify the provider is available if not provider.is_available(): raise SpeechSynthesisException(f"TTS provider {provider_name} is not available") logger.info(f"Created TTS provider: {provider_name}") return provider except Exception as e: logger.error(f"Failed to create TTS provider {provider_name}: {e}") raise SpeechSynthesisException(f"Failed to create TTS provider {provider_name}: {e}") from e def get_provider_with_fallback(self, preferred_providers: List[str] = None, **kwargs) -> TTSProviderBase: """ Get a TTS provider with fallback logic. Args: preferred_providers: List of preferred providers in order of preference **kwargs: Additional arguments for provider initialization Returns: TTSProviderBase: The first available provider Raises: SpeechSynthesisException: If no providers are available """ if preferred_providers is None: preferred_providers = ['chatterbox', 'dummy'] logger.info(f"🔄 Getting TTS provider with fallback, preferred order: {preferred_providers}") available_providers = self.get_available_providers() # Try preferred providers in order for provider_name in preferred_providers: logger.info(f"🔍 Trying preferred provider: {provider_name}") if provider_name in available_providers: logger.info(f"✅ Provider {provider_name} is available, attempting to create...") try: provider = self.create_provider(provider_name, **kwargs) logger.info(f"🎉 Successfully created provider: {provider_name}") return provider except Exception as e: logger.warning(f"❌ Failed to create preferred provider {provider_name}: {e}") continue else: logger.info(f"❌ Provider {provider_name} is not in available providers list") # If no preferred providers work, try any available provider for provider_name in available_providers: if provider_name not in preferred_providers: try: return self.create_provider(provider_name, **kwargs) except Exception as e: logger.warning(f"Failed to create fallback provider {provider_name}: {e}") continue raise SpeechSynthesisException("No TTS providers are available") def get_provider_info(self, provider_name: str) -> Dict: """ Get information about a specific provider. Args: provider_name: Name of the provider Returns: Dict: Provider information including availability and supported features """ if provider_name not in self._providers: return {"available": False, "error": "Provider not registered"} try: # Create instance if not cached if provider_name not in self._provider_instances: provider_class = self._providers[provider_name] if provider_name == 'chatterbox': self._provider_instances[provider_name] = provider_class() else: self._provider_instances[provider_name] = provider_class() provider = self._provider_instances[provider_name] return { "available": provider.is_available(), "name": provider.provider_name, "supported_languages": provider.supported_languages, "available_voices": provider.get_available_voices() if provider.is_available() else [] } except Exception as e: return { "available": False, "error": str(e) } def cleanup_providers(self): """Clean up provider instances and resources.""" for provider in self._provider_instances.values(): try: if hasattr(provider, '_cleanup_temp_files'): provider._cleanup_temp_files() except Exception as e: logger.warning(f"Failed to cleanup provider {provider.provider_name}: {e}") self._provider_instances.clear() logger.info("Cleaned up TTS provider instances")