app/services/chat_tts_stream.py

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any, Callable

from app.services.tts_service import TTSGenerationManager
from app.services.tts_text import normalize_for_speech


@dataclass
class ChatTTSStream:
    manager: TTSGenerationManager
    generation_id: str | None
    text_parts: list[str] = field(default_factory=list)

    @classmethod
    async def create(
        cls,
        *,
        manager: TTSGenerationManager,
        project_id: str,
        message_id: str,
        send: Callable[[str, dict[str, Any]], Any],
        input_mode: str,
        auto_read_reply: bool,
        model_id: str = "pocket-tts",
        voice_id: str = "anshuman-normal-custom",
    ) -> "ChatTTSStream":
        generation_id = None
        if input_mode == "voice" and auto_read_reply:
            generation_id = await manager.begin(
                project_id,
                message_id,
                send,
                model_id=model_id,
                voice_id=voice_id,
            )
        return cls(manager=manager, generation_id=generation_id)

    async def feed(self, token: str) -> None:
        if not self.generation_id:
            return
        self.text_parts.append(token)

    async def finish(self) -> None:
        if not self.generation_id:
            return
        utterance = normalize_for_speech("".join(self.text_parts))
        if utterance:
            await self.manager.enqueue(self.generation_id, 0, utterance)
        await self.manager.finish(self.generation_id)