refator tts part

utils/tts.py

@@ -1,85 +1,52 @@
-import os
 import logging
-import time
-import soundfile as sf
-from gradio_client import Client
-
 
+# Configure logging
 logger = logging.getLogger(__name__)
 
-# Flag to track TTS engine availability
-KOKORO_AVAILABLE = False
-KOKORO_SPACE_AVAILABLE = True
-DIA_AVAILABLE = False
+# Import from the new factory pattern implementation
+from utils.tts_factory import get_tts_engine, generate_speech, TTSFactory
+from utils.tts_engines import get_available_engines
 
-# Try to import Kokoro first
-try:
-    from kokoro import KPipeline
-    KOKORO_AVAILABLE = True
-    logger.info("Kokoro TTS engine is available")
-except AttributeError as e:
-    # Specifically catch the EspeakWrapper.set_data_path error
-    if "EspeakWrapper" in str(e) and "set_data_path" in str(e):
-        logger.warning("Kokoro import failed due to EspeakWrapper.set_data_path issue, falling back to Kokoro FastAPI server")
-    else:
-        # Re-raise if it's a different error
-        logger.error(f"Kokoro import failed with unexpected error: {str(e)}")
-        raise
-except ImportError:
-    logger.warning("Kokoro TTS engine is not available")
+# For backward compatibility
+from utils.tts_engines import KOKORO_AVAILABLE, KOKORO_SPACE_AVAILABLE, DIA_AVAILABLE
 
+# Backward compatibility class
 class TTSEngine:
+    """Legacy TTSEngine class for backward compatibility
+    
+    This class is maintained for backward compatibility with existing code.
+    New code should use the factory pattern implementation directly.
+    """
+    
     def __init__(self, lang_code='z'):
-        """Initialize TTS Engine with Kokoro or Dia as fallback
+        """Initialize TTS Engine using the factory pattern
         
         Args:
             lang_code (str): Language code ('a' for US English, 'b' for British English,
                            'j' for Japanese, 'z' for Mandarin Chinese)
-                           Note: lang_code is only used for Kokoro, not for Dia
         """
-        logger.info("Initializing TTS Engine")
+        logger.info("Initializing legacy TTSEngine wrapper")
         logger.info(f"Available engines - Kokoro: {KOKORO_AVAILABLE}, Dia: {DIA_AVAILABLE}")
-        self.engine_type = None
         
-        if KOKORO_AVAILABLE:
-            logger.info(f"Using Kokoro as primary TTS engine with language code: {lang_code}")
-            try:
-                self.pipeline = KPipeline(lang_code=lang_code)
-                self.engine_type = "kokoro"
-                logger.info("TTS engine successfully initialized with Kokoro")
-            except Exception as kokoro_err:
-                logger.error(f"Failed to initialize Kokoro pipeline: {str(kokoro_err)}")
-                logger.error(f"Error type: {type(kokoro_err).__name__}")
-                logger.info("Will try to fall back to Dia TTS engine")
-
-        if KOKORO_SPACE_AVAILABLE:
-            logger.info(f"Using Kokoro FastAPI server as primary TTS engine with language code: {lang_code}")
-            try:
-                self.client = Client("Remsky/Kokoro-TTS-Zero")
-                self.engine_type = "kokoro_space"
-                logger.info("TTS engine successfully initialized with Kokoro FastAPI server")
-            except Exception as kokoro_err:
-                logger.error(f"Failed to initialize Kokoro space: {str(kokoro_err)}")
-                logger.error(f"Error type: {type(kokoro_err).__name__}")
-                logger.info("Will try to fall back to Dia TTS engine")
-
-        # Try Dia if Kokoro is not available or failed to initialize
-        if self.engine_type is None and DIA_AVAILABLE:
-            logger.info("Using Dia as fallback TTS engine")
-            # For Dia, we don't need to initialize anything here
-            # The model will be lazy-loaded when needed
-            self.pipeline = None
-            self.client = None
-            self.engine_type = "dia"
-            logger.info("TTS engine initialized with Dia (lazy loading)")
+        # Create the appropriate engine using the factory
+        self._engine = TTSFactory.create_engine(lang_code=lang_code)
         
-        # Use dummy if no TTS engines are available
-        if self.engine_type is None:
-            logger.warning("Using dummy TTS implementation as no TTS engines are available")
-            logger.warning("Check logs above for specific errors that prevented Kokoro or Dia initialization")
-            self.pipeline = None
-            self.client = None
+        # Set engine_type for backward compatibility
+        engine_class = self._engine.__class__.__name__
+        if 'Kokoro' in engine_class and 'Space' in engine_class:
+            self.engine_type = "kokoro_space"
+        elif 'Kokoro' in engine_class:
+            self.engine_type = "kokoro"
+        elif 'Dia' in engine_class:
+            self.engine_type = "dia"
+        else:
             self.engine_type = "dummy"
+            
+        # Set pipeline and client attributes for backward compatibility
+        self.pipeline = getattr(self._engine, 'pipeline', None)
+        self.client = getattr(self._engine, 'client', None)
+        
+        logger.info(f"Legacy TTSEngine wrapper initialized with engine type: {self.engine_type}")
 
     def generate_speech(self, text: str, voice: str = 'af_heart', speed: float = 1.0) -> str:
         """Generate speech from text using available TTS engine
@@ -87,272 +54,79 @@ class TTSEngine:
         Args:
             text (str): Input text to synthesize
             voice (str): Voice ID to use (e.g., 'af_heart', 'af_bella', etc.)
-                         Note: voice parameter is only used for Kokoro, not for Dia
             speed (float): Speech speed multiplier (0.5 to 2.0)
-                           Note: speed parameter is only used for Kokoro, not for Dia
             
         Returns:
             str: Path to the generated audio file
         """
-        logger.info(f"Generating speech for text length: {len(text)}")
-        
-        try:
-            # Create output directory if it doesn't exist
-            os.makedirs("temp/outputs", exist_ok=True)
-            
-            # Generate unique output path
-            output_path = f"temp/outputs/output_{int(time.time())}.wav"
-            
-            # Use the appropriate TTS engine based on availability
-            if self.engine_type == "kokoro":
-                # Use Kokoro for TTS generation
-                generator = self.pipeline(text, voice=voice, speed=speed)
-                for _, _, audio in generator:
-                    logger.info(f"Saving Kokoro audio to {output_path}")
-                    sf.write(output_path, audio, 24000)
-                    break
-            elif self.engine_type == "kokoro_space":
-                # Use Kokoro FastAPI server for TTS generation
-                logger.info("Generating speech using Kokoro FastAPI server")
-                logger.info(f"text to generate speech on is: {text}")
-                try:
-                    result = self.client.predict(
-                            text=text,
-                            voice_names='af_nova',
-                            speed=speed,
-                            api_name="/generate_speech_from_ui"
-                    )
-                    logger.info(f"Received audio from Kokoro FastAPI server: {result}")
-                except Exception as e:
-                    logger.error(f"Failed to generate speech from Kokoro FastAPI server: {str(e)}")
-                    logger.error(f"Error type: {type(e).__name__}")
-                    logger.info("Falling back to dummy audio generation")
-            elif self.engine_type == "dia":
-                # Use Dia for TTS generation
-                try:
-                    logger.info("Attempting to use Dia TTS for speech generation")
-                    # Import here to avoid circular imports
-                    try:
-                        logger.info("Importing Dia speech generation module")
-                        from utils.tts_dia import generate_speech as dia_generate_speech
-                        logger.info("Successfully imported Dia speech generation function")
-                    except ImportError as import_err:
-                        logger.error(f"Failed to import Dia speech generation function: {str(import_err)}")
-                        logger.error(f"Import path: {import_err.__traceback__.tb_frame.f_globals.get('__name__', 'unknown')}")
-                        raise
-                    
-                    # Call Dia's generate_speech function
-                    logger.info("Calling Dia's generate_speech function")
-                    output_path = dia_generate_speech(text)
-                    logger.info(f"Generated audio with Dia: {output_path}")
-                except ImportError as import_err:
-                    logger.error(f"Dia TTS generation failed due to import error: {str(import_err)}")
-                    logger.error("Falling back to dummy audio generation")
-                    return self._generate_dummy_audio(output_path)
-                except Exception as dia_error:
-                    logger.error(f"Dia TTS generation failed: {str(dia_error)}", exc_info=True)
-                    logger.error(f"Error type: {type(dia_error).__name__}")
-                    logger.error("Falling back to dummy audio generation")
-                    # Fall back to dummy audio if Dia fails
-                    return self._generate_dummy_audio(output_path)
-            else:
-                # Generate dummy audio as fallback
-                return self._generate_dummy_audio(output_path)
-            
-            logger.info(f"Audio generation complete: {output_path}")
-            return output_path
-
-        except Exception as e:
-            logger.error(f"TTS generation failed: {str(e)}", exc_info=True)
-            raise
-            
-    def _generate_dummy_audio(self, output_path):
-        """Generate a dummy audio file with a simple sine wave
-        
-        Args:
-            output_path (str): Path to save the dummy audio file
-            
-        Returns:
-            str: Path to the generated dummy audio file
-        """
-        import numpy as np
-        sample_rate = 24000
-        duration = 3.0  # seconds
-        t = np.linspace(0, duration, int(sample_rate * duration), False)
-        tone = np.sin(2 * np.pi * 440 * t) * 0.3
-        
-        logger.info(f"Saving dummy audio to {output_path}")
-        sf.write(output_path, tone, sample_rate)
-        logger.info(f"Dummy audio generation complete: {output_path}")
-        return output_path
-
+        logger.info(f"Legacy TTSEngine wrapper calling generate_speech for text length: {len(text)}")
+        return self._engine.generate_speech(text, voice, speed)
+    
     def generate_speech_stream(self, text: str, voice: str = 'af_heart', speed: float = 1.0):
         """Generate speech from text and yield each segment
         
         Args:
             text (str): Input text to synthesize
-            voice (str): Voice ID to use (e.g., 'af_heart', 'af_bella', etc.)
-            speed (float): Speech speed multiplier (0.5 to 2.0)
+            voice (str): Voice ID to use
+            speed (float): Speech speed multiplier
             
         Yields:
             tuple: (sample_rate, audio_data) pairs for each segment
         """
-        try:
-            # Use the appropriate TTS engine based on availability
-            if self.engine_type == "kokoro":
-                # Use Kokoro for streaming TTS
-                generator = self.pipeline(text, voice=voice, speed=speed)
-                for _, _, audio in generator:
-                    yield 24000, audio
-            elif self.engine_type == "dia":
-                # Dia doesn't support streaming natively, so we generate the full audio
-                # and then yield it as a single chunk
-                try:
-                    logger.info("Attempting to use Dia TTS for speech streaming")
-                    # Import here to avoid circular imports
-                    try:
-                        logger.info("Importing required modules for Dia streaming")
-                        import torch
-                        logger.info("PyTorch successfully imported for Dia streaming")
-                        
-                        try:
-                            from utils.tts_dia import _get_model, DEFAULT_SAMPLE_RATE
-                            logger.info("Successfully imported Dia model and sample rate")
-                        except ImportError as import_err:
-                            logger.error(f"Failed to import Dia model for streaming: {str(import_err)}")
-                            logger.error(f"Import path: {import_err.__traceback__.tb_frame.f_globals.get('__name__', 'unknown')}")
-                            raise
-                    except ImportError as torch_err:
-                        logger.error(f"PyTorch import failed for Dia streaming: {str(torch_err)}")
-                        raise
-                    
-                    # Get the Dia model
-                    logger.info("Getting Dia model instance")
-                    try:
-                        model = _get_model()
-                        logger.info("Successfully obtained Dia model instance")
-                    except Exception as model_err:
-                        logger.error(f"Failed to get Dia model instance: {str(model_err)}")
-                        logger.error(f"Error type: {type(model_err).__name__}")
-                        raise
-                    
-                    # Generate audio
-                    logger.info("Generating audio with Dia model")
-                    with torch.inference_mode():
-                        output_audio_np = model.generate(
-                            text,
-                            max_tokens=None,
-                            cfg_scale=3.0,
-                            temperature=1.3,
-                            top_p=0.95,
-                            cfg_filter_top_k=35,
-                            use_torch_compile=False,
-                            verbose=False
-                        )
-                    
-                    if output_audio_np is not None:
-                        logger.info(f"Successfully generated audio with Dia (length: {len(output_audio_np)})")
-                        yield DEFAULT_SAMPLE_RATE, output_audio_np
-                    else:
-                        logger.warning("Dia model returned None for audio output")
-                        logger.warning("Falling back to dummy audio stream")
-                        # Fall back to dummy audio if Dia fails
-                        yield from self._generate_dummy_audio_stream()
-                except ImportError as import_err:
-                    logger.error(f"Dia TTS streaming failed due to import error: {str(import_err)}")
-                    logger.error("Falling back to dummy audio stream")
-                    # Fall back to dummy audio if Dia fails
-                    yield from self._generate_dummy_audio_stream()
-                except Exception as dia_error:
-                    logger.error(f"Dia TTS streaming failed: {str(dia_error)}", exc_info=True)
-                    logger.error(f"Error type: {type(dia_error).__name__}")
-                    logger.error("Falling back to dummy audio stream")
-                    # Fall back to dummy audio if Dia fails
-                    yield from self._generate_dummy_audio_stream()
-            else:
-                # Generate dummy audio chunks as fallback
-                yield from self._generate_dummy_audio_stream()
-                
-        except Exception as e:
-            logger.error(f"TTS streaming failed: {str(e)}", exc_info=True)
-            raise
+        logger.info(f"Legacy TTSEngine wrapper calling generate_speech_stream for text length: {len(text)}")
+        yield from self._engine.generate_speech_stream(text, voice, speed)
+    
+    # For backward compatibility
+    def _generate_dummy_audio(self, output_path):
+        """Generate a dummy audio file with a simple sine wave (backward compatibility)
+        
+        Args:
+            output_path (str): Path to save the dummy audio file
             
+        Returns:
+            str: Path to the generated dummy audio file
+        """
+        from utils.tts_base import DummyTTSEngine
+        dummy_engine = DummyTTSEngine()
+        return dummy_engine.generate_speech("", "", 1.0)
+    
+    # For backward compatibility
     def _generate_dummy_audio_stream(self):
-        """Generate dummy audio chunks with simple sine waves
+        """Generate dummy audio chunks (backward compatibility)
         
         Yields:
             tuple: (sample_rate, audio_data) pairs for each dummy segment
         """
-        import numpy as np
-        sample_rate = 24000
-        duration = 1.0  # seconds per chunk
-        
-        # Create 3 chunks of dummy audio
-        for i in range(3):
-            t = np.linspace(0, duration, int(sample_rate * duration), False)
-            freq = 440 + (i * 220)  # Different frequency for each chunk
-            tone = np.sin(2 * np.pi * freq * t) * 0.3
-            yield sample_rate, tone
+        from utils.tts_base import DummyTTSEngine
+        dummy_engine = DummyTTSEngine()
+        yield from dummy_engine.generate_speech_stream("", "", 1.0)
 
-# Initialize TTS engine with cache decorator if using Streamlit
-def get_tts_engine(lang_code='a'):
-    """Get or create TTS engine instance
-    
-    Args:
-        lang_code (str): Language code for the pipeline
-        
-    Returns:
-        TTSEngine: Initialized TTS engine instance
-    """
-    logger.info(f"Requesting TTS engine with language code: {lang_code}")
-    try:
-        import streamlit as st
-        logger.info("Streamlit detected, using cached TTS engine")
-        @st.cache_resource
-        def _get_engine():
-            logger.info("Creating cached TTS engine instance")
-            engine = TTSEngine(lang_code)
-            logger.info(f"Cached TTS engine created with type: {engine.engine_type}")
-            return engine
-        
-        engine = _get_engine()
-        logger.info(f"Retrieved TTS engine from cache with type: {engine.engine_type}")
-        return engine
-    except ImportError:
-        logger.info("Streamlit not available, creating direct TTS engine instance")
-        engine = TTSEngine(lang_code)
-        logger.info(f"Direct TTS engine created with type: {engine.engine_type}")
-        return engine
+# Import the new implementations from tts_base
+# These functions are already defined in tts_base.py and imported at the top of this file
+# They are kept here as comments for reference
 
-def generate_speech(text: str, voice: str = 'af_heart', speed: float = 1.0) -> str:
-    """Public interface for TTS generation
-    
-    Args:
-        text (str): Input text to synthesize
-        voice (str): Voice ID to use
-        speed (float): Speech speed multiplier
-        
-    Returns:
-        str: Path to generated audio file
-    """
-    logger.info(f"Public generate_speech called with text length: {len(text)}, voice: {voice}, speed: {speed}")
-    try:
-        # Get the TTS engine
-        logger.info("Getting TTS engine instance")
-        engine = get_tts_engine()
-        logger.info(f"Using TTS engine type: {engine.engine_type}")
-        
-        # Generate speech
-        logger.info("Calling engine.generate_speech")
-        output_path = engine.generate_speech(text, voice, speed)
-        logger.info(f"Speech generation complete, output path: {output_path}")
-        return output_path
-    except Exception as e:
-        logger.error(f"Error in public generate_speech function: {str(e)}", exc_info=True)
-        logger.error(f"Error type: {type(e).__name__}")
-        if hasattr(e, '__traceback__'):
-            tb = e.__traceback__
-            while tb.tb_next:
-                tb = tb.tb_next
-            logger.error(f"Error occurred in file: {tb.tb_frame.f_code.co_filename}, line {tb.tb_lineno}")
-        raise
+# def get_tts_engine(lang_code='a'):
+#     """Get or create TTS engine instance
+#     
+#     Args:
+#         lang_code (str): Language code for the pipeline
+#         
+#     Returns:
+#         TTSEngineBase: Initialized TTS engine instance
+#     """
+#     # Implementation moved to tts_base.py
+#     pass
+
+# def generate_speech(text: str, voice: str = 'af_heart', speed: float = 1.0) -> str:
+#     """Public interface for TTS generation
+#     
+#     Args:
+#         text (str): Input text to synthesize
+#         voice (str): Voice ID to use
+#         speed (float): Speech speed multiplier
+#         
+#     Returns:
+#         str: Path to generated audio file
+#     "\"""  
+#     # Implementation moved to tts_base.py
+#     pass

utils/tts_base.py

@@ -0,0 +1,152 @@
+import os
+import time
+import logging
+import soundfile as sf
+import numpy as np
+from abc import ABC, abstractmethod
+from typing import Tuple, Generator, Optional
+
+# Configure logging
+logger = logging.getLogger(__name__)
+
+class TTSEngineBase(ABC):
+    """Base class for all TTS engines
+    
+    This abstract class defines the interface that all TTS engines must implement.
+    It also provides common utility methods for file handling and audio generation.
+    """
+    
+    def __init__(self, lang_code: str = 'z'):
+        """Initialize the TTS engine
+        
+        Args:
+            lang_code (str): Language code ('a' for US English, 'b' for British English,
+                           'j' for Japanese, 'z' for Mandarin Chinese)
+                           Note: Not all engines support all language codes
+        """
+        self.lang_code = lang_code
+        logger.info(f"Initializing {self.__class__.__name__} with language code: {lang_code}")
+    
+    @abstractmethod
+    def generate_speech(self, text: str, voice: str = 'af_heart', speed: float = 1.0) -> str:
+        """Generate speech from text
+        
+        Args:
+            text (str): Input text to synthesize
+            voice (str): Voice ID to use (e.g., 'af_heart', 'af_bella', etc.)
+                         Note: Not all engines support all voices
+            speed (float): Speech speed multiplier (0.5 to 2.0)
+                           Note: Not all engines support speed adjustment
+            
+        Returns:
+            str: Path to the generated audio file
+        """
+        pass
+    
+    def generate_speech_stream(self, text: str, voice: str = 'af_heart', speed: float = 1.0) -> Generator[Tuple[int, np.ndarray], None, None]:
+        """Generate speech from text and yield each segment
+        
+        Args:
+            text (str): Input text to synthesize
+            voice (str): Voice ID to use
+            speed (float): Speech speed multiplier
+            
+        Yields:
+            tuple: (sample_rate, audio_data) pairs for each segment
+        """
+        # Default implementation: generate full audio and yield as a single chunk
+        output_path = self.generate_speech(text, voice, speed)
+        audio_data, sample_rate = sf.read(output_path)
+        yield sample_rate, audio_data
+    
+    def _create_output_dir(self) -> str:
+        """Create output directory for audio files
+        
+        Returns:
+            str: Path to the output directory
+        """
+        output_dir = "temp/outputs"
+        os.makedirs(output_dir, exist_ok=True)
+        return output_dir
+    
+    def _generate_output_path(self, prefix: str = "output") -> str:
+        """Generate a unique output path for audio files
+        
+        Args:
+            prefix (str): Prefix for the output filename
+            
+        Returns:
+            str: Path to the output file
+        """
+        output_dir = self._create_output_dir()
+        timestamp = int(time.time())
+        return f"{output_dir}/{prefix}_{timestamp}.wav"
+
+
+class DummyTTSEngine(TTSEngineBase):
+    """Dummy TTS engine that generates a simple sine wave
+    
+    This engine is used as a fallback when no other engines are available.
+    """
+    
+    def __init__(self, lang_code: str = 'z'):
+        super().__init__(lang_code)
+        logger.warning("Using dummy TTS implementation as no other engines are available")
+    
+    def generate_speech(self, text: str, voice: str = 'af_heart', speed: float = 1.0) -> str:
+        """Generate a dummy audio file with a simple sine wave
+        
+        Args:
+            text (str): Input text (not used)
+            voice (str): Voice ID (not used)
+            speed (float): Speed multiplier (not used)
+            
+        Returns:
+            str: Path to the generated dummy audio file
+        """
+        logger.info(f"Generating dummy speech for text length: {len(text)}")
+        
+        # Generate unique output path
+        output_path = self._generate_output_path("dummy")
+        
+        # Generate a simple sine wave
+        sample_rate = 24000
+        duration = 3.0  # seconds
+        t = np.linspace(0, duration, int(sample_rate * duration), False)
+        tone = np.sin(2 * np.pi * 440 * t) * 0.3
+        
+        # Save the audio file
+        logger.info(f"Saving dummy audio to {output_path}")
+        sf.write(output_path, tone, sample_rate)
+        logger.info(f"Dummy audio generation complete: {output_path}")
+        
+        return output_path
+    
+    def generate_speech_stream(self, text: str, voice: str = 'af_heart', speed: float = 1.0) -> Generator[Tuple[int, np.ndarray], None, None]:
+        """Generate dummy audio chunks with simple sine waves
+        
+        Args:
+            text (str): Input text (not used)
+            voice (str): Voice ID (not used)
+            speed (float): Speed multiplier (not used)
+            
+        Yields:
+            tuple: (sample_rate, audio_data) pairs for each dummy segment
+        """
+        logger.info(f"Generating dummy speech stream for text length: {len(text)}")
+        
+        sample_rate = 24000
+        duration = 1.0  # seconds per chunk
+        
+        # Create 3 chunks of dummy audio
+        for i in range(3):
+            t = np.linspace(0, duration, int(sample_rate * duration), False)
+            freq = 440 + (i * 220)  # Different frequency for each chunk
+            tone = np.sin(2 * np.pi * freq * t) * 0.3
+            yield sample_rate, tone
+
+
+# Factory functionality moved to tts_factory.py to avoid circular imports
+
+
+# Note: Backward compatibility functions moved to tts_factory.py

utils/tts_dia.py

@@ -68,6 +68,9 @@ def _get_model() -> Dia:
 def generate_speech(text: str, language: str = "zh") -> str:
     """Public interface for TTS generation using Dia model
     
+    This is a legacy function maintained for backward compatibility.
+    New code should use the factory pattern implementation directly.
+    
     Args:
         text (str): Input text to synthesize
         language (str): Language code (not used in Dia model, kept for API compatibility)
@@ -75,122 +78,18 @@ def generate_speech(text: str, language: str = "zh") -> str:
     Returns:
         str: Path to the generated audio file
     """
-    logger.info(f"Generating speech for text length: {len(text)}")
-    logger.info(f"Text content (first 50 chars): {text[:50]}...")
-    
-    # Create output directory if it doesn't exist
-    output_dir = "temp/outputs"
-    logger.info(f"Ensuring output directory exists: {output_dir}")
-    try:
-        os.makedirs(output_dir, exist_ok=True)
-        logger.info(f"Output directory ready: {output_dir}")
-    except PermissionError as perm_err:
-        logger.error(f"Permission error creating output directory: {perm_err}")
-        # Fall back to dummy TTS
-        logger.info("Falling back to dummy TTS due to directory creation error")
-        from utils.tts_dummy import generate_speech as dummy_generate_speech
-        return dummy_generate_speech(text, language)
-    except Exception as dir_err:
-        logger.error(f"Error creating output directory: {dir_err}")
-        # Fall back to dummy TTS
-        logger.info("Falling back to dummy TTS due to directory creation error")
-        from utils.tts_dummy import generate_speech as dummy_generate_speech
-        return dummy_generate_speech(text, language)
-    
-    # Generate unique output path
-    timestamp = int(time.time())
-    output_path = f"{output_dir}/output_{timestamp}.wav"
-    logger.info(f"Output will be saved to: {output_path}")
-    
-    # Get the model
-    logger.info("Retrieving Dia model instance")
-    try:
-        model = _get_model()
-        logger.info("Successfully retrieved Dia model instance")
-    except Exception as model_err:
-        logger.error(f"Failed to get Dia model: {model_err}")
-        logger.error(f"Error type: {type(model_err).__name__}")
-        # Fall back to dummy TTS
-        logger.info("Falling back to dummy TTS due to model loading error")
-        from utils.tts_dummy import generate_speech as dummy_generate_speech
-        return dummy_generate_speech(text, language)
+    logger.info(f"Legacy Dia generate_speech called with text length: {len(text)}")
     
-    # Generate audio
-    logger.info("Starting audio generation with Dia model")
-    start_time = time.time()
+    # Use the new implementation via factory pattern
+    from utils.tts_engines import DiaTTSEngine
     
     try:
-        with torch.inference_mode():
-            logger.info("Calling model.generate() with inference_mode")
-            output_audio_np = model.generate(
-                text,
-                max_tokens=None,  # Use default from model config
-                cfg_scale=3.0,
-                temperature=1.3,
-                top_p=0.95,
-                cfg_filter_top_k=35,
-                use_torch_compile=False,  # Keep False for stability
-                verbose=False
-            )
-            logger.info("Model.generate() completed")
-    except RuntimeError as rt_err:
-        logger.error(f"Runtime error during generation: {rt_err}")
-        if "CUDA out of memory" in str(rt_err):
-            logger.error("CUDA out of memory error - consider reducing batch size or model size")
+        # Create a Dia engine and generate speech
+        dia_engine = DiaTTSEngine(language)
+        return dia_engine.generate_speech(text)
+    except Exception as e:
+        logger.error(f"Error in legacy Dia generate_speech: {str(e)}", exc_info=True)
         # Fall back to dummy TTS
-        logger.info("Falling back to dummy TTS due to runtime error during generation")
-        from utils.tts_dummy import generate_speech as dummy_generate_speech
-        return dummy_generate_speech(text, language)
-    except Exception as gen_err:
-        logger.error(f"Error during audio generation: {gen_err}")
-        logger.error(f"Error type: {type(gen_err).__name__}")
-        # Fall back to dummy TTS
-        logger.info("Falling back to dummy TTS due to error during generation")
-        from utils.tts_dummy import generate_speech as dummy_generate_speech
-        return dummy_generate_speech(text, language)
-    
-    end_time = time.time()
-    generation_time = end_time - start_time
-    logger.info(f"Generation finished in {generation_time:.2f} seconds")
-    
-    # Process the output
-    if output_audio_np is not None:
-        logger.info(f"Generated audio array shape: {output_audio_np.shape}, dtype: {output_audio_np.dtype}")
-        logger.info(f"Audio stats - min: {output_audio_np.min():.4f}, max: {output_audio_np.max():.4f}, mean: {output_audio_np.mean():.4f}")
-        
-        # Apply a slight slowdown for better quality (0.94x speed)
-        speed_factor = 0.94
-        original_len = len(output_audio_np)
-        target_len = int(original_len / speed_factor)
-        
-        logger.info(f"Applying speed adjustment factor: {speed_factor}")
-        if target_len != original_len and target_len > 0:
-            try:
-                x_original = np.arange(original_len)
-                x_resampled = np.linspace(0, original_len - 1, target_len)
-                output_audio_np = np.interp(x_resampled, x_original, output_audio_np)
-                logger.info(f"Resampled audio from {original_len} to {target_len} samples for {speed_factor:.2f}x speed")
-            except Exception as resample_err:
-                logger.error(f"Error during audio resampling: {resample_err}")
-                logger.warning("Using original audio without resampling")
-        
-        # Save the audio file
-        logger.info(f"Saving audio to file: {output_path}")
-        try:
-            sf.write(output_path, output_audio_np, DEFAULT_SAMPLE_RATE)
-            logger.info(f"Audio successfully saved to {output_path}")
-        except Exception as save_err:
-            logger.error(f"Error saving audio file: {save_err}")
-            logger.error(f"Error type: {type(save_err).__name__}")
-            # Fall back to dummy TTS
-            logger.info("Falling back to dummy TTS due to error saving audio file")
-            from utils.tts_dummy import generate_speech as dummy_generate_speech
-            return dummy_generate_speech(text, language)
-        
-        return output_path
-    else:
-        logger.warning("Generation produced no output (None returned from model)")
-        logger.warning("This may indicate a model configuration issue or empty input text")
-        dummy_path = f"{output_dir}/dummy_{timestamp}.wav"
-        logger.warning(f"Returning dummy audio path: {dummy_path}")
-        return dummy_path
+        from utils.tts_base import DummyTTSEngine
+        dummy_engine = DummyTTSEngine()
+        return dummy_engine.generate_speech(text)

utils/tts_dummy.py

@@ -1,25 +1,11 @@
 def generate_speech(text: str, language: str = "zh") -> str:
-    """Public interface for TTS generation"""
-    import os
-    import numpy as np
-    import soundfile as sf
-    import time
+    """Public interface for TTS generation
     
-    # Create output directory if it doesn't exist
-    output_dir = "temp/outputs"
-    os.makedirs(output_dir, exist_ok=True)
+    This is a legacy function maintained for backward compatibility.
+    New code should use the factory pattern implementation directly.
+    """
+    from utils.tts_base import DummyTTSEngine
     
-    # Generate a unique filename
-    timestamp = int(time.time())
-    output_path = f"{output_dir}/dummy_{timestamp}.wav"
-    
-    # Generate a simple sine wave as dummy audio
-    sample_rate = 24000
-    duration = 2.0  # seconds
-    t = np.linspace(0, duration, int(sample_rate * duration), False)
-    tone = np.sin(2 * np.pi * 440 * t) * 0.3
-    
-    # Save the audio file
-    sf.write(output_path, tone, sample_rate)
-    
-    return output_path
+    # Create a dummy engine and generate speech
+    dummy_engine = DummyTTSEngine()
+    return dummy_engine.generate_speech(text, "af_heart", 1.0)

utils/tts_engines.py

@@ -0,0 +1,322 @@
+import logging
+import time
+import os
+import numpy as np
+import soundfile as sf
+from typing import Dict, List, Optional, Tuple, Generator, Any
+
+from utils.tts_base import TTSEngineBase, DummyTTSEngine
+
+# Configure logging
+logger = logging.getLogger(__name__)
+
+# Flag to track TTS engine availability
+KOKORO_AVAILABLE = False
+KOKORO_SPACE_AVAILABLE = True
+DIA_AVAILABLE = False
+
+# Try to import Kokoro
+try:
+    from kokoro import KPipeline
+    KOKORO_AVAILABLE = True
+    logger.info("Kokoro TTS engine is available")
+except AttributeError as e:
+    # Specifically catch the EspeakWrapper.set_data_path error
+    if "EspeakWrapper" in str(e) and "set_data_path" in str(e):
+        logger.warning("Kokoro import failed due to EspeakWrapper.set_data_path issue, falling back to Kokoro FastAPI server")
+    else:
+        # Re-raise if it's a different error
+        logger.error(f"Kokoro import failed with unexpected error: {str(e)}")
+        raise
+except ImportError:
+    logger.warning("Kokoro TTS engine is not available")
+
+# Try to import Dia dependencies to check availability
+try:
+    import torch
+    from dia.model import Dia
+    DIA_AVAILABLE = True
+    logger.info("Dia TTS engine is available")
+except ImportError:
+    logger.warning("Dia TTS engine is not available")
+
+
+class KokoroTTSEngine(TTSEngineBase):
+    """Kokoro TTS engine implementation
+    
+    This engine uses the Kokoro library for TTS generation.
+    """
+    
+    def __init__(self, lang_code: str = 'z'):
+        super().__init__(lang_code)
+        try:
+            self.pipeline = KPipeline(lang_code=lang_code)
+            logger.info("Kokoro TTS engine successfully initialized")
+        except Exception as e:
+            logger.error(f"Failed to initialize Kokoro pipeline: {str(e)}")
+            logger.error(f"Error type: {type(e).__name__}")
+            raise
+    
+    def generate_speech(self, text: str, voice: str = 'af_heart', speed: float = 1.0) -> str:
+        """Generate speech using Kokoro TTS engine
+        
+        Args:
+            text (str): Input text to synthesize
+            voice (str): Voice ID to use (e.g., 'af_heart', 'af_bella', etc.)
+            speed (float): Speech speed multiplier (0.5 to 2.0)
+            
+        Returns:
+            str: Path to the generated audio file
+        """
+        logger.info(f"Generating speech with Kokoro for text length: {len(text)}")
+        
+        # Generate unique output path
+        output_path = self._generate_output_path()
+        
+        # Generate speech
+        generator = self.pipeline(text, voice=voice, speed=speed)
+        for _, _, audio in generator:
+            logger.info(f"Saving Kokoro audio to {output_path}")
+            sf.write(output_path, audio, 24000)
+            break
+        
+        logger.info(f"Kokoro audio generation complete: {output_path}")
+        return output_path
+    
+    def generate_speech_stream(self, text: str, voice: str = 'af_heart', speed: float = 1.0) -> Generator[Tuple[int, np.ndarray], None, None]:
+        """Generate speech stream using Kokoro TTS engine
+        
+        Args:
+            text (str): Input text to synthesize
+            voice (str): Voice ID to use
+            speed (float): Speech speed multiplier
+            
+        Yields:
+            tuple: (sample_rate, audio_data) pairs for each segment
+        """
+        logger.info(f"Generating speech stream with Kokoro for text length: {len(text)}")
+        
+        # Generate speech stream
+        generator = self.pipeline(text, voice=voice, speed=speed)
+        for _, _, audio in generator:
+            yield 24000, audio
+
+
+class KokoroSpaceTTSEngine(TTSEngineBase):
+    """Kokoro Space TTS engine implementation
+    
+    This engine uses the Kokoro FastAPI server for TTS generation.
+    """
+    
+    def __init__(self, lang_code: str = 'z'):
+        super().__init__(lang_code)
+        try:
+            from gradio_client import Client
+            self.client = Client("Remsky/Kokoro-TTS-Zero")
+            logger.info("Kokoro Space TTS engine successfully initialized")
+        except Exception as e:
+            logger.error(f"Failed to initialize Kokoro Space client: {str(e)}")
+            logger.error(f"Error type: {type(e).__name__}")
+            raise
+    
+    def generate_speech(self, text: str, voice: str = 'af_heart', speed: float = 1.0) -> str:
+        """Generate speech using Kokoro Space TTS engine
+        
+        Args:
+            text (str): Input text to synthesize
+            voice (str): Voice ID to use (e.g., 'af_heart', 'af_bella', etc.)
+            speed (float): Speech speed multiplier (0.5 to 2.0)
+            
+        Returns:
+            str: Path to the generated audio file
+        """
+        logger.info(f"Generating speech with Kokoro Space for text length: {len(text)}")
+        logger.info(f"Text to generate speech on is: {text[:50]}..." if len(text) > 50 else f"Text to generate speech on is: {text}")
+        
+        # Generate unique output path
+        output_path = self._generate_output_path()
+        
+        try:
+            # Use af_nova as the default voice for Kokoro Space
+            voice_to_use = 'af_nova' if voice == 'af_heart' else voice
+            
+            # Generate speech
+            result = self.client.predict(
+                text=text,
+                voice_names=voice_to_use,
+                speed=speed,
+                api_name="/generate_speech_from_ui"
+            )
+            logger.info(f"Received audio from Kokoro FastAPI server: {result}")
+            
+            # TODO: Process the result and save to output_path
+            # For now, we'll return the result path directly if it's a string
+            if isinstance(result, str) and os.path.exists(result):
+                return result
+            else:
+                logger.warning("Unexpected result from Kokoro Space, falling back to dummy audio")
+                return DummyTTSEngine().generate_speech(text, voice, speed)
+            
+        except Exception as e:
+            logger.error(f"Failed to generate speech from Kokoro FastAPI server: {str(e)}")
+            logger.error(f"Error type: {type(e).__name__}")
+            logger.info("Falling back to dummy audio generation")
+            return DummyTTSEngine().generate_speech(text, voice, speed)
+
+
+class DiaTTSEngine(TTSEngineBase):
+    """Dia TTS engine implementation
+    
+    This engine uses the Dia model for TTS generation.
+    """
+    
+    def __init__(self, lang_code: str = 'z'):
+        super().__init__(lang_code)
+        # Dia doesn't need initialization here, it will be lazy-loaded when needed
+        logger.info("Dia TTS engine initialized (lazy loading)")
+    
+    def generate_speech(self, text: str, voice: str = 'af_heart', speed: float = 1.0) -> str:
+        """Generate speech using Dia TTS engine
+        
+        Args:
+            text (str): Input text to synthesize
+            voice (str): Voice ID (not used in Dia)
+            speed (float): Speech speed multiplier (not used in Dia)
+            
+        Returns:
+            str: Path to the generated audio file
+        """
+        logger.info(f"Generating speech with Dia for text length: {len(text)}")
+        
+        try:
+            # Import here to avoid circular imports
+            from utils.tts_dia import generate_speech as dia_generate_speech
+            logger.info("Successfully imported Dia speech generation function")
+            
+            # Call Dia's generate_speech function
+            # Note: Dia's function expects a language parameter, not voice or speed
+            output_path = dia_generate_speech(text, language=self.lang_code)
+            logger.info(f"Generated audio with Dia: {output_path}")
+            return output_path
+            
+        except ImportError as import_err:
+            logger.error(f"Dia TTS generation failed due to import error: {str(import_err)}")
+            logger.error("Falling back to dummy audio generation")
+            return DummyTTSEngine().generate_speech(text, voice, speed)
+            
+        except Exception as dia_error:
+            logger.error(f"Dia TTS generation failed: {str(dia_error)}", exc_info=True)
+            logger.error(f"Error type: {type(dia_error).__name__}")
+            logger.error("Falling back to dummy audio generation")
+            return DummyTTSEngine().generate_speech(text, voice, speed)
+    
+    def generate_speech_stream(self, text: str, voice: str = 'af_heart', speed: float = 1.0) -> Generator[Tuple[int, np.ndarray], None, None]:
+        """Generate speech stream using Dia TTS engine
+        
+        Args:
+            text (str): Input text to synthesize
+            voice (str): Voice ID (not used in Dia)
+            speed (float): Speech speed multiplier (not used in Dia)
+            
+        Yields:
+            tuple: (sample_rate, audio_data) pairs for each segment
+        """
+        logger.info(f"Generating speech stream with Dia for text length: {len(text)}")
+        
+        try:
+            # Import required modules
+            import torch
+            from utils.tts_dia import _get_model, DEFAULT_SAMPLE_RATE
+            
+            # Get the Dia model
+            model = _get_model()
+            
+            # Generate audio
+            with torch.inference_mode():
+                output_audio_np = model.generate(
+                    text,
+                    max_tokens=None,
+                    cfg_scale=3.0,
+                    temperature=1.3,
+                    top_p=0.95,
+                    cfg_filter_top_k=35,
+                    use_torch_compile=False,
+                    verbose=False
+                )
+            
+            if output_audio_np is not None:
+                logger.info(f"Successfully generated audio with Dia (length: {len(output_audio_np)})")
+                yield DEFAULT_SAMPLE_RATE, output_audio_np
+            else:
+                logger.warning("Dia model returned None for audio output")
+                logger.warning("Falling back to dummy audio stream")
+                yield from DummyTTSEngine().generate_speech_stream(text, voice, speed)
+                
+        except ImportError as import_err:
+            logger.error(f"Dia TTS streaming failed due to import error: {str(import_err)}")
+            logger.error("Falling back to dummy audio stream")
+            yield from DummyTTSEngine().generate_speech_stream(text, voice, speed)
+            
+        except Exception as dia_error:
+            logger.error(f"Dia TTS streaming failed: {str(dia_error)}", exc_info=True)
+            logger.error(f"Error type: {type(dia_error).__name__}")
+            logger.error("Falling back to dummy audio stream")
+            yield from DummyTTSEngine().generate_speech_stream(text, voice, speed)
+
+
+def get_available_engines() -> List[str]:
+    """Get a list of available TTS engines
+    
+    Returns:
+        List[str]: List of available engine names
+    """
+    available = []
+    
+    if KOKORO_AVAILABLE:
+        available.append('kokoro')
+    
+    if KOKORO_SPACE_AVAILABLE:
+        available.append('kokoro_space')
+    
+    if DIA_AVAILABLE:
+        available.append('dia')
+    
+    # Dummy is always available
+    available.append('dummy')
+    
+    return available
+
+
+def create_engine(engine_type: str, lang_code: str = 'z') -> TTSEngineBase:
+    """Create a specific TTS engine
+    
+    Args:
+        engine_type (str): Type of engine to create ('kokoro', 'kokoro_space', 'dia', 'dummy')
+        lang_code (str): Language code for the engine
+        
+    Returns:
+        TTSEngineBase: An instance of the requested TTS engine
+        
+    Raises:
+        ValueError: If the requested engine type is not supported
+    """
+    if engine_type == 'kokoro':
+        if not KOKORO_AVAILABLE:
+            raise ValueError("Kokoro TTS engine is not available")
+        return KokoroTTSEngine(lang_code)
+    
+    elif engine_type == 'kokoro_space':
+        if not KOKORO_SPACE_AVAILABLE:
+            raise ValueError("Kokoro Space TTS engine is not available")
+        return KokoroSpaceTTSEngine(lang_code)
+    
+    elif engine_type == 'dia':
+        if not DIA_AVAILABLE:
+            raise ValueError("Dia TTS engine is not available")
+        return DiaTTSEngine(lang_code)
+    
+    elif engine_type == 'dummy':
+        return DummyTTSEngine(lang_code)
+    
+    else:
+        raise ValueError(f"Unsupported TTS engine type: {engine_type}")

utils/tts_factory.py

@@ -0,0 +1,118 @@
+import logging
+from typing import Optional, List
+
+# Configure logging
+logger = logging.getLogger(__name__)
+
+# Import the base class
+from utils.tts_base import TTSEngineBase, DummyTTSEngine
+
+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}")
+                return create_engine(engine_type, lang_code)
+            else:
+                logger.warning(f"Requested engine '{engine_type}' is not available")
+        
+        # Try to create the best available engine
+        # Priority: kokoro > kokoro_space > dia > dummy
+        for engine in ['kokoro', 'kokoro_space', 'dia']:
+            if engine in available_engines:
+                logger.info(f"Creating best available engine: {engine}")
+                return create_engine(engine, lang_code)
+        
+        # Fall back to dummy engine
+        logger.warning("No TTS engines available, falling back to dummy engine")
+        return DummyTTSEngine(lang_code)
+
+
+# Backward compatibility function
+def get_tts_engine(lang_code: str = 'a') -> TTSEngineBase:
+    """Get or create TTS engine instance (backward compatibility function)
+    
+    Args:
+        lang_code (str): Language code for the pipeline
+        
+    Returns:
+        TTSEngineBase: Initialized TTS engine instance
+    """
+    logger.info(f"Requesting TTS engine with language code: {lang_code}")
+    try:
+        import streamlit as st
+        logger.info("Streamlit detected, using cached TTS engine")
+        @st.cache_resource
+        def _get_engine():
+            logger.info("Creating cached TTS engine instance")
+            engine = TTSFactory.create_engine(lang_code=lang_code)
+            logger.info(f"Cached TTS engine created with type: {engine.__class__.__name__}")
+            return engine
+        
+        engine = _get_engine()
+        logger.info(f"Retrieved TTS engine from cache with type: {engine.__class__.__name__}")
+        return engine
+    except ImportError:
+        logger.info("Streamlit not available, creating direct TTS engine instance")
+        engine = TTSFactory.create_engine(lang_code=lang_code)
+        logger.info(f"Direct TTS engine created with type: {engine.__class__.__name__}")
+        return engine
+
+
+# Backward compatibility function
+def generate_speech(text: str, voice: str = 'af_heart', speed: float = 1.0) -> str:
+    """Public interface for TTS generation (backward compatibility function)
+    
+    Args:
+        text (str): Input text to synthesize
+        voice (str): Voice ID to use
+        speed (float): Speech speed multiplier
+        
+    Returns:
+        str: Path to generated audio file
+    """
+    logger.info(f"Public generate_speech called with text length: {len(text)}, voice: {voice}, speed: {speed}")
+    try:
+        # Get the TTS engine
+        logger.info("Getting TTS engine instance")
+        engine = get_tts_engine()
+        logger.info(f"Using TTS engine type: {engine.__class__.__name__}")
+        
+        # Generate speech
+        logger.info("Calling engine.generate_speech")
+        output_path = engine.generate_speech(text, voice, speed)
+        logger.info(f"Speech generation complete, output path: {output_path}")
+        return output_path
+    except Exception as e:
+        logger.error(f"Error in public generate_speech function: {str(e)}", exc_info=True)
+        logger.error(f"Error type: {type(e).__name__}")
+        if hasattr(e, '__traceback__'):
+            tb = e.__traceback__
+            while tb.tb_next:
+                tb = tb.tb_next
+            logger.error(f"Error occurred in file: {tb.tb_frame.f_code.co_filename}, line {tb.tb_lineno}")
+        raise