import logging from typing import Optional, List # Configure logging logger = logging.getLogger(__name__) # Import the base class from utils.tts_base import TTSEngineBase, DummyTTSEngine from utils.tts_cascading import CascadingTTSEngine class TTSFactory: """Factory class for creating TTS engines This class is responsible for creating the appropriate TTS engine based on availability and configuration. """ @staticmethod def create_engine(engine_type: Optional[str] = None, lang_code: str = 'z') -> TTSEngineBase: """Create a TTS engine instance Args: engine_type (str, optional): Type of engine to create ('kokoro', 'kokoro_space', 'dia', 'dummy') If None, the best available engine will be used lang_code (str): Language code for the engine Returns: TTSEngineBase: An instance of a TTS engine """ from utils.tts_engines import get_available_engines, create_engine # Get available engines available_engines = get_available_engines() logger.info(f"Available TTS engines: {available_engines}") # If engine_type is specified, try to create that specific engine if engine_type is not None: if engine_type in available_engines: logger.info(f"Creating requested engine: {engine_type}") engine = create_engine(engine_type, lang_code) return engine else: logger.warning(f"Requested engine '{engine_type}' is not available") # Fall back to dummy engine if no engines are available if not available_engines or (len(available_engines) == 1 and available_engines[0] == 'dummy'): logger.warning("No TTS engines available, falling back to dummy engine") return DummyTTSEngine(lang_code) return TTSFactory.create_cascading_engine(available_engines, lang_code) @staticmethod def create_cascading_engine(available_engines: List[str], lang_code: str = 'z') -> TTSEngineBase: """Create a cascading TTS engine that tries multiple engines in order Args: available_engines (List[str]): List of available engine names lang_code (str): Language code for the engines Returns: TTSEngineBase: A cascading TTS engine instance """ from utils.tts_engines import create_engine # Define the priority order for engines priority_order = ['kokoro', 'kokoro_space', 'dia', 'dia_space', 'dummy'] # Filter and sort available engines by priority engines_by_priority = [engine for engine in priority_order if engine in available_engines] # Always ensure dummy is the last fallback if 'dummy' not in engines_by_priority: engines_by_priority.append('dummy') logger.info(f"Creating cascading engine with priority: {engines_by_priority}") return CascadingTTSEngine(engines_by_priority, lang_code)