update: add dioula models for STT and TTS

.gitignore

@@ -1 +1,2 @@
-venv
+venv
+.env

Dockerfile

@@ -9,4 +9,6 @@ COPY . .
 
 RUN pip install --no-cache-dir -r requirements.txt
 
+# EXPOSE 8000
+# CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
 CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860"]

functions/utils.py

@@ -0,0 +1,11 @@
+import soundfile as sf
+
+
+def getAudioDuration(filePath: str) -> float:
+    try:
+        data, samplerate = sf.read(filePath)
+        duration = len(data) / samplerate
+        return duration
+    except Exception as e:
+        print(f"Error getting audio duration: {e}")
+        return 0.0

language/dioula/dyu_stt.py

@@ -0,0 +1,23 @@
+import base64, tempfile, os
+from transformers import pipeline
+from functions.utils import getAudioDuration
+
+MODEL_NAME = "facebook/mms-1b-all"
+pipe = pipeline("automatic-speech-recognition", model=MODEL_NAME, model_kwargs={"target_lang": "dyu"})
+
+
+def dioulaSTT(audioBase64: str) -> dict:
+    audioBytes = base64.b64decode(audioBase64)
+
+    with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tempFile:
+        tempFile.write(audioBytes)
+        tempAudioPath = tempFile.name
+
+    try:
+        result = pipe(tempAudioPath)
+        text = result["text"]
+        duration = getAudioDuration(tempAudioPath)
+    finally:
+        os.remove(tempAudioPath)
+
+    return {'text': text, 'language': 'mos', 'duration': duration}

language/dioula/dyu_tts.py

@@ -0,0 +1,28 @@
+import torch, base64, tempfile, os
+import scipy.io.wavfile as wavfile
+from transformers import VitsModel, VitsTokenizer
+
+MODEL_NAME = "facebook/mms-tts-dyu"
+tokenizer = VitsTokenizer.from_pretrained(MODEL_NAME)
+model = VitsModel.from_pretrained(MODEL_NAME)
+
+
+def dioulaTTS(text: str) -> str:
+    inputs = tokenizer(text, return_tensors="pt")
+
+    with torch.no_grad():
+        output = model(**inputs)
+
+    waveform = output.waveform[0].cpu().numpy()
+
+    with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tempFile:
+        wavfile.write(tempFile.name, rate=model.config.sampling_rate, data=waveform)
+        tempAudioPath = tempFile.name
+
+    try:
+        with open(tempAudioPath, "rb") as file:
+            audioBase64 = base64.b64encode(file.read()).decode("utf-8")
+    finally:
+        os.remove(tempAudioPath) 
+
+    return audioBase64

language/{mos_stt.py → moore/mos_stt.py}

@@ -1,6 +1,6 @@
 import base64, tempfile, os
 from transformers import pipeline
-import soundfile as sf
+from functions.utils import getAudioDuration
 # from huggingface_hub import login
 
 MODEL_NAME = "facebook/mms-1b-all"
@@ -25,13 +25,3 @@ def mooreSTT(audioBase64: str) -> dict:
         os.remove(tempAudioPath)
 
     return {'text': text, 'language': 'mos', 'duration': duration}
-
-
-def getAudioDuration(filePath: str) -> float:
-    try:
-        data, samplerate = sf.read(filePath)
-        duration = len(data) / samplerate
-        return duration
-    except Exception as e:
-        print(f"Error getting audio duration: {e}")
-        return 0.0

main.py

@@ -2,8 +2,11 @@ from fastapi import FastAPI, Request, HTTPException
 from functions.translation import translateText
 from functions.speech_to_text import speechToText
 from functions.text_to_speech import textToSpeech
-from language.mos_stt import mooreSTT
-from language.mos_tts import mooreTTS
+from language.moore.mos_stt import mooreSTT
+from language.moore.mos_tts import mooreTTS
+from language.dioula.dyu_stt import dioulaSTT
+from language.dioula.dyu_tts import dioulaTTS
+
 
 
 app = FastAPI(
@@ -12,7 +15,7 @@ app = FastAPI(
 )
 
 
-@app.post("/translateText")
+@app.post("/nllb/translateText")
 async def translate(request: Request):
     body: dict = await request.json()
     try:
@@ -27,8 +30,8 @@ async def translate(request: Request):
         raise HTTPException(status_code=400, detail=f"Translate error: {e}")
 
 
-@app.post("/speechToText")
-async def convertSpeechToText(request: Request):
+@app.post("/whisper/speechToText")
+async def whisperSpeechToText(request: Request):
     body: dict = await request.json()
     try:
         audioBase64 = body.get('audioBase64')
@@ -41,8 +44,8 @@ async def convertSpeechToText(request: Request):
         raise HTTPException(status_code=400, detail=f"STT error: {e}")
 
 
-@app.post("/textToSpeech")
-async def convertTextToSpeech(request: Request):
+@app.post("/edge/textToSpeech")
+async def edgeTextToSpeech(request: Request):
     body: dict = await request.json()
     try:
         text = body.get('text')
@@ -55,27 +58,41 @@ async def convertTextToSpeech(request: Request):
         raise HTTPException(status_code=400, detail=f"TTS error: {e}")
 
 
-@app.post("/moore/speechToText")
-async def mooreSpeechToText(request: Request):
+@app.post("/mms/speechToText")
+async def mmsSpeechToText(request: Request):
     body: dict = await request.json()
     try:
         audioBase64 = body.get('audioBase64')
-        
-        data = mooreSTT(audioBase64=audioBase64)
-        return data
+        sourceLang = body.get('sourceLang')
+
+        if sourceLang == 'mos':
+            data = mooreSTT(audioBase64=audioBase64)
+            return data
+        elif sourceLang == 'dyu':
+            data = dioulaSTT(audioBase64=audioBase64)
+            return data
+        else:
+            raise HTTPException(status_code=400, detail=f"STT error: Invalid sourceLang - {sourceLang}")
     except Exception as e:
         print(f"STT error: {e}")
         raise HTTPException(status_code=400, detail=f"STT error: {e}")
 
 
-@app.post("/moore/textToSpeech")
-async def mooreTextToSpeech(request: Request):
+@app.post("/mms/textToSpeech")
+async def mmsTextToSpeech(request: Request):
     body: dict = await request.json()
     try:
         text = body.get('text')
+        sourceLang = body.get('sourceLang')
 
-        audioBase64 = mooreTTS(text=text)
-        return { 'audioBase64': audioBase64 }
+        if sourceLang == 'mos':
+            audioBase64 = mooreTTS(text=text)
+            return { 'audioBase64': audioBase64 }
+        elif sourceLang == 'dyu':
+            audioBase64 = dioulaTTS(text=text)
+            return { 'audioBase64': audioBase64 }
+        else:
+            raise HTTPException(status_code=400, detail=f"STT error: Invalid sourceLang - {sourceLang}")
     except Exception as e:
         print(f"TTS error: {e}")
         raise HTTPException(status_code=400, detail=f"TTS error: {e}")