Spaces:
Sleeping
Sleeping
File size: 1,558 Bytes
2e818da | 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 44 45 46 47 48 49 50 51 52 | 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)
|