File size: 3,182 Bytes
3ed3b5a
 
 
 
 
 
 
 
58d9769
3ed3b5a
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
58d9769
 
3ed3b5a
 
 
58d9769
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
3ed3b5a
58d9769
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
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)