Upload app.py with huggingface_hub

app.py

@@ -60,8 +60,8 @@ if not hasattr(torchaudio, "info"):
 
 from df.enhance import enhance, init_df, load_audio, save_audio
 
-# FORCE BUILD TRIGGER: 10:10:00 Jan 21 2026
-# v86: Pre-load to CPU RAM + Fast Transfer to GPU (Prevents ZeroGPU timeouts)
+# 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:
@@ -89,77 +89,67 @@ os.environ["COQUI_TOS_AGREED"] = "1"
 # Global models (Resident in RAM)
 MODELS = {"stt": None, "translate": None, "tts": None, "denoiser": None}
 
-def load_models():
-    """Fast GPU Activation: Moves pre-loaded CPU models to GPU (v86)"""
+def activate_gpu_models(action):
+    """v87: Targetted activation of models on GPU to save time"""
     global MODELS
     
-    # 1. Faster-Whisper (Must re-init for device change, but disk cache is hot)
-    if MODELS["stt"] is None:
-        print("🎙️ Initializing Faster-Whisper...")
-        from faster_whisper import WhisperModel
-        dev = "cuda" if torch.cuda.is_available() else "cpu"
-        ct = "float16" if dev == "cuda" else "int8"
-        MODELS["stt"] = WhisperModel("large-v3", device=dev, compute_type=ct)
-    
-    # 2. XTTS-v2 (Efficient .to("cuda") transfer)
-    if MODELS["tts"] is None:
-        print("🔊 Loading XTTS-v2 into Engine (CPU Base)...")
-        from TTS.api import TTS
-        MODELS["tts"] = TTS(model_name="tts_models/multilingual/multi-dataset/xtts_v2", gpu=False)
-    
-    if torch.cuda.is_available():
-        # Move XTTS to GPU if it's currently on CPU
-        # We check the device of the underlying model to avoid redundant moves
+    # 1. Faster-Whisper (Activate only if action needs it)
+    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:
+            print(f"🎙️ Activating Whisper on GPU for {action}...")
+            from faster_whisper import WhisperModel
+            MODELS["stt"] = WhisperModel("large-v3", device="cuda", compute_type="float16")
+
+    # 2. XTTS-v2 (Activate only if action needs it)
+    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("🚀 Moving XTTS-v2 to GPU...")
+                print(f"🚀 Activating XTTS-v2 on GPU for {action}...")
                 MODELS["tts"].to("cuda")
-        except: 
-            # Fallback if structure differs
-            MODELS["tts"].to("cuda")
-
-    # 3. DeepFilterNet
-    if MODELS["denoiser"] is None:
-        try: MODELS["denoiser"] = init_df()
-        except: pass
-
-    # 4. Chatterbox ONNX
-    chatterbox_utils.load_chatterbox(device="cuda" if torch.cuda.is_available() else "cpu")
+        except:
+             MODELS["tts"].to("cuda")
 
-    # 5. Translate
-    if MODELS["translate"] is None:
-        MODELS["translate"] = "active"
+    # 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")
 
-    # 🧹 Proactive Memory Cleanup
+    # 🧹 Cleanup
     gc.collect()
     if torch.cuda.is_available():
         torch.cuda.empty_cache()
 
 def warmup_models():
-    """PRE-LOAD EVERYTHING INTO SYSTEM RAM (CPU)"""
-    print("\n🔥 --- SYSTEM WARMUP: RESIDENT RAM LOADING (v86) ---")
+    """Download models at startup to System RAM"""
+    print("\n🔥 --- SYSTEM WARMUP: RAM CACHING (v87) ---")
     start = time.time()
     try:
-        # Load Whisper into RAM
-        print("📥 Pre-loading Whisper to RAM...")
         from faster_whisper import WhisperModel
+        print("📥 Caching Whisper to RAM...")
         MODELS["stt"] = WhisperModel("large-v3", device="cpu", compute_type="int8")
         
-        # Load XTTS into RAM (The heaviest part)
-        print("📥 Pre-loading XTTS-v2 to RAM...")
         from TTS.api import TTS
+        print("📥 Caching XTTS-v2 to RAM...")
         MODELS["tts"] = TTS(model_name="tts_models/multilingual/multi-dataset/xtts_v2", gpu=False)
         
-        # Load Denoiser
-        print("📥 Pre-loading Denoiser...")
-        try: MODELS["denoiser"] = init_df()
-        except: pass
-        
-        # Pre-download ONNX weights
         chatterbox_utils.warmup_chatterbox()
-        
-        print(f"✅ --- WARMUP COMPLETE: All models resident in RAM ({time.time()-start:.2f}s) --- \n")
+        print(f"✅ --- WARMUP COMPLETE ({time.time()-start:.2f}s) --- \n")
     except Exception as e:
         print(f"⚠️ Warmup warning: {e}")
 
@@ -167,12 +157,10 @@ def _stt_logic(request_dict):
     audio_bytes = base64.b64decode(request_dict.get("file"))
     lang = request_dict.get("lang")
     with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
-        f.write(audio_bytes)
-        temp_path = f.name
+        f.write(audio_bytes); temp_path = f.name
     try:
         segments, _ = MODELS["stt"].transcribe(temp_path, language=lang, beam_size=1)
-        text = " ".join([s.text for s in segments]).strip()
-        return {"text": text}
+        return {"text": " ".join([s.text for s in segments]).strip()}
     finally:
         if os.path.exists(temp_path): os.unlink(temp_path)
 
@@ -182,28 +170,22 @@ def _translate_logic(text, target_lang):
 
 def _tts_logic(text, lang, speaker_wav_b64):
     if not text or not text.strip(): return {"error": "Input empty"}
-        
     XTTS_MAP = {
         "en": "en", "de": "de", "fr": "fr", "es": "es", "it": "it", "pl": "pl",
         "pt": "pt", "tr": "tr", "ru": "ru", "nl": "nl", "cs": "cs", "ar": "ar",
         "hu": "hu", "ko": "ko", "hi": "hi", "zh": "zh-cn"
     }
-    
     clean_lang = lang.strip().lower().split('-')[0]
-    mapped_lang = XTTS_MAP.get(clean_lang)
-    if clean_lang == "zh": mapped_lang = "zh-cn"
-        
-    print(f"[v86] TTS: {lang} -> {mapped_lang}")
+    mapped_lang = XTTS_MAP.get(clean_lang) or ("zh-cn" if clean_lang == "zh" else None)
     
     if mapped_lang:
-        print(f"[v86] GPU Mode: XTTS-v2")
+        print(f"[v87] Use XTTS: {mapped_lang}")
         speaker_wav_path = None
         if speaker_wav_b64:
             sb = base64.b64decode(speaker_wav_b64)
             with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
                 f.write(sb); speaker_wav_path = f.name
         else: speaker_wav_path = "default_speaker.wav"
-            
         try:
             with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as output_file:
                 output_path = output_file.name
@@ -214,27 +196,25 @@ def _tts_logic(text, lang, speaker_wav_b64):
             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 'output_path' in locals() and os.path.exists(output_path): os.unlink(output_path)
-            
-    # Fallback to Chatterbox
-    print(f"[v86] Fallback Mode: Chatterbox ONNX")
+    
+    print(f"[v87] Use Chatterbox: {clean_lang}")
     try:
         temp_ref = None
         if speaker_wav_b64:
-             sb = base64.b64decode(speaker_wav_b64)
-             with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
-                 f.write(sb); temp_ref = f.name
+            sb = base64.b64decode(speaker_wav_b64)
+            with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
+                f.write(sb); temp_ref = f.name
         audio_bytes = chatterbox_utils.run_chatterbox_inference(text, clean_lang, speaker_wav_path=temp_ref)
         if temp_ref and os.path.exists(temp_ref): os.unlink(temp_ref)
         return {"audio": base64.b64encode(audio_bytes).decode()}
-    except Exception as e:
-        return {"error": f"TTS Failure: {str(e)}"}
+    except Exception as e: return {"error": f"TTS Failure: {str(e)}"}
 
 @spaces.GPU
 def core_process(request_dict):
     action = request_dict.get("action")
     t0 = time.time()
-    print(f"--- [v86] 🚀 GPU SESSION START: {action} ---")
-    load_models() 
+    print(f"--- [v87] 🚀 GPU START: {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"))}
@@ -242,48 +222,18 @@ def core_process(request_dict):
         elif action == "s2st":
             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 detected"}
+            if not text: return {"error": "No speech"}
             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"--- [v86] ✨ SESSION END: {action} ({time.time()-t0:.2f}s) ---")
+        print(f"--- [v87] ✨ END: {action} ({time.time()-t0:.2f}s) ---")
         gc.collect()
         if torch.cuda.is_available(): torch.cuda.empty_cache()
     return res
 
-def create_wav_header(sample_rate=24000, channels=1, bit_depth=16):
-    header = bytearray(b'RIFF')
-    header.extend((1000000000).to_bytes(4, 'little')) 
-    header.extend(b'WAVEfmt ')
-    header.extend((16).to_bytes(4, 'little'))
-    header.extend((1).to_bytes(2, 'little'))
-    header.extend((channels).to_bytes(2, 'little'))
-    header.extend((sample_rate).to_bytes(4, 'little'))
-    header.extend((sample_rate * channels * (bit_depth // 8)).to_bytes(4, 'little'))
-    header.extend((channels * (bit_depth // 8)).to_bytes(2, 'little'))
-    header.extend((bit_depth).to_bytes(2, 'little'))
-    header.extend(b'data')
-    header.extend((0xFFFFFFFF).to_bytes(4, 'little'))
-    return bytes(header)
-
-@spaces.GPU
-def gpu_tts_generator(text, lang, speaker_wav_path):
-    load_models()
-    try:
-        yield bytes(create_wav_header(sample_rate=24000))
-        for chunk in MODELS["tts"].synthesizer.tts_model.inference_stream(
-            text, lang, *MODELS["tts"].synthesizer.tts_model.get_conditioning_latents(audio_path=[speaker_wav_path]),
-            stream_chunk_size=20
-        ):
-            yield bytes((chunk * 32767).to(torch.int16).cpu().numpy().tobytes())
-    finally:
-        if speaker_wav_path and "default_speaker" not in speaker_wav_path and os.path.exists(speaker_wav_path): os.unlink(speaker_wav_path)
-        gc.collect()
-        if torch.cuda.is_available(): torch.cuda.empty_cache()
-
 app = FastAPI()
 
 @app.post("/api/v1/process")
@@ -295,19 +245,6 @@ async def api_process(request: Request):
         traceback.print_exc()
         return {"error": str(e)}
 
-@app.post("/api/v1/tts_stream")
-async def api_tts_stream(request: Request):
-    try:
-        data = await request.json()
-        speaker_wav_b64 = data.get("speaker_wav")
-        if speaker_wav_b64:
-            sb = base64.b64decode(speaker_wav_b64)
-            with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
-                f.write(sb); speaker_wav_path = f.name
-        else: speaker_wav_path = "default_speaker.wav"
-        return StreamingResponse(gpu_tts_generator(data.get("text"), data.get("lang"), speaker_wav_path), media_type="audio/wav")
-    except Exception as e: return {"error": str(e)}
-
 @app.get("/health")
 def health(): return {"status": "ok", "gpu": torch.cuda.is_available(), "time": time.ctime()}