Update app.py

app.py

@@ -1,89 +1,71 @@
 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.middleware.cors import CORSMiddleware
-from fastapi.responses import StreamingResponse
-from transformers import VitsModel, AutoTokenizer, Wav2Vec2ForCTC, AutoProcessor
 
-# 1. Environment and App Setup
 os.environ["HF_HOME"] = "/tmp/hf"
-app = FastAPI(title="Bambara AI API")
-
-app.add_middleware(
-    CORSMiddleware,
-    allow_origins=["*"], 
-    allow_credentials=True,
-    allow_methods=["*"],
-    allow_headers=["*"],
-)
-
-device = "cpu"
-
-# 2. Load Models (Switching to 300M for stability)
-# ASR Model
-asr_model_id = "facebook/mms-300m-1107" # Smaller, faster, more stable
-asr_processor = AutoProcessor.from_pretrained(asr_model_id)
-asr_model = Wav2Vec2ForCTC.from_pretrained(asr_model_id).to(device)
-
-# Load Bambara Adapter
-asr_processor.tokenizer.set_target_lang("bam")
-asr_model.load_adapter("bam")
-
-# 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)
-
-@app.post("/transcribe")
-async def transcribe(audio_file: UploadFile = File(...)):
-    try:
-        # Read file
-        content = await audio_file.read()
-        if not content:
-            raise HTTPException(status_code=400, detail="Empty audio file")
+os.environ["TRANSFORMERS_CACHE"] = "/tmp/hf"
+os.environ["HF_DATASETS_CACHE"] = "/tmp/hf"
+os.makedirs("/tmp/hf", exist_ok=True)
 
-        # Load audio into memory
-        # Resampling here to 16kHz is mandatory
-        audio_data, _ = librosa.load(io.BytesIO(content), sr=16000)
+from fastapi import FastAPI, Query
+from fastapi.responses import StreamingResponse
+from transformers import VitsModel, AutoTokenizer
+import torch, scipy.io.wavfile as wavfile
+import io
+import edge_tts
 
-        # Prepare for model
-        inputs = asr_processor(audio_data, sampling_rate=16000, return_tensors="pt").to(device)
-        
-        # Inference
-        with torch.inference_mode():
-            logits = asr_model(**inputs).logits
 
-        # Decode
-        predicted_ids = torch.argmax(logits, dim=-1)
-        transcription = asr_processor.batch_decode(predicted_ids)[0]
+app = FastAPI(title="Bambara TTS API")
 
-        return {"text": transcription}
+# 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
 
-    except Exception as e:
-        print(f"Error: {e}")
-        raise HTTPException(status_code=500, detail=str(e))
 
 @app.get("/tts/")
-async def tts(text: str = Query(..., description="Bambara text")):
-    inputs = tts_tokenizer(text, return_tensors="pt").to(device)
-    with torch.inference_mode():
-        output = tts_model(**inputs).waveform
-    
+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()}
+
+    with torch.no_grad():
+        output = model(**inputs).waveform
+
+    waveform = output[0]
+
+    # Stream audio instead of saving to disk
     buffer = io.BytesIO()
-    wavfile.write(buffer, rate=tts_model.config.sampling_rate, data=output[0].cpu().numpy())
+    wavfile.write(buffer, rate=sampling_rate, data=waveform.numpy())
     buffer.seek(0)
+
     return StreamingResponse(buffer, media_type="audio/wav")
 
+
 @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")
+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))