46 lines
1.3 KiB
Python
46 lines
1.3 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Optional
|
|
|
|
import numpy as np
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class TTSService:
|
|
def __init__(self) -> None:
|
|
self._pipeline = None
|
|
self._ready = False
|
|
self._attempted = False
|
|
self._init_error: Optional[str] = None
|
|
|
|
def _ensure_loaded(self) -> None:
|
|
if self._ready or self._attempted:
|
|
return
|
|
self._attempted = True
|
|
try:
|
|
import kokoro
|
|
|
|
self._pipeline = kokoro.KPipeline(lang_code="z")
|
|
self._ready = True
|
|
except Exception as exc: # pragma: no cover
|
|
self._init_error = str(exc)
|
|
logger.warning("TTS fallback mode: %s", exc)
|
|
|
|
async def synthesize(self, text: str) -> tuple[np.ndarray, int]:
|
|
self._ensure_loaded()
|
|
if self._ready and self._pipeline is not None:
|
|
chunks = []
|
|
for _, _, audio in self._pipeline(text, voice="zf_xiaoxiao"):
|
|
chunks.append(audio)
|
|
if chunks:
|
|
return np.concatenate(chunks), 24000
|
|
|
|
# 0.5s silence fallback for flow verification
|
|
return np.zeros(12000, dtype=np.float32), 24000
|
|
|
|
@property
|
|
def health(self) -> dict:
|
|
return {"ready": self._ready, "attempted": self._attempted, "error": self._init_error}
|