Update app.py

app.py

@@ -1,110 +1,74 @@
 import os
+import io
+import torch
+import librosa
+import edge_tts
+import scipy.io.wavfile as wavfile
+from fastapi import FastAPI, Query, File, UploadFile, HTTPException
+from fastapi.responses import StreamingResponse
+from transformers import VitsModel, AutoTokenizer, Wav2Vec2ForCTC, AutoProcessor
 
+# 1. Set cache before importing/loading models
 os.environ["HF_HOME"] = "/tmp/hf"
-os.environ["TRANSFORMERS_CACHE"] = "/tmp/hf"
-os.environ["HF_DATASETS_CACHE"] = "/tmp/hf"
 os.makedirs("/tmp/hf", exist_ok=True)
 
-from fastapi import FastAPI, Query, File, UploadFile, HTTPException
-from fastapi.responses import StreamingResponse
-from transformers import VitsModel, AutoTokenizer, Wav2Vec2ForCTC, AutoProcessor
-import torch, scipy.io.wavfile as wavfile
-import io
-import librosa
-import edge_tts
+app = FastAPI(title="Bambara AI API")
 
+# 2. Load Models (Memory Efficient)
+# Use .to("cpu") explicitly if you don't have a GPU on the free tier
+device = "cuda" if torch.cuda.is_available() else "cpu"
 
-app = FastAPI(title="Bambara TTS API")
+# TTS Model
+tts_model_id = "facebook/mms-tts-bam"
+tts_tokenizer = AutoTokenizer.from_pretrained(tts_model_id)
+tts_model = VitsModel.from_pretrained(tts_model_id).to(device)
 
-# Load model once at startup
-model = VitsModel.from_pretrained("facebook/mms-tts-bam")
-tokenizer = AutoTokenizer.from_pretrained("facebook/mms-tts-bam")
-sampling_rate = model.config.sampling_rate
-# Load model once when the server starts
-speech_model_id = "facebook/mms-1b-all"
-processor = AutoProcessor.from_pretrained(speech_model_id)
-speech_model = Wav2Vec2ForCTC.from_pretrained(speech_model_id)
+# ASR (Speech-to-Text) Model
+asr_model_id = "facebook/mms-1b-all"
+asr_processor = AutoProcessor.from_pretrained(asr_model_id)
+asr_model = Wav2Vec2ForCTC.from_pretrained(asr_model_id).to(device)
 
+# Pre-load the Bambara adapter so it doesn't slow down the first request
+asr_processor.tokenizer.set_target_lang("bam")
+asr_model.load_adapter("bam")
 
 @app.get("/tts/")
-async def tts(text: str = Query(..., description="Bambara text to synthesize")):
-    inputs = tokenizer(text, return_tensors="pt")
-    inputs = {k: v.to("cpu") for k, v in inputs.items()}
-
+async def tts(text: str = Query(..., description="Bambara text")):
+    inputs = tts_tokenizer(text, return_tensors="pt").to(device)
     with torch.no_grad():
-        output = model(**inputs).waveform
-
-    waveform = output[0]
-
-    # Stream audio instead of saving to disk
+        output = tts_model(**inputs).waveform
+    
     buffer = io.BytesIO()
-    wavfile.write(buffer, rate=sampling_rate, data=waveform.numpy())
+    wavfile.write(buffer, rate=tts_model.config.sampling_rate, data=output[0].cpu().numpy())
     buffer.seek(0)
-
     return StreamingResponse(buffer, media_type="audio/wav")
 
-
-@app.get("/noneBmTts/")
-async def noneBmTts(
-    text: str = Query(..., description="Text to synthesize"),
-    voice: str = Query(
-        "fr-FR-DeniseNeural", description="Voice ID (e.g., en-US-GuyNeural)"
-    ),
-):
-    try:
-        # Create the Communicate object with the requested text and voice
-        communicate = edge_tts.Communicate(text, voice)
-
-        buffer = io.BytesIO()
-
-        # Stream the audio chunks into the buffer
-        async for chunk in communicate.stream():
-            if chunk["type"] == "audio":
-                buffer.write(chunk["data"])
-
-        # Check if we actually got data
-        if buffer.tell() == 0:
-            raise HTTPException(
-                status_code=400, detail="Synthesis failed to produce audio."
-            )
-
-        buffer.seek(0)
-        return StreamingResponse(buffer, media_type="audio/mpeg")
-
-    except Exception as e:
-        # Catch errors like invalid voice names
-        raise HTTPException(status_code=400, detail=str(e))
-
-
-
-
 @app.post("/transcribe")
 async def transcribe(audio_file: UploadFile = File(...)):
-    # 1. Check if a file was actually uploaded
-    if not audio_file:
-        raise HTTPException(status_code=400, detail="No file uploaded")
-
     try:
-        # 2. Read the file into memory
+        # Read and load audio
         audio_bytes = await audio_file.read()
-        
-        # 3. Load and Resample to 16,000 Hz using librosa
-        # io.BytesIO(audio_bytes) lets librosa treat the bytes like a file
         audio_data, _ = librosa.load(io.BytesIO(audio_bytes), sr=16000)
 
-        # 4. Setup Bambara Adapter
-        processor.tokenizer.set_target_lang("bam")
-        model.load_adapter("bam")
-
-        # 5. Perform Inference
-        inputs = processor(audio_data, sampling_rate=16_000, return_tensors="pt")
+        # Prepare inputs
+        inputs = asr_processor(audio_data, sampling_rate=16000, return_tensors="pt").to(device)
+        
         with torch.no_grad():
-            logits = speech_model(**inputs).logits
+            logits = asr_model(**inputs).logits
 
         predicted_ids = torch.argmax(logits, dim=-1)
-        transcription = processor.batch_decode(predicted_ids)[0]
+        transcription = asr_processor.batch_decode(predicted_ids)[0]
 
         return {"text": transcription}
-
     except Exception as e:
-        raise HTTPException(status_code=500, detail=f"Error processing audio: {str(e)}")
+        raise HTTPException(status_code=500, detail=str(e))
+
+@app.get("/noneBmTts/")
+async def noneBmTts(text: str, voice: str = "fr-FR-DeniseNeural"):
+    communicate = edge_tts.Communicate(text, voice)
+    buffer = io.BytesIO()
+    async for chunk in communicate.stream():
+        if chunk["type"] == "audio":
+            buffer.write(chunk["data"])
+    buffer.seek(0)
+    return StreamingResponse(buffer, media_type="audio/mpeg")