Text-to-Speech (TTS)

import asyncio
import os

import aiohttp
from livekit.agents import (
    DEFAULT_API_CONNECT_OPTIONS,
    APIConnectionError,
    APIConnectOptions,
    APIStatusError,
    APITimeoutError,
    tts,
    utils,
)

TELNYX_TTS_URL = "https://api.telnyx.com/v2/text-to-speech/speech"


class UltraTTS(tts.TTS):
    """Telnyx Ultra over the TTS REST API. LiveKit sends one request per sentence."""

    def __init__(self, voice: str = "Telnyx.Ultra.01eaafa9-308a-4276-a017-6ab0cf061b1f") -> None:
        super().__init__(
            capabilities=tts.TTSCapabilities(streaming=False),
            sample_rate=24000,
            num_channels=1,
        )
        self.voice = voice
        self.api_key = os.environ["TELNYX_API_KEY"]
        self.session: aiohttp.ClientSession | None = None

    def synthesize(
        self, text: str, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS
    ) -> tts.ChunkedStream:
        return UltraChunkedStream(tts=self, input_text=text, conn_options=conn_options)

    async def aclose(self) -> None:
        if self.session:
            await self.session.close()


class UltraChunkedStream(tts.ChunkedStream):
    async def _run(self, output_emitter: tts.AudioEmitter) -> None:
        ultra: UltraTTS = self._tts
        ultra.session = ultra.session or aiohttp.ClientSession()
        try:
            async with ultra.session.post(
                TELNYX_TTS_URL,
                headers={"Authorization": f"Bearer {ultra.api_key}"},
                json={"text": self.input_text, "voice": ultra.voice},
                timeout=aiohttp.ClientTimeout(total=30, sock_connect=self._conn_options.timeout),
            ) as resp:
                resp.raise_for_status()
                output_emitter.initialize(
                    request_id=utils.shortuuid(),
                    sample_rate=ultra.sample_rate,
                    num_channels=ultra.num_channels,
                    mime_type="audio/mpeg",
                )
                async for chunk in resp.content.iter_chunked(4096):
                    output_emitter.push(chunk)
        # Convert aiohttp errors so LiveKit retries transient failures (timeouts, 429, 5xx).
        except asyncio.TimeoutError:
            raise APITimeoutError() from None
        except aiohttp.ClientResponseError as e:
            raise APIStatusError(e.message, status_code=e.status) from None
        except aiohttp.ClientError as e:
            raise APIConnectionError() from e