TextToSpeech

class TextToSpeech(ServiceWithStartup)

def __init__(
    self,
    tts_instance: str = TTS_SERVER,
    recorder: Recorder | None = None,
    get_time: Callable[[], float] | None = None,
    voice: str | None = None,
)

voice: str | None

received_samples: int

received_samples_yielded: int

async def send(self, message: str | TTSClientMessage) -> None

async def start_up()

async def shutdown()

def state(self) -> WebsocketState

async def __aiter__(self) -> AsyncIterator[TTSMessage]

async for message in tts:
    if isinstance(message, TTSAudioMessage):
        # Audio data as list of float32 PCM samples
        audio_array = np.array(message.pcm, dtype=np.float32)
        # Play or process audio...
    elif isinstance(message, TTSTextMessage):
        print(f"[{message.start_s:.2f}s - {message.stop_s:.2f}s] {message.text}")

class TTSClientTextMessage(BaseModel):
    type: Literal["Text"] = "Text"
    text: str

class TTSClientVoiceMessage(BaseModel):
    type: Literal["Voice"] = "Voice"
    embeddings: list[float]
    shape: list[int]

class TTSClientEosMessage(BaseModel):
    type: Literal["Eos"] = "Eos"

class TTSAudioMessage(BaseModel):
    type: Literal["Audio"]
    pcm: list[float]

class TTSTextMessage(BaseModel):
    type: Literal["Text"]
    text: str
    start_s: float
    stop_s: float

class TTSErrorMessage(BaseModel):
    type: Literal["Error"]
    message: str

class TTSReadyMessage(BaseModel):
    type: Literal["Ready"]

def prepare_text_for_tts(text: str) -> str

text = prepare_text_for_tts("What's *this* thing?")
# Result: "What's this thing?"

import asyncio
from unmute.tts.text_to_speech import TextToSpeech, TTSClientEosMessage

async def synthesize_speech():
    tts = TextToSpeech(voice="alloy")
    
    try:
        await tts.start_up()
        print(f"TTS state: {tts.state()}")
        
        # Send text for synthesis
        await tts.send("Hello, world!")
        await tts.send("This is a test of the text to speech system.")
        await tts.send(TTSClientEosMessage())  # Signal end of input
        
        # Receive synthesized audio
        async for message in tts:
            if isinstance(message, TTSAudioMessage):
                print(f"Received {len(message.pcm)} audio samples")
            elif isinstance(message, TTSTextMessage):
                print(f"Text timing: '{message.text}' from {message.start_s}s to {message.stop_s}s")
    
    finally:
        await tts.shutdown()

asyncio.run(synthesize_speech())

async def stream_synthesis():
    tts = TextToSpeech()
    await tts.start_up()
    
    async def send_text():
        """Send text word by word"""
        words = "The quick brown fox jumps over the lazy dog".split()
        for word in words:
            await tts.send(word + " ")
            await asyncio.sleep(0.1)  # Simulate streaming
        await tts.send(TTSClientEosMessage())
    
    async def receive_audio():
        """Receive and process audio"""
        async for message in tts:
            if isinstance(message, TTSAudioMessage):
                # Process audio in real-time
                pass
    
    await asyncio.gather(send_text(), receive_audio())
    await tts.shutdown()

async def use_custom_voice():
    # Custom voice embeddings must be pre-loaded in voice_embeddings_cache
    tts = TextToSpeech(voice="custom:my_voice")
    await tts.start_up()
    
    await tts.send("Speaking with a custom voice.")
    await tts.send(TTSClientEosMessage())
    
    async for message in tts:
        pass  # Process messages
    
    await tts.shutdown()

class TtsStreamingQuery(BaseModel):
    seed: int | None = None
    temperature: float | None = None
    top_k: int | None = None
    format: str = "PcmMessagePack"
    voice: str | None = None
    voices: list[str] | None = None
    max_seq_len: int | None = None
    cfg_alpha: float | None = None  # Default: 1.5 in code
    auth_id: str | None = None