Spaces:
Running
Running
File size: 2,331 Bytes
992207f 73677f4 992207f 1e6dfe6 992207f e9bda65 992207f 91a4c08 992207f 91a4c08 992207f d5a16e7 992207f b221992 8582da0 b221992 59ec87d 992207f b221992 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 |
from fastapi import APIRouter, File, UploadFile, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from ai4b import BhashiniClient
from fast_langdetect import detect
import io
import base64
import os
router = APIRouter()
ULCA_USER_ID = os.getenv("ULCA_USER_ID")
ULCA_API_KEY = os.getenv("ULCA_API_KEY")
client = BhashiniClient(user_id=ULCA_USER_ID, api_key=ULCA_API_KEY)
class TTSRequest(BaseModel):
text: str
voice: str = "female"
SUPPORTED_LANGUAGES = {'pa', 'mr', 'bn', 'en', 'as', 'or', 'ta', 'te', 'kn', 'gu', 'hi', 'ml'}
def detect_language(text):
text = text.replace("\n", " ")
try:
result = detect(text, low_memory=False)
detected_lang = result['lang']
if detected_lang in SUPPORTED_LANGUAGES:
return detected_lang
except:
pass
if any('\u0980' <= char <= '\u09FF' for char in text):
return 'brx'
elif any('\uABC0' <= char <= '\uABFF' for char in text):
return 'mni'
return 'en'
@router.post("/tts")
async def text_to_speech(request: TTSRequest):
try:
detected_language = detect_language(request.text)
tts_result = client.tts(
request.text,
source_language=detected_language,
gender=request.voice
)
audio_base64 = tts_result['pipelineResponse'][0]['audio'][0]['audioContent']
audio_data = base64.b64decode(audio_base64)
return StreamingResponse(io.BytesIO(audio_data), media_type="audio/wav")
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.post("/asr")
async def speech_to_text(file: UploadFile = File(...), source_language: str = "en"):
try:
audio_content = await file.read()
asr_result = client.asr(audio_content, source_language=source_language)
# Extract the transcribed text from the complex JSON structure
transcribed_text = asr_result['pipelineResponse'][0]['output'][0]['source']
return {"text": transcribed_text}
except KeyError:
raise HTTPException(status_code=500, detail=f"Unexpected response structure from ASR service out:{asr_result}")
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
|