Update app.py

app.py

@@ -2,7 +2,9 @@ import sys
 import os
 import types
 import subprocess
+import logging
 
+# Shim for removed audioop module (Python 3.13+)
 if 'audioop' not in sys.modules:
     sys.modules['audioop'] = types.ModuleType('audioop')
 
@@ -12,6 +14,82 @@ import matplotlib.pyplot as plt
 import matplotlib
 matplotlib.use('Agg')
 
+logger = logging.getLogger(__name__)
+
+# ---------------------------------------------------------------------------
+# 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.
+# ---------------------------------------------------------------------------
+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]
+
+    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)
+
+    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
+    )
+
+    import pandas as pd
+    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
+            words.append({
+                "text": word["word"].replace('"', ""),
+                "start": word["start"],
+                "duration": word["end"] - word["start"],
+                "sequence_id": i,
+                "sentence": sentence,
+            })
+
+    return pd.DataFrame(words)
+
+
+def apply_whisperx_patch():
+    """Apply the monkey-patch to TRIBE's ExtractWordsFromAudio class."""
+    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")
+    except Exception as e:
+        logger.warning(f"Could not patch ExtractWordsFromAudio: {e}")
+
+# Apply the patch before loading the model
+apply_whisperx_patch()
+
+# ---------------------------------------------------------------------------
+# Model loading
+# ---------------------------------------------------------------------------
 model = None
 
 def load_model():
@@ -19,12 +97,19 @@ def load_model():
     if model is not None:
         return "✅ Already loaded!"
     try:
+        # Re-apply patch in case import order matters
+        apply_whisperx_patch()
         from tribev2 import TribeModel
         model = TribeModel.from_pretrained("facebook/tribev2", cache_folder="/tmp/tribe_cache")
         return "✅ Model loaded!"
     except Exception as e:
+        import traceback
+        traceback.print_exc()
         return f"❌ Error loading model: {str(e)}"
 
+# ---------------------------------------------------------------------------
+# Brain region definitions (approximate vertex ranges on fsaverage5)
+# ---------------------------------------------------------------------------
 REGIONS = [
     ("Visual cortex",           0.00, 0.15, "#378ADD"),
     ("Auditory cortex",         0.15, 0.30, "#D85A30"),
@@ -108,43 +193,9 @@ def generate_suggestions(scores, overall):
     status = "🟢 Strong" if overall >= 75 else "🟡 Good, needs polish" if overall >= 55 else "🔴 Needs work"
     return f"**Overall: {overall}/100 — {status}**\n\n" + "\n".join(tips)
 
-def transcribe_audio_to_text(audio_path):
-    """Use whisperx directly via Python to transcribe audio to a timed text file."""
-    import whisperx
-    import json
-
-    hf_token = os.environ.get("HF_TOKEN", "")
-    device = "cpu"
-
-    print("Loading whisper model...")
-    wx_model = whisperx.load_model("base", device, compute_type="int8")
-
-    print("Transcribing audio...")
-    audio = whisperx.load_audio(audio_path)
-    result = wx_model.transcribe(audio, batch_size=4)
-
-    print("Aligning words...")
-    model_a, metadata = whisperx.load_align_model(
-        language_code=result["language"], device=device
-    )
-    result = whisperx.align(
-        result["segments"], model_a, metadata, audio, device,
-        return_char_alignments=False
-    )
-
-    # Build a simple timed transcript text file
-    lines = []
-    for seg in result["segments"]:
-        lines.append(seg["text"].strip())
-    transcript = " ".join(lines)
-
-    text_path = "/tmp/transcript.txt"
-    with open(text_path, "w") as f:
-        f.write(transcript)
-
-    print(f"Transcript: {transcript[:200]}...")
-    return text_path
-
+# ---------------------------------------------------------------------------
+# Main analysis function
+# ---------------------------------------------------------------------------
 def analyze(input_mode, script_text, audio_file, progress=gr.Progress()):
     if input_mode == "Text" and (not script_text or not script_text.strip()):
         return None, None, "⚠️ Please paste your script text first.", None
@@ -159,12 +210,22 @@ def analyze(input_mode, script_text, audio_file, progress=gr.Progress()):
 
     try:
         if input_mode == "Text":
-            progress(0.2, desc="Preparing text input...")
-            text_path = "/tmp/script.txt"
-            with open(text_path, "w") as f:
-                f.write(script_text.strip())
-            progress(0.4, desc="Running TRIBE v2 on text...")
-            df = model.get_events_dataframe(text_path=text_path)
+            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
+
+            text = script_text.strip()
+            lang = detect(text)
+            audio_path = "/tmp/script_audio.mp3"
+            tts = gTTS(text=text, lang=lang)
+            tts.save(audio_path)
+
+            progress(0.4, desc="Running TRIBE v2 on generated audio...")
+            df = model.get_events_dataframe(audio_path=audio_path)
 
         else:
             import shutil
@@ -173,19 +234,8 @@ def analyze(input_mode, script_text, audio_file, progress=gr.Progress()):
             audio_path = f"/tmp/input_audio{ext}"
             shutil.copy(audio_file, audio_path)
 
-            progress(0.3, desc="Transcribing audio with WhisperX...")
-            try:
-                text_path = transcribe_audio_to_text(audio_path)
-                progress(0.45, desc="Running TRIBE v2 on transcript...")
-                df = model.get_events_dataframe(
-                    audio_path=audio_path,
-                    text_path=text_path
-                )
-            except Exception as wx_err:
-                # Fallback: use audio path directly and let TRIBE handle it
-                print(f"WhisperX error (falling back): {wx_err}")
-                progress(0.45, desc="Running TRIBE v2 on audio directly...")
-                df = model.get_events_dataframe(audio_path=audio_path)
+            progress(0.4, desc="Running TRIBE v2 on audio...")
+            df = model.get_events_dataframe(audio_path=audio_path)
 
         progress(0.6, desc="Predicting brain response...")
         preds, segments = model.predict(events=df)
@@ -208,6 +258,9 @@ def analyze(input_mode, script_text, audio_file, progress=gr.Progress()):
         print(full_error)
         return None, None, f"❌ Error:\n{str(e)}\n\nFull traceback:\n{full_error}", None
 
+# ---------------------------------------------------------------------------
+# Gradio UI
+# ---------------------------------------------------------------------------
 css = "#title{text-align:center} #subtitle{text-align:center;color:#888;font-size:14px}"
 
 with gr.Blocks(theme=gr.themes.Soft(primary_hue="indigo"), css=css) as demo:
@@ -257,4 +310,4 @@ with gr.Blocks(theme=gr.themes.Soft(primary_hue="indigo"), css=css) as demo:
     gr.Markdown("---\n*Powered by [TRIBE v2](https://github.com/facebookresearch/tribev2) by Meta FAIR*")
 
 if __name__ == "__main__":
-    demo.launch(server_name="0.0.0.0", server_port=7860)
+    demo.launch(server_name="0.0.0.0", server_port=7860)