Spaces:
Sleeping
Sleeping
File size: 1,777 Bytes
0925810 c2fa877 0925810 c2fa877 0925810 c2fa877 0925810 c2fa877 0925810 c2fa877 0925810 |
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 |
import typing as t
from dotenv import load_dotenv
from elevenlabs.client import AsyncElevenLabs, ElevenLabs
from elevenlabs import VoiceSettings
load_dotenv()
from src.config import logger, ELEVENLABS_API_KEY
ELEVEN_CLIENT = ElevenLabs(api_key=ELEVENLABS_API_KEY)
ELEVEN_CLIENT_ASYNC = AsyncElevenLabs(api_key=ELEVENLABS_API_KEY)
def tts_stream(voice_id: str, text: str) -> t.Iterator[bytes]:
async_iter = ELEVEN_CLIENT.text_to_speech.convert(voice_id=voice_id, text=text)
for chunk in async_iter:
if chunk:
yield chunk
def tts(voice_id: str, text: str):
tts_iter = tts_stream(voice_id=voice_id, text=text)
combined = b"".join(tts_iter)
return combined
async def tts_astream(
voice_id: str, text: str, params: dict | None = None
) -> t.AsyncIterator[bytes]:
params_all = dict(voice_id=voice_id, text=text)
if params is not None:
params_all["voice_settings"] = VoiceSettings( # type: ignore
stability=params.get("stability"),
similarity_boost=params.get("similarity_boost"),
style=params.get("style"),
)
logger.info(f"call to 11labs TTS endpoint with params: {params_all}")
async_iter = ELEVEN_CLIENT_ASYNC.text_to_speech.convert(**params_all)
async for chunk in async_iter:
if chunk:
yield chunk
async def sound_generation_astream(
sound_generation_data: dict,
) -> t.AsyncIterator[bytes]:
async_iter = ELEVEN_CLIENT_ASYNC.text_to_sound_effects.convert(
text=sound_generation_data["text"],
duration_seconds=sound_generation_data["duration_seconds"],
prompt_influence=sound_generation_data["prompt_influence"],
)
async for chunk in async_iter:
if chunk:
yield chunk
|