File size: 1,317 Bytes
87337b1 |
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 |
#
#
# Agora Real Time Engagement
# Created by XinHui Li in 2024.
# Copyright (c) 2024 Agora IO. All rights reserved.
#
#
from dataclasses import dataclass
from typing import AsyncIterator, Optional, Literal
from pyneuphonic import Neuphonic, TTSConfig
from pyneuphonic.models import APIResponse, TTSResponse
from ten_ai_base.config import BaseConfig
@dataclass
class NeuphonicTTSConfig(BaseConfig):
api_key: str = ""
lang_code: str = "en"
sample_rate: int = 16000
voice_id: Optional[str] = None
speed: float = 1.0
encoding: Literal["pcm_linear", "pcm_mulaw"] = "pcm_linear"
request_timeout_seconds: int = 10
class NeuphonicTTS:
def __init__(self, config: NeuphonicTTSConfig) -> None:
self.config = config
self.client = Neuphonic(api_key=config.api_key)
def text_to_speech_stream(self, text: str) -> AsyncIterator[APIResponse[TTSResponse]]:
sse_client = self.client.tts.AsyncSSEClient(timeout=self.config.request_timeout_seconds)
tts_config = TTSConfig(
lang_code=self.config.lang_code,
sampling_rate=self.config.sample_rate,
voice_id=self.config.voice_id,
speed=self.config.speed,
encoding=self.config.encoding,
)
return sse_client.send(text, tts_config=tts_config)
|