Update app.py

app.py

@@ -306,7 +306,7 @@ def load_whisperx_models():
     if whisperx_model is None:
         log("Loading WhisperX models for English-only processing...")
         try:
-            # Try loading with base.en first
+            # Load WhisperX model with English-only configuration
             whisperx_model = whisperx.load_model("base.en", device="cpu", compute_type="float32", language="en")
             log("WhisperX base.en model loaded successfully")
             
@@ -314,44 +314,17 @@ def load_whisperx_models():
             whisperx_align_model, whisperx_metadata = whisperx.load_align_model(language_code="en", device="cpu")
             log("WhisperX English alignment model loaded successfully")
             
-        except ImportError as ie:
-            log(f"Import error loading WhisperX models: {ie}")
-            # Try without ctranslate2 by using int8 compute type
-            try:
-                log("Trying fallback with int8 compute type...")
-                whisperx_model = whisperx.load_model("base.en", device="cpu", compute_type="int8", language="en")
-                whisperx_align_model, whisperx_metadata = whisperx.load_align_model(language_code="en", device="cpu")
-                log("WhisperX models loaded with int8 compute type")
-            except Exception as fallback_error:
-                log(f"Int8 fallback also failed: {fallback_error}")
-                # Last resort: try tiny model with default compute
-                try:
-                    log("Trying final fallback with tiny.en model and default compute...")
-                    whisperx_model = whisperx.load_model("tiny.en", device="cpu", language="en")
-                    whisperx_align_model, whisperx_metadata = whisperx.load_align_model(language_code="en", device="cpu")
-                    log("WhisperX models loaded with tiny.en and default compute")
-                except Exception as final_error:
-                    log(f"All WhisperX loading attempts failed: {final_error}")
-                    raise RuntimeError("Unable to load WhisperX models. Please check environment setup.")
         except Exception as e:
             log(f"Error loading WhisperX models: {e}")
             # Fallback: try with smaller English-only model
             try:
                 log("Trying fallback with tiny.en model...")
-                whisperx_model = whisperx.load_model("tiny.en", device="cpu", compute_type="int8", language="en")
+                whisperx_model = whisperx.load_model("tiny.en", device="cpu", compute_type="float32", language="en")
                 whisperx_align_model, whisperx_metadata = whisperx.load_align_model(language_code="en", device="cpu")
                 log("WhisperX models loaded with fallback (tiny.en model)")
             except Exception as fallback_error:
                 log(f"Fallback also failed: {fallback_error}")
-                # Final attempt without compute_type specification
-                try:
-                    log("Final attempt with default settings...")
-                    whisperx_model = whisperx.load_model("tiny.en", device="cpu", language="en")
-                    whisperx_align_model, whisperx_metadata = whisperx.load_align_model(language_code="en", device="cpu")
-                    log("WhisperX models loaded with default settings")
-                except Exception as final_error:
-                    log(f"All attempts failed: {final_error}")
-                    raise RuntimeError("Unable to load WhisperX models in this environment")
+                raise
 
 def convert_webm_to_wav(bts):
     p = subprocess.run(["ffmpeg", "-i", "pipe:0", "-f", "wav", "-ar", "16000", "-ac", "1", "pipe:1"],
@@ -947,37 +920,103 @@ def trim_audio_segment_by_phoneme_position(audio_segment: torch.Tensor,
     return trimmed_segment
 
 def get_expected_phonemes(words: List[str]) -> List[str]:
-    """Get expected phonemes using espeak phonemizer"""
+    """Get expected phonemes using CMUdict instead of espeak phonemizer"""
     cache_key = tuple(words)
     if cache_key in phoneme_cache:
-        log(f"📚 Using cached phonemes for: {words}")
-        cached_result = phoneme_cache[cache_key]
-        log(f"   Cached phonemes: {list(zip(words, cached_result))}")
-        return cached_result
+        return phoneme_cache[cache_key]
+    
+    log(f"Getting expected phonemes from CMUdict for: {words}")
+    
+    # ARPABET to IPA conversion mapping (same as in build_phoneme_reverse_lookup)
+    arpabet_to_ipa = {
+        'AA': 'ɑ',   'AE': 'æ',   'AH': 'ə',   'AO': 'ɔ',   'AW': 'aʊ',
+        'AY': 'aɪ',  'B': 'b',    'CH': 'tʃ',  'D': 'd',    'DH': 'ð',
+        'EH': 'ɛ',   'ER': 'ɝ',   'EY': 'eɪ',  'F': 'f',    'G': 'ɡ',
+        'HH': 'h',   'IH': 'ɪ',   'IY': 'i',   'JH': 'dʒ',  'K': 'k',
+        'L': 'l',    'M': 'm',    'N': 'n',    'NG': 'ŋ',   'OW': 'oʊ',
+        'OY': 'ɔɪ',  'P': 'p',    'R': 'r',    'S': 's',    'SH': 'ʃ',
+        'T': 't',    'TH': 'θ',   'UH': 'ʊ',   'UW': 'u',   'V': 'v',
+        'W': 'w',    'Y': 'j',    'Z': 'z',    'ZH': 'ʒ',   'DX': 'ɾ'
+    }
     
-    log(f"🔤 Getting expected phonemes using phonemizer for: {words}")
+    # Load CMUdict on first use
+    cmudict_lookup = {}
+    if not hasattr(get_expected_phonemes, '_cmudict_loaded'):
+        log("Loading CMUdict for expected phonemes...")
+        try:
+            cmudict_path = "/tmp/cmudict.dict"
+            if os.path.exists(cmudict_path):
+                with open(cmudict_path, 'r', encoding='latin-1') as f:
+                    for line in f:
+                        line = line.strip()
+                        if not line or line.startswith(';;;'):
+                            continue
+                        
+                        parts = line.split()
+                        if len(parts) < 2:
+                            continue
+                        
+                        word = parts[0].lower()
+                        # Remove variant indicators like (2), (3)
+                        if '(' in word:
+                            word = word.split('(')[0]
+                        
+                        # Store ARPABET phones (we'll convert to IPA as needed)
+                        arpabet_phones = parts[1:]
+                        cmudict_lookup[word] = arpabet_phones
+                
+                get_expected_phonemes._cmudict_loaded = True
+                get_expected_phonemes._cmudict_lookup = cmudict_lookup
+                log(f"Loaded {len(cmudict_lookup)} words from CMUdict")
+            else:
+                log("⚠️  CMUdict not found, falling back to phonemizer")
+                # Fallback to original phonemizer
+                phonemes = phonemize(words, language='en-us', backend='espeak', strip=True)
+                phoneme_cache[cache_key] = phonemes
+                return phonemes
+        except Exception as e:
+            log(f"❌ Error loading CMUdict: {e}, falling back to phonemizer")
+            phonemes = phonemize(words, language='en-us', backend='espeak', strip=True)
+            phoneme_cache[cache_key] = phonemes
+            return phonemes
+    else:
+        cmudict_lookup = get_expected_phonemes._cmudict_lookup
     
-    try:
-        # Use espeak phonemizer to get IPA phonemes
-        phonemes = phonemize(words, language='en-us', backend='espeak', strip=True)
-        
-        # Cache the results
-        phoneme_cache[cache_key] = phonemes
-        
-        # Log the phoneme results
-        log(f"✅ Phonemizer results:")
-        for word, phoneme in zip(words, phonemes):
-            log(f"   '{word}' → '{phoneme}'")
+    # Convert words to phonemes using CMUdict
+    results = []
+    for word in words:
+        word_lower = word.lower()
         
-        return phonemes
-        
-    except Exception as e:
-        log(f"❌ Error in phonemizer: {e}")
-        log(f"   Returning empty phonemes for all words")
-        # Return empty strings as fallback
-        empty_results = [""] * len(words)
-        phoneme_cache[cache_key] = empty_results
-        return empty_results
+        if word_lower in cmudict_lookup:
+            # Convert ARPABET to IPA
+            arpabet_phones = cmudict_lookup[word_lower]
+            ipa_phones = []
+            
+            for phone in arpabet_phones:
+                # Remove stress markers (0,1,2)
+                clean_phone = ''.join(c for c in phone if not c.isdigit())
+                if clean_phone in arpabet_to_ipa:
+                    ipa_phones.append(arpabet_to_ipa[clean_phone])
+                else:
+                    log(f"⚠️  Unknown ARPABET phone '{clean_phone}' in word '{word}'")
+            
+            ipa_string = ''.join(ipa_phones)
+            results.append(ipa_string)
+            log(f"CMUdict: '{word}' → ARPABET {arpabet_phones} → IPA '{ipa_string}'")
+            
+        else:
+            # Fallback to phonemizer for out-of-vocabulary words
+            log(f"⚠️  '{word}' not in CMUdict, using phonemizer fallback")
+            try:
+                fallback_phoneme = phonemize([word], language='en-us', backend='espeak', strip=True)[0]
+                results.append(fallback_phoneme)
+            except Exception as e:
+                log(f"❌ Phonemizer fallback failed for '{word}': {e}")
+                results.append("")  # Empty string as last resort
+    
+    phoneme_cache[cache_key] = results
+    log(f"Final expected phonemes: {list(zip(words, results))}")
+    return results
 
 async def generate_tts_audio(word: str) -> str:
     """Generate TTS audio for a word with silence padding"""