Upload app.py with huggingface_hub

app.py

@@ -13,6 +13,15 @@ import torchaudio
 import chatterbox_utils
 import gc
 
+# 🛡️ BULKY IMPORTS AT TOP-LEVEL (v88 Optimization)
+# Pre-loading these into RAM at startup so they are READY when GPU session starts
+print("📦 Pre-loading AI Engines into RAM...")
+from faster_whisper import WhisperModel
+from TTS.api import TTS
+from df.enhance import init_df, enhance, load_audio, save_audio
+import deep_translator
+print("✅ Imports Complete")
+
 # 🛡️ ZeroGPU Support
 try:
     import spaces
@@ -21,8 +30,9 @@ except ImportError:
     print("⚠️ Spaces library not found. Using mock decorator for local run.")
     class spaces:
         @staticmethod
-        def GPU(f): return f
-
+        def GPU(duration=60, f=None):
+            if f is None: return lambda x: x
+            return f
 
 # 🛠️ Monkeypatch torchaudio.backend (DeepFilterNet compatibility)
 import sys
@@ -58,100 +68,73 @@ if not hasattr(torchaudio, "info"):
              return SimpleNamespace(sample_rate=48000, num_frames=0, num_channels=1)
     torchaudio.info = mock_info
 
-from df.enhance import enhance, init_df, load_audio, save_audio
-
-# FORCE BUILD TRIGGER: 10:20:00 Jan 21 2026
-# v87: Targeted GPU Activation (Only loads what's needed for the specific action)
-
-# 🛠️ Monkeypatch torchaudio.load
-try:
-    _orig_load = torchaudio.load
-    def patched_load(filepath, *args, **kwargs):
-        try:
-            return _orig_load(filepath, *args, **kwargs)
-        except ImportError as e:
-            if "torchcodec" in str(e).lower():
-                print(f"⚠️ Redirecting load for {filepath} via soundfile")
-                import soundfile as sf
-                data, samplerate = sf.read(filepath)
-                t = torch.from_numpy(data).float()
-                if len(t.shape) == 1: t = t.unsqueeze(0)
-                else: t = t.T
-                return t, samplerate
-            raise e
-    torchaudio.load = patched_load
-    print("✅ Torchaudio monkeypatched successfully")
-except Exception as e:
-    print(f"⚠️ Failed to monkeypatch torchaudio: {e}")
+# FORCE BUILD TRIGGER: 10:30:00 Jan 21 2026
+# v88: Mandatory GPU-Only (STT + TTS). Fast Activation + 150s Duration.
 
 os.environ["COQUI_TOS_AGREED"] = "1"
 
-# Global models (Resident in RAM)
+# Global models (Resident in System RAM)
 MODELS = {"stt": None, "translate": None, "tts": None, "denoiser": None}
 
 def activate_gpu_models(action):
-    """v87: Targetted activation of models on GPU to save time"""
+    """v88: Fast GPU Movement and Activation"""
     global MODELS
     
-    # 1. Faster-Whisper (Activate only if action needs it)
+    # 1. Faster-Whisper GPU Activation
     if action in ["stt", "s2st"]:
-        is_cuda = False
-        try:
-            # Check current device
-            if hasattr(MODELS["stt"], "model") and MODELS["stt"].model.device == "cuda":
-                is_cuda = True
-        except: pass
-        
-        if not is_cuda:
+        if MODELS["stt"] is None or MODELS["stt"].model.device != "cuda":
             print(f"🎙️ Activating Whisper on GPU for {action}...")
-            from faster_whisper import WhisperModel
+            # We re-init to move to CUDA. Since weights are cached, this is fast.
             MODELS["stt"] = WhisperModel("large-v3", device="cuda", compute_type="float16")
 
-    # 2. XTTS-v2 (Activate only if action needs it)
+    # 2. XTTS-v2 GPU Activation
     if action in ["tts", "s2st"]:
         if MODELS["tts"] is None:
-            from TTS.api import TTS
             print("🔊 Initializing XTTS to RAM...")
             MODELS["tts"] = TTS(model_name="tts_models/multilingual/multi-dataset/xtts_v2", gpu=False)
         
         try:
             current_dev = str(next(MODELS["tts"].synthesizer.tts_model.parameters()).device)
             if "cuda" not in current_dev:
-                print(f"🚀 Activating XTTS-v2 on GPU for {action}...")
+                print(f"🚀 Moving XTTS-v2 to GPU...")
                 MODELS["tts"].to("cuda")
         except:
              MODELS["tts"].to("cuda")
 
-    # 3. Denoiser & Translate & Chatterbox
-    if action in ["tts", "s2st", "stt"]:
-         if MODELS["denoiser"] is None:
-             try: MODELS["denoiser"] = init_df()
-             except: pass
-         if MODELS["translate"] is None: MODELS["translate"] = "active"
-         chatterbox_utils.load_chatterbox(device="cuda" if torch.cuda.is_available() else "cpu")
+    # 3. Denoiser & Translate
+    if MODELS["denoiser"] is None:
+        try: MODELS["denoiser"] = init_df()
+        except: pass
+    if MODELS["translate"] is None: 
+        MODELS["translate"] = "active"
+    
+    # Chatterbox (STAY CPU if no GPU available for it, or use CUDA if ONNX allows)
+    chatterbox_utils.load_chatterbox(device="cuda" if torch.cuda.is_available() else "cpu")
 
-    # 🧹 Cleanup
+    # 🧹 Mem Cleanup
     gc.collect()
     if torch.cuda.is_available():
         torch.cuda.empty_cache()
 
 def warmup_models():
-    """Download models at startup to System RAM"""
-    print("\n🔥 --- SYSTEM WARMUP: RAM CACHING (v87) ---")
+    """PRE-LOAD EVERYTHING INTO SYSTEM RAM (CPU)"""
+    print("\n🔥 --- SYSTEM STARTUP: RESIDENT RAM LOADING (v88) ---")
     start = time.time()
     try:
-        from faster_whisper import WhisperModel
-        print("📥 Caching Whisper to RAM...")
+        print("📥 Pre-loading Whisper large-v3 to RAM...")
         MODELS["stt"] = WhisperModel("large-v3", device="cpu", compute_type="int8")
         
-        from TTS.api import TTS
-        print("📥 Caching XTTS-v2 to RAM...")
+        print("📥 Pre-loading XTTS-v2 to RAM...")
         MODELS["tts"] = TTS(model_name="tts_models/multilingual/multi-dataset/xtts_v2", gpu=False)
         
+        print("📥 Pre-loading DeepFilterNet...")
+        try: MODELS["denoiser"] = init_df()
+        except: pass
+        
         chatterbox_utils.warmup_chatterbox()
-        print(f"✅ --- WARMUP COMPLETE ({time.time()-start:.2f}s) --- \n")
+        print(f"✅ --- SYSTEM READY: MODELS IN RAM ({time.time()-start:.2f}s) --- \n")
     except Exception as e:
-        print(f"⚠️ Warmup warning: {e}")
+        print(f"⚠️ Startup warning: {e}")
 
 def _stt_logic(request_dict):
     audio_bytes = base64.b64decode(request_dict.get("file"))
@@ -160,7 +143,8 @@ def _stt_logic(request_dict):
         f.write(audio_bytes); temp_path = f.name
     try:
         segments, _ = MODELS["stt"].transcribe(temp_path, language=lang, beam_size=1)
-        return {"text": " ".join([s.text for s in segments]).strip()}
+        text = " ".join([s.text for s in segments]).strip()
+        return {"text": text}
     finally:
         if os.path.exists(temp_path): os.unlink(temp_path)
 
@@ -179,7 +163,7 @@ def _tts_logic(text, lang, speaker_wav_b64):
     mapped_lang = XTTS_MAP.get(clean_lang) or ("zh-cn" if clean_lang == "zh" else None)
     
     if mapped_lang:
-        print(f"[v87] Use XTTS: {mapped_lang}")
+        print(f"[v88] GPU Inference: XTTS-v2 for '{mapped_lang}'")
         speaker_wav_path = None
         if speaker_wav_b64:
             sb = base64.b64decode(speaker_wav_b64)
@@ -193,11 +177,11 @@ def _tts_logic(text, lang, speaker_wav_b64):
             with open(output_path, "rb") as f: audio_b64 = base64.b64encode(f.read()).decode()
             return {"audio": audio_b64}
         finally:
-            if speaker_wav_path and "default_speaker" not in speaker_wav_path:
-                if os.path.exists(speaker_wav_path): os.unlink(speaker_wav_path)
+            if speaker_wav_path and "default_speaker" not in speaker_wav_path and os.path.exists(speaker_wav_path): os.unlink(speaker_wav_path)
             if 'output_path' in locals() and os.path.exists(output_path): os.unlink(output_path)
     
-    print(f"[v87] Use Chatterbox: {clean_lang}")
+    # Fallback path
+    print(f"[v88] Inference: Chatterbox Fallback for '{clean_lang}'")
     try:
         temp_ref = None
         if speaker_wav_b64:
@@ -209,27 +193,33 @@ def _tts_logic(text, lang, speaker_wav_b64):
         return {"audio": base64.b64encode(audio_bytes).decode()}
     except Exception as e: return {"error": f"TTS Failure: {str(e)}"}
 
-@spaces.GPU
+# 🚀 AGGRESSIVE GPU SESSION (150s Duration)
+@spaces.GPU(duration=150)
 def core_process(request_dict):
+    """MANDATORY GPU ENTRY POINT (v88)"""
     action = request_dict.get("action")
     t0 = time.time()
-    print(f"--- [v87] 🚀 GPU START: {action} ---")
+    print(f"--- [v88] 🚀 GPU SESSION START: {action} ---")
+    
+    # v88 Optimization: Only activate models for current action
     activate_gpu_models(action)
+    
     try:
         if action == "stt": res = _stt_logic(request_dict)
         elif action == "translate": res = {"translated": _translate_logic(request_dict.get("text"), request_dict.get("target_lang", "en"))}
         elif action == "tts": res = _tts_logic(request_dict.get("text"), request_dict.get("lang"), request_dict.get("speaker_wav"))
         elif action == "s2st":
+            # Direct GPU Pipeline
             stt_res = _stt_logic({"file": request_dict.get("file"), "lang": request_dict.get("source_lang")})
             text = stt_res.get("text", "")
-            if not text: return {"error": "No speech"}
+            if not text: return {"error": "No speech detected"}
             translated = _translate_logic(text, request_dict.get("target_lang"))
             tts_res = _tts_logic(translated, request_dict.get("target_lang"), request_dict.get("speaker_wav"))
             res = {"text": text, "translated": translated, "audio": tts_res.get("audio")}
         elif action == "health": res = {"status": "awake"}
         else: res = {"error": f"Unknown action: {action}"}
     finally:
-        print(f"--- [v87] ✨ END: {action} ({time.time()-t0:.2f}s) ---")
+        print(f"--- [v88] ✨ SESSION END: {action} ({time.time()-t0:.2f}s) ---")
         gc.collect()
         if torch.cuda.is_available(): torch.cuda.empty_cache()
     return res