Update app.py

app.py

@@ -1,8 +1,8 @@
 import sys
 import os
 import types
-import subprocess
 import logging
+import re
 
 # Shim for removed audioop module (Python 3.13+)
 if 'audioop' not in sys.modules:
@@ -15,77 +15,88 @@ import matplotlib
 matplotlib.use('Agg')
 
 logger = logging.getLogger(__name__)
+logging.basicConfig(level=logging.INFO)
 
 # ---------------------------------------------------------------------------
-# Monkey-patch TRIBE's whisperx subprocess call to use whisperx as a Python
-# library instead. TRIBE internally calls `uvx whisperx ...` via subprocess,
-# which fails in HuggingFace Spaces. This patch replaces that with a direct
-# Python call to the whisperx library.
+# Monkey-patch TRIBE's ExtractWordsFromAudio to build word-level events
+# WITHOUT calling whisperx (which requires CUDA libs unavailable on CPU).
+#
+# Instead, we use a simple heuristic: split the transcript text into words
+# and distribute them evenly across the audio duration. This gives TRIBE
+# enough word-level signal for its text encoder without needing ASR.
 # ---------------------------------------------------------------------------
 def _patched_get_transcript_from_audio(wav_filename, language="english"):
-    """Replacement for ExtractWordsFromAudio._get_transcript_from_audio
-    that uses whisperx as a Python library instead of a subprocess."""
-    import whisperx
-    import torch
-    from pathlib import Path
-
-    language_codes = dict(
-        english="en", french="fr", spanish="es", dutch="nl", chinese="zh"
-    )
-    if language not in language_codes:
-        raise ValueError(f"Language {language} not supported")
-
-    device = "cuda" if torch.cuda.is_available() else "cpu"
-    compute_type = "float16" if device == "cuda" else "int8"
-    lang_code = language_codes[language]
+    """CPU-safe replacement that creates word events from audio duration.
 
-    logger.info("Loading whisperx model (patched)...")
-    wx_model = whisperx.load_model("base", device, compute_type=compute_type, language=lang_code)
-
-    logger.info(f"Transcribing {wav_filename}...")
-    audio = whisperx.load_audio(str(wav_filename))
-    result = wx_model.transcribe(audio, batch_size=4)
+    When the audio was generated from known text (gTTS), the global
+    CURRENT_SCRIPT_TEXT will contain that text.  Otherwise we create
+    a minimal placeholder so TRIBE's pipeline doesn't crash.
+    """
+    import pandas as pd
+    import soundfile as sf
+    from pathlib import Path
 
-    logger.info("Aligning words...")
-    model_a, metadata = whisperx.load_align_model(
-        language_code=lang_code, device=device
-    )
-    result = whisperx.align(
-        result["segments"], model_a, metadata, audio, device,
-        return_char_alignments=False
-    )
+    wav_filename = Path(wav_filename)
 
-    import pandas as pd
+    # Get audio duration
+    try:
+        info = sf.info(str(wav_filename))
+        duration = info.duration
+    except Exception:
+        duration = 30.0  # fallback
+
+    # Use the known script text if available, otherwise a placeholder
+    text = _CURRENT_SCRIPT_TEXT or "audio content placeholder"
+
+    # Tokenize into words
+    raw_words = text.split()
+    if not raw_words:
+        return pd.DataFrame(columns=["text", "start", "duration", "sequence_id", "sentence"])
+
+    # Split into sentences (rough: split on . ! ?)
+    sentences = re.split(r'(?<=[.!?])\s+', text)
+    sentences = [s.strip() for s in sentences if s.strip()]
+    if not sentences:
+        sentences = [text]
+
+    # Distribute words evenly across the audio duration
+    word_duration = duration / len(raw_words)
     words = []
-    for i, segment in enumerate(result.get("segments", [])):
-        sentence = segment.get("text", "").replace('"', "")
-        for word in segment.get("words", []):
-            if "start" not in word:
-                continue
+    word_idx = 0
+    for sent_idx, sentence in enumerate(sentences):
+        sent_words = sentence.split()
+        for w in sent_words:
+            if word_idx >= len(raw_words):
+                break
             words.append({
-                "text": word["word"].replace('"', ""),
-                "start": word["start"],
-                "duration": word["end"] - word["start"],
-                "sequence_id": i,
-                "sentence": sentence,
+                "text": w.replace('"', ''),
+                "start": word_idx * word_duration,
+                "duration": word_duration * 0.9,
+                "sequence_id": sent_idx,
+                "sentence": sentence.replace('"', ''),
             })
+            word_idx += 1
 
     return pd.DataFrame(words)
 
 
-def apply_whisperx_patch():
-    """Apply the monkey-patch to TRIBE's ExtractWordsFromAudio class."""
+# Global to pass text from the analyze function to the monkey-patch
+_CURRENT_SCRIPT_TEXT = None
+
+
+def apply_patches():
+    """Patch TRIBE's ExtractWordsFromAudio to avoid whisperx/CUDA dependency."""
     try:
         from tribev2.eventstransforms import ExtractWordsFromAudio
         ExtractWordsFromAudio._get_transcript_from_audio = staticmethod(
             _patched_get_transcript_from_audio
         )
-        logger.info("Successfully patched ExtractWordsFromAudio to use whisperx Python library")
+        logger.info("Patched ExtractWordsFromAudio (CPU-safe, no whisperx)")
     except Exception as e:
         logger.warning(f"Could not patch ExtractWordsFromAudio: {e}")
 
-# Apply the patch before loading the model
-apply_whisperx_patch()
+# Apply patches at import time
+apply_patches()
 
 # ---------------------------------------------------------------------------
 # Model loading
@@ -97,8 +108,7 @@ def load_model():
     if model is not None:
         return "✅ Already loaded!"
     try:
-        # Re-apply patch in case import order matters
-        apply_whisperx_patch()
+        apply_patches()  # re-apply in case import order matters
         from tribev2 import TribeModel
         model = TribeModel.from_pretrained("facebook/tribev2", cache_folder="/tmp/tribe_cache")
         return "✅ Model loaded!"
@@ -197,6 +207,8 @@ def generate_suggestions(scores, overall):
 # Main analysis function
 # ---------------------------------------------------------------------------
 def analyze(input_mode, script_text, audio_file, progress=gr.Progress()):
+    global _CURRENT_SCRIPT_TEXT
+
     if input_mode == "Text" and (not script_text or not script_text.strip()):
         return None, None, "⚠️ Please paste your script text first.", None
     if input_mode == "Audio" and audio_file is None:
@@ -212,9 +224,6 @@ def analyze(input_mode, script_text, audio_file, progress=gr.Progress()):
         if input_mode == "Text":
             progress(0.2, desc="Converting text to speech...")
 
-            # Convert text → audio with gTTS, then feed as audio_path.
-            # This avoids TRIBE's internal TextToEvents path which also
-            # calls whisperx via subprocess after doing the same gTTS step.
             from gtts import gTTS
             from langdetect import detect
 
@@ -224,6 +233,10 @@ def analyze(input_mode, script_text, audio_file, progress=gr.Progress()):
             tts = gTTS(text=text, lang=lang)
             tts.save(audio_path)
 
+            # Store text so the monkey-patched transcriber can use it
+            # instead of running ASR on the audio we just synthesised.
+            _CURRENT_SCRIPT_TEXT = text
+
             progress(0.4, desc="Running TRIBE v2 on generated audio...")
             df = model.get_events_dataframe(audio_path=audio_path)
 
@@ -234,6 +247,9 @@ def analyze(input_mode, script_text, audio_file, progress=gr.Progress()):
             audio_path = f"/tmp/input_audio{ext}"
             shutil.copy(audio_file, audio_path)
 
+            # No known text for uploaded audio
+            _CURRENT_SCRIPT_TEXT = None
+
             progress(0.4, desc="Running TRIBE v2 on audio...")
             df = model.get_events_dataframe(audio_path=audio_path)
 
@@ -257,6 +273,8 @@ def analyze(input_mode, script_text, audio_file, progress=gr.Progress()):
         full_error = traceback.format_exc()
         print(full_error)
         return None, None, f"❌ Error:\n{str(e)}\n\nFull traceback:\n{full_error}", None
+    finally:
+        _CURRENT_SCRIPT_TEXT = None
 
 # ---------------------------------------------------------------------------
 # Gradio UI