TTS Integrations

browser_audio:
  # ... other configuration
  tts:
    name: azure

recipe: default.v1
language: en
additional_languages:
  - de-DE

pipeline:
- name: SingleStepLLMCommandGenerator

policies:
- name: FlowPolicy

assistant_id: 20240418-073244-narrow-archive

browser_audio:
  tts:
    name: azure
    speech_region: germanywestcentral
    language_map:
      en:
        language: en-US
        voice: en-US-JennyMultilingualNeural
      de-DE:
        language: de-DE
        voice: de-DE-KatjaNeural

browser_audio:
  # ... other configuration
  tts:
    name: cartesia

recipe: default.v1
language: en
additional_languages:
  - de-DE

pipeline:
- name: SingleStepLLMCommandGenerator

policies:
- name: FlowPolicy

assistant_id: 20240418-073244-narrow-archive

browser_audio:
  tts:
    name: cartesia
    language_map:
      en:
        language: en
      de-DE:
        language: de

browser_audio:
  # ... other configuration
  tts:
    name: deepgram

recipe: default.v1
language: en
additional_languages:
  - de-DE

pipeline:
- name: SingleStepLLMCommandGenerator

policies:
- name: FlowPolicy

assistant_id: 20240418-073244-narrow-archive

browser_audio:
  tts:
    name: deepgram
    language_map:
      en:
        model: aura-2-thalia-en
      de-DE:
        model: aura-2-julius-de

browser_audio:
  tts:
    name: deepgram_flux
    language_map:
      en:
        model: flux-alexis-en

browser_audio:
  tts:
    name: elevenlabs
    language_map:
      en:
        voice: "<ElevenLabs voice ID>"

browser_audio:
  # ... other configuration
  tts:
    name: rime

recipe: default.v1
language: en
additional_languages:
  - de-DE

pipeline:
- name: SingleStepLLMCommandGenerator

policies:
- name: FlowPolicy

assistant_id: 20240418-073244-narrow-archive

browser_audio:
  tts:
    name: rime
    language_map:
      en:
        language: eng
        voice: abbie
      de-DE:
        language: ger
        voice: amalia

import os
from dataclasses import dataclass
from typing import AsyncIterator, Dict, Optional
from urllib.parse import urlencode

import aiohttp
from aiohttp import ClientTimeout, WSMsgType

from rasa.core.channels.voice_stream.audio_bytes import RasaAudioBytes
from rasa.core.channels.voice_stream.tts.tts_engine import (
    TTSEngine,
    TTSEngineConfig,
    TTSError,
)

@dataclass
class MyTTSConfig(TTSEngineConfig):
    endpoint: str = "wss://api.example.com/v1/speak"
    model_id: str = "en-US-standard"

class MyTTS(TTSEngine[MyTTSConfig]):
    session: Optional[aiohttp.ClientSession] = None
    required_env_vars = ("MY_TTS_API_KEY",)  # Optional: required environment variables
    required_packages = ("aiohttp",)  # Optional: required Python packages
    streaming_input = True
    ws: Optional[aiohttp.ClientWebSocketResponse] = None

    def __init__(self, config: Optional[MyTTSConfig] = None):
        super().__init__(config)
        timeout = ClientTimeout(total=self.config.timeout)
        # All class instances share the same session
        if self.__class__.session is None or self.__class__.session.closed:
            self.__class__.session = aiohttp.ClientSession(timeout=timeout)

    async def connect(self, config: Optional[MyTTSConfig] = None) -> None:
        """Establish WebSocket connection to the TTS engine."""
        headers = {
            "Authorization": f"Bearer {os.environ['MY_TTS_API_KEY']}",
        }
        
        query_params = {
            "model": merged_config.model_id,
            "language": merged_config.language,
            "voice": merged_config.voice,
            "encoding": "mulaw",
            "sample_rate": "8000",
        }
        
        ws_url = f"{merged_config.endpoint}?{urlencode(query_params)}"
        
        try:
            self.ws = await self.session.ws_connect(
                ws_url,
                headers=headers,
                timeout=float(self.config.timeout) if self.config.timeout else 30,
            )
        except Exception as e:
            raise TTSError(f"Failed to connect to TTS service: {e}")

    async def close_connection(self) -> None:
        """Close WebSocket connection if it exists."""
        if self.ws and not self.ws.closed:
            await self.ws.close()
            self.ws = None

    async def send_text_chunk(self, text: str) -> None:
        """Send text chunk to TTS engine for streaming synthesis."""
        await self.ws.send_json({"text": text})

    async def signal_text_done(self) -> None:
        """Signal TTS engine that all text has been sent."""
        await self.ws.send_json({"operation": "flush"})

    async def stream_audio(self) -> AsyncIterator[RasaAudioBytes]:
        """Stream audio output from the TTS engine."""
        try:
            async for msg in self.ws:
                if msg.type == WSMsgType.TEXT:
                    data = msg.json()
                    if data.get("type") == "audio":
                        # Assume base64 encoded audio data
                        import base64
                        audio_bytes = base64.b64decode(data.get("data", ""))
                        if audio_bytes:
                            yield self.engine_bytes_to_rasa_audio_bytes(audio_bytes)
                    elif data.get("type") == "error":
                        raise TTSError(f"TTS error: {data.get('message')}")
                elif msg.type == WSMsgType.CLOSED:
                    break
                elif msg.type == WSMsgType.ERROR:
                    raise TTSError("WebSocket error during audio streaming")
        except Exception as e:
            raise TTSError(f"Error during audio streaming: {e}")

    async def synthesize(
        self, text: str, config: Optional[MyTTSConfig] = None
    ) -> AsyncIterator[RasaAudioBytes]:
        """Generate speech from text using streaming WebSocket."""
        await self.connect(config)
        
        try:
            await self.send_text_chunk(text)
            await self.signal_text_done()
            
            async for audio_chunk in self.stream_audio():
                yield audio_chunk
        finally:
            await self.close_connection()

    def engine_bytes_to_rasa_audio_bytes(self, chunk: bytes) -> RasaAudioBytes:
        """Convert the generated TTS audio bytes into Rasa audio bytes."""
        # If your service returns audio in mulaw format already, return directly
        return RasaAudioBytes(chunk)
        
        # If your service returns audio in a different format (e.g., PCM linear),
        # you'll need to convert it. For example, to convert from linear PCM:
        # import audioop
        # mulaw_data = audioop.lin2ulaw(chunk, 2)  # 2 = 16-bit samples
        # return RasaAudioBytes(mulaw_data)

    @staticmethod
    def get_default_config() -> MyTTSConfig:
        return MyTTSConfig(
            endpoint="wss://api.example.com/v1/speak",
            model_id="en-US-standard",
            language="en-US",
            voice="female-1",
            timeout=30,
        )

    @classmethod
    def from_config_dict(cls, config: Dict) -> "MyTTS":
        return cls(MyTTSConfig.from_dict(config))

browser_audio:
  # ... other configuration
  tts:
    name: addons.custom_tts.MyTTS
    api_key: "your_api_key"
    endpoint: "wss://api.example.com/v1/speak"
    language: "en-US"
    voice: "en-US-JennyNeural"
    timeout: 30
    # any other custom parameters your TTS needs