update tts

app.py

@@ -21,7 +21,7 @@ import time
 import subprocess
 from utils.stt import transcribe_audio
 from utils.translation import translate_text
-from utils.tts_dummy import generate_speech
+from utils.tts import generate_speech
 
 # Hugging Face Spaces Setup Automation
 def setup_huggingface_space():
@@ -112,7 +112,7 @@ def handle_file_processing(upload_path):
         logger.info("Beginning TTS generation")
         status_text.markdown("🎵 **Generating Chinese Speech...**")
         with st.spinner("Initializing TTS engine..."):
-            output_path = generate_speech(chinese_text, language="zh")
+            output_path = generate_speech(chinese_text, voice="zf_xiaobei")
         progress_bar.progress(100)
         logger.info(f"TTS completed. Output file: {output_path}")
         

pyproject.toml

@@ -14,12 +14,12 @@ python = "^3.9"
 
 # Core application dependencies
 streamlit = ">=1.31,<2.0"
-pydub = ">=0.25"
 python-dotenv = ">=1.0"
 nltk = ">=3.8"
 librosa = ">=0.10"
 soundfile = ">=0.12"
 ffmpeg-python = ">=0.2"
+kokoro = ">=0.7.9"
 
 # Machine learning dependencies
 #torch = [

utils/tts.py

@@ -1,78 +1,50 @@
 import os
-import torch
-import time
 import logging
-from pydub import AudioSegment
-from phonemizer.backend.espeak.wrapper import EspeakWrapper
-from models import build_model
+import time
+import soundfile as sf
+from kokoro import KPipeline
 
 logger = logging.getLogger(__name__)
 
-# Hugging Face Spaces setup
-MODEL_DIR = "./kokoro"
-os.makedirs(MODEL_DIR, exist_ok=True)
-
-# Configure espeak-ng for Hugging Face environment
-EspeakWrapper.set_library('/usr/lib/x86_64-linux-gnu/libespeak-ng.so.1')
-
 class TTSEngine:
-    def __init__(self):
+    def __init__(self, lang_code='z'):
+        """Initialize TTS Engine with Kokoro
+        
+        Args:
+            lang_code (str): Language code ('a' for US English, 'b' for British English,
+                           'j' for Japanese, 'z' for Mandarin Chinese)
+        """
         logger.info("Initializing TTS Engine")
-        self.device = "cuda" if torch.cuda.is_available() else "cpu"
-        logger.info(f"Using device: {self.device}")
-        self._verify_model_files()
-        logger.info("Loading Kokoro model")
-        self.model = build_model(f"{MODEL_DIR}/kokoro-v0_19.pth", self.device)
-        logger.info("Loading voice model")
-        self.voice = torch.load(f"{MODEL_DIR}/voices/af_bella.pt", 
-                              map_location=self.device)
+        self.pipeline = KPipeline(lang_code=lang_code)
         logger.info("TTS engine initialized")
 
-    def _verify_model_files(self):
-        """Ensure required model files exist"""
-        required_files = [
-            f"{MODEL_DIR}/kokoro-v0_19.pth",
-            f"{MODEL_DIR}/voices/af_bella.pt"
-        ]
+    def generate_speech(self, text: str, voice: str = 'af_heart', speed: float = 1.0) -> str:
+        """Generate speech from text using Kokoro
         
-        missing = [f for f in required_files if not os.path.exists(f)]
-        if missing:
-            logger.error(f"Missing model files: {missing}")
-            raise FileNotFoundError(
-                f"Missing model files: {missing}\n"
-                "Add this to your Hugging Face Space settings:\n"
-                "App setup -> Clone Kokoro repository: "
-                "git clone https://huggingface.co/hexgrad/Kokoro-82M ./kokoro"
-            )
-
-    def generate_speech(self, text: str, language: str = "zh") -> str:
-        """Generate speech from Chinese text"""
+        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 for text length: {len(text)}")
         
         try:
-            from kokoro import generate_full
-            
-            if len(text) > 500:
-                logger.warning(f"Truncating long text ({len(text)} characters)")
-                text = text[:495] + "[TRUNCATED]"
-            
-            logger.info("Starting audio generation")
-            audio, _ = generate_full(
-                self.model,
-                text,
-                self.voice,
-                lang='en-us',
-                max_len=200 if self.device == "cpu" else 500
-            )
+            # 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"
-            logger.info(f"Saving audio to {output_path}")
-            AudioSegment(
-                audio.numpy().tobytes(),
-                frame_rate=24000,
-                sample_width=2,
-                channels=1
-            ).export(output_path, format="wav")
+            
+            # Get the first generated segment
+            # We only take the first segment since the original code handled single segments
+            generator = self.pipeline(text, voice=voice, speed=speed)
+            for _, _, audio in generator:
+                logger.info(f"Saving audio to {output_path}")
+                sf.write(output_path, audio, 24000)
+                break
             
             logger.info(f"Audio generation complete: {output_path}")
             return output_path
@@ -81,11 +53,55 @@ class TTSEngine:
             logger.error(f"TTS generation failed: {str(e)}", exc_info=True)
             raise
 
-# Initialize TTS engine once
-@st.cache_resource
-def get_tts_engine():
-    return TTSEngine()
+    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)
+            
+        Yields:
+            tuple: (sample_rate, audio_data) pairs for each segment
+        """
+        try:
+            generator = self.pipeline(text, voice=voice, speed=speed)
+            for _, _, audio in generator:
+                yield 24000, audio
+                
+        except Exception as e:
+            logger.error(f"TTS streaming failed: {str(e)}", exc_info=True)
+            raise
+
+# 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
+    """
+    try:
+        import streamlit as st
+        @st.cache_resource
+        def _get_engine():
+            return TTSEngine(lang_code)
+        return _get_engine()
+    except ImportError:
+        return TTSEngine(lang_code)
 
-def generate_speech(text: str, language: str = "zh") -> str:
-    """Public interface for TTS generation"""
-    return get_tts_engine().generate_speech(text, language)
+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
+    """
+    engine = get_tts_engine()
+    return engine.generate_speech(text, voice, speed)