437 lines
18 KiB
Python
437 lines
18 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import audioop
|
|
import base64
|
|
import json
|
|
import logging
|
|
import os
|
|
import time
|
|
from collections.abc import AsyncGenerator
|
|
from collections.abc import AsyncIterable
|
|
from urllib.parse import quote
|
|
from urllib.parse import urlencode
|
|
|
|
from realtime_voice_service.providers.base import BaseTTS
|
|
|
|
|
|
LOGGER = logging.getLogger("uvicorn.error")
|
|
|
|
|
|
def _api_base() -> str:
|
|
return (os.getenv("ELEVENLABS_API_BASE", "https://api.elevenlabs.io").strip() or "https://api.elevenlabs.io").rstrip("/")
|
|
|
|
|
|
def _ws_base(http_base: str) -> str:
|
|
normalized = http_base.rstrip("/")
|
|
if normalized.startswith("https://"):
|
|
return f"wss://{normalized[len('https://') :]}"
|
|
if normalized.startswith("http://"):
|
|
return f"ws://{normalized[len('http://') :]}"
|
|
return normalized
|
|
|
|
|
|
def _timeout_seconds() -> float:
|
|
raw = os.getenv("ELEVENLABS_TIMEOUT_SECONDS")
|
|
if raw is None:
|
|
return 30.0
|
|
try:
|
|
return max(float(raw.strip()), 1.0)
|
|
except ValueError:
|
|
return 30.0
|
|
|
|
|
|
def _parse_chunk_schedule(raw: str | None) -> list[int]:
|
|
if not raw:
|
|
return [80, 120, 160, 220]
|
|
values: list[int] = []
|
|
for part in raw.split(","):
|
|
try:
|
|
values.append(max(int(part.strip()), 20))
|
|
except ValueError:
|
|
continue
|
|
return values or [80, 120, 160, 220]
|
|
|
|
|
|
def _sample_rate_from_pcm_format(output_format: str) -> int | None:
|
|
normalized = str(output_format or "").strip().lower()
|
|
if not normalized.startswith("pcm_"):
|
|
return None
|
|
try:
|
|
return int(normalized.split("_", 1)[1])
|
|
except (IndexError, ValueError):
|
|
return None
|
|
|
|
|
|
def _preview_text(text: str, *, limit: int = 120) -> str:
|
|
normalized = " ".join(str(text or "").split())
|
|
if len(normalized) <= limit:
|
|
return normalized
|
|
return f"{normalized[:limit]}..."
|
|
|
|
|
|
class ElevenLabsTTS(BaseTTS):
|
|
def __init__(
|
|
self,
|
|
*,
|
|
api_key: str | None = None,
|
|
api_base: str | None = None,
|
|
voice_id: str | None = None,
|
|
model_id: str | None = None,
|
|
language_code: str | None = None,
|
|
output_format: str | None = None,
|
|
target_sample_rate_hz: int | None = None,
|
|
inactivity_timeout_seconds: int = 20,
|
|
timeout_seconds: float | None = None,
|
|
auto_mode: bool | None = None,
|
|
chunk_length_schedule: list[int] | None = None,
|
|
) -> None:
|
|
self._api_key = str(api_key if api_key is not None else os.getenv("ELEVENLABS_API_KEY", "")).strip()
|
|
self._api_base = str(api_base or _api_base()).strip().rstrip("/")
|
|
self._ws_base = _ws_base(self._api_base)
|
|
self._voice_id = str(voice_id or os.getenv("ELEVENLABS_TTS_VOICE_ID", "")).strip()
|
|
self._requested_model_id = (
|
|
str(model_id or os.getenv("ELEVENLABS_TTS_MODEL_ID", "eleven_multilingual_v2")).strip()
|
|
or "eleven_multilingual_v2"
|
|
)
|
|
self._websocket_fallback_model_id = (
|
|
str(os.getenv("ELEVENLABS_TTS_WS_FALLBACK_MODEL_ID", "eleven_multilingual_v2")).strip()
|
|
or "eleven_multilingual_v2"
|
|
)
|
|
self._model_id = self._resolve_websocket_model_id(
|
|
requested_model_id=self._requested_model_id,
|
|
fallback_model_id=self._websocket_fallback_model_id,
|
|
)
|
|
self._language_code = str(language_code or os.getenv("ELEVENLABS_TTS_LANGUAGE_CODE", "ru")).strip() or None
|
|
requested_output_format = (
|
|
str(output_format or os.getenv("ELEVENLABS_TTS_OUTPUT_FORMAT", "pcm_16000")).strip().lower()
|
|
or "pcm_16000"
|
|
)
|
|
self._target_sample_rate_hz = int(
|
|
target_sample_rate_hz
|
|
if target_sample_rate_hz is not None
|
|
else (_sample_rate_from_pcm_format(requested_output_format) or 16000)
|
|
)
|
|
self._output_format = self._resolve_provider_output_format(
|
|
requested_output_format=requested_output_format,
|
|
target_sample_rate_hz=self._target_sample_rate_hz,
|
|
)
|
|
self._provider_sample_rate_hz = _sample_rate_from_pcm_format(self._output_format) or self._target_sample_rate_hz
|
|
self._timeout_seconds = max(float(timeout_seconds if timeout_seconds is not None else _timeout_seconds()), 1.0)
|
|
self._inactivity_timeout_seconds = max(int(inactivity_timeout_seconds), 5)
|
|
self._auto_mode = (
|
|
str(os.getenv("ELEVENLABS_TTS_AUTO_MODE", "1")).strip().lower() in {"1", "true", "yes", "on"}
|
|
if auto_mode is None
|
|
else bool(auto_mode)
|
|
)
|
|
self._chunk_length_schedule = list(
|
|
chunk_length_schedule
|
|
or _parse_chunk_schedule(os.getenv("ELEVENLABS_TTS_CHUNK_LENGTH_SCHEDULE"))
|
|
)
|
|
self._voice_settings = {
|
|
"stability": self._read_float_env("ELEVENLABS_TTS_STABILITY", 0.5),
|
|
"similarity_boost": self._read_float_env("ELEVENLABS_TTS_SIMILARITY_BOOST", 0.75),
|
|
"speed": self._read_float_env("ELEVENLABS_TTS_SPEED", 1.0),
|
|
"use_speaker_boost": self._read_bool_env("ELEVENLABS_TTS_USE_SPEAKER_BOOST", False),
|
|
}
|
|
LOGGER.info(
|
|
"ElevenLabs TTS config: voice_id=%s requested_model=%s websocket_model=%s "
|
|
"provider_output_format=%s provider_sample_rate=%s target_sample_rate=%s "
|
|
"language=%s auto_mode=%s chunk_schedule=%s voice_settings=%s",
|
|
self._voice_id,
|
|
self._requested_model_id,
|
|
self._model_id,
|
|
self._output_format,
|
|
self._provider_sample_rate_hz,
|
|
self._target_sample_rate_hz,
|
|
self._language_code,
|
|
self._auto_mode,
|
|
self._chunk_length_schedule,
|
|
self._voice_settings,
|
|
)
|
|
|
|
async def synthesize_stream(self, text_stream: AsyncIterable[str]) -> AsyncGenerator[bytes, None]:
|
|
if not self._api_key:
|
|
raise RuntimeError("ELEVENLABS_API_KEY is required for ElevenLabs TTS")
|
|
if not self._voice_id:
|
|
raise RuntimeError("ELEVENLABS_TTS_VOICE_ID is required for ElevenLabs TTS")
|
|
|
|
try:
|
|
from websockets.exceptions import WebSocketException
|
|
from websockets.legacy.client import connect as websocket_connect
|
|
except Exception as exc: # noqa: BLE001
|
|
raise RuntimeError("The `websockets` package is required for ElevenLabs TTS WebSocket streaming") from exc
|
|
|
|
websocket_url = self._build_websocket_url()
|
|
started_monotonic = time.perf_counter()
|
|
audio_chunk_count = 0
|
|
audio_byte_count = 0
|
|
yielded_byte_count = 0
|
|
resample_state = None
|
|
pcm_remainder = b""
|
|
LOGGER.info(
|
|
"ElevenLabs TTS WebSocket connecting: voice_id=%s model=%s output_format=%s "
|
|
"provider_sample_rate=%s target_sample_rate=%s language=%s auto_mode=%s",
|
|
self._voice_id,
|
|
self._model_id,
|
|
self._output_format,
|
|
self._provider_sample_rate_hz,
|
|
self._target_sample_rate_hz,
|
|
self._language_code,
|
|
self._auto_mode,
|
|
)
|
|
try:
|
|
async with websocket_connect(
|
|
websocket_url,
|
|
extra_headers=[("xi-api-key", self._api_key)],
|
|
open_timeout=min(self._timeout_seconds, 10.0),
|
|
close_timeout=1.0,
|
|
ping_interval=20.0,
|
|
ping_timeout=20.0,
|
|
max_size=None,
|
|
) as websocket:
|
|
LOGGER.info(
|
|
"ElevenLabs TTS WebSocket connected: voice_id=%s model=%s connect_ms=%s",
|
|
self._voice_id,
|
|
self._model_id,
|
|
int((time.perf_counter() - started_monotonic) * 1000.0),
|
|
)
|
|
await websocket.send(json.dumps(self._initial_payload()))
|
|
sender_task = asyncio.create_task(
|
|
self._send_text_chunks(websocket, text_stream),
|
|
name="elevenlabs-tts-sender",
|
|
)
|
|
try:
|
|
while True:
|
|
raw_message = await asyncio.wait_for(websocket.recv(), timeout=self._timeout_seconds)
|
|
if isinstance(raw_message, bytes):
|
|
raw_message = raw_message.decode("utf-8")
|
|
payload = json.loads(raw_message)
|
|
if not isinstance(payload, dict):
|
|
continue
|
|
if payload.get("audio"):
|
|
audio_chunk = base64.b64decode(str(payload["audio"]))
|
|
audio_chunk_count += 1
|
|
audio_byte_count += len(audio_chunk)
|
|
audio_chunk, resample_state, pcm_remainder = self._normalize_pcm_chunk(
|
|
audio_chunk,
|
|
resample_state=resample_state,
|
|
pcm_remainder=pcm_remainder,
|
|
)
|
|
yielded_byte_count += len(audio_chunk)
|
|
if audio_chunk_count == 1:
|
|
LOGGER.info(
|
|
"ElevenLabs TTS first audio: voice_id=%s model=%s ttfa_ms=%s "
|
|
"provider_bytes=%s yielded_bytes=%s",
|
|
self._voice_id,
|
|
self._model_id,
|
|
int((time.perf_counter() - started_monotonic) * 1000.0),
|
|
audio_byte_count,
|
|
len(audio_chunk),
|
|
)
|
|
elif audio_chunk_count % 10 == 0:
|
|
LOGGER.info(
|
|
"ElevenLabs TTS audio summary: voice_id=%s chunks=%s provider_bytes=%s yielded_bytes=%s",
|
|
self._voice_id,
|
|
audio_chunk_count,
|
|
audio_byte_count,
|
|
yielded_byte_count,
|
|
)
|
|
if audio_chunk:
|
|
yield audio_chunk
|
|
if self._is_error_payload(payload):
|
|
LOGGER.error(
|
|
"ElevenLabs TTS error payload: voice_id=%s model=%s payload=%s",
|
|
self._voice_id,
|
|
self._model_id,
|
|
self._format_error_payload(payload),
|
|
)
|
|
raise RuntimeError(self._format_error_payload(payload))
|
|
if bool(payload.get("isFinal")) or bool(payload.get("is_final")):
|
|
LOGGER.info(
|
|
"ElevenLabs TTS final payload: voice_id=%s model=%s chunks=%s "
|
|
"provider_bytes=%s yielded_bytes=%s total_ms=%s",
|
|
self._voice_id,
|
|
self._model_id,
|
|
audio_chunk_count,
|
|
audio_byte_count,
|
|
yielded_byte_count,
|
|
int((time.perf_counter() - started_monotonic) * 1000.0),
|
|
)
|
|
if sender_task.done():
|
|
break
|
|
finally:
|
|
if not sender_task.done():
|
|
sender_task.cancel()
|
|
try:
|
|
await sender_task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
except asyncio.TimeoutError as exc:
|
|
raise RuntimeError("ElevenLabs TTS WebSocket request timed out") from exc
|
|
except WebSocketException as exc:
|
|
raise RuntimeError("ElevenLabs TTS WebSocket stream failed") from exc
|
|
except OSError as exc:
|
|
raise RuntimeError("ElevenLabs TTS WebSocket connection failed") from exc
|
|
|
|
def _build_websocket_url(self) -> str:
|
|
query = {
|
|
"model_id": self._model_id,
|
|
"output_format": self._output_format,
|
|
"inactivity_timeout": self._inactivity_timeout_seconds,
|
|
"auto_mode": str(self._auto_mode).lower(),
|
|
"sync_alignment": "false",
|
|
"apply_text_normalization": "auto",
|
|
}
|
|
if self._language_code:
|
|
query["language_code"] = self._language_code
|
|
encoded_voice_id = quote(self._voice_id, safe="")
|
|
return f"{self._ws_base}/v1/text-to-speech/{encoded_voice_id}/stream-input?{urlencode(query)}"
|
|
|
|
def _initial_payload(self) -> dict[str, object]:
|
|
payload: dict[str, object] = {
|
|
"text": " ",
|
|
"xi_api_key": self._api_key,
|
|
"voice_settings": self._voice_settings,
|
|
}
|
|
if not self._auto_mode:
|
|
payload["generation_config"] = {
|
|
"chunk_length_schedule": self._chunk_length_schedule,
|
|
}
|
|
return payload
|
|
|
|
async def _send_text_chunks(self, websocket, text_stream: AsyncIterable[str]) -> None:
|
|
pending_chunk: str | None = None
|
|
sent_count = 0
|
|
sent_chars = 0
|
|
async for raw_chunk in text_stream:
|
|
normalized = str(raw_chunk).strip()
|
|
if not normalized:
|
|
continue
|
|
normalized_chunk = self._normalize_stream_text(normalized)
|
|
if pending_chunk is not None:
|
|
await websocket.send(json.dumps({
|
|
"text": pending_chunk,
|
|
"try_trigger_generation": True,
|
|
}))
|
|
sent_count += 1
|
|
sent_chars += len(pending_chunk)
|
|
LOGGER.info(
|
|
"ElevenLabs TTS text chunk sent: index=%s chars=%s total_chars=%s preview=%r",
|
|
sent_count,
|
|
len(pending_chunk),
|
|
sent_chars,
|
|
_preview_text(pending_chunk),
|
|
)
|
|
pending_chunk = normalized_chunk
|
|
|
|
if pending_chunk is not None:
|
|
await websocket.send(json.dumps({
|
|
"text": pending_chunk,
|
|
"try_trigger_generation": True,
|
|
"flush": True,
|
|
}))
|
|
sent_count += 1
|
|
sent_chars += len(pending_chunk)
|
|
LOGGER.info(
|
|
"ElevenLabs TTS final text chunk sent: index=%s chars=%s total_chars=%s preview=%r",
|
|
sent_count,
|
|
len(pending_chunk),
|
|
sent_chars,
|
|
_preview_text(pending_chunk),
|
|
)
|
|
|
|
await websocket.send(json.dumps({"text": ""}))
|
|
LOGGER.info("ElevenLabs TTS text stream closed: chunks=%s chars=%s", sent_count, sent_chars)
|
|
|
|
@staticmethod
|
|
def _normalize_stream_text(text: str) -> str:
|
|
if text.endswith((" ", "\n", "\t")):
|
|
return text
|
|
return f"{text} "
|
|
|
|
@staticmethod
|
|
def _resolve_websocket_model_id(*, requested_model_id: str, fallback_model_id: str) -> str:
|
|
normalized_requested = requested_model_id.strip() or "eleven_multilingual_v2"
|
|
if normalized_requested not in {"eleven_v3", "eleven_ttv_v3"}:
|
|
return normalized_requested
|
|
|
|
normalized_fallback = fallback_model_id.strip() or "eleven_multilingual_v2"
|
|
LOGGER.warning(
|
|
"ElevenLabs WebSocket TTS does not support model_id=%s; falling back to model_id=%s",
|
|
normalized_requested,
|
|
normalized_fallback,
|
|
)
|
|
return normalized_fallback
|
|
|
|
@staticmethod
|
|
def _resolve_provider_output_format(*, requested_output_format: str, target_sample_rate_hz: int) -> str:
|
|
normalized = requested_output_format.strip().lower() or "pcm_16000"
|
|
if normalized == "pcm_8000":
|
|
LOGGER.warning(
|
|
"ElevenLabs TTS does not provide reliable PCM16 8kHz streaming; requesting pcm_16000 "
|
|
"and resampling to %sHz locally",
|
|
target_sample_rate_hz,
|
|
)
|
|
return "pcm_16000"
|
|
return normalized
|
|
|
|
def _normalize_pcm_chunk(
|
|
self,
|
|
audio_chunk: bytes,
|
|
*,
|
|
resample_state,
|
|
pcm_remainder: bytes,
|
|
) -> tuple[bytes, object, bytes]:
|
|
if not audio_chunk:
|
|
return b"", resample_state, pcm_remainder
|
|
if self._provider_sample_rate_hz == self._target_sample_rate_hz:
|
|
return audio_chunk, resample_state, pcm_remainder
|
|
|
|
payload = pcm_remainder + audio_chunk
|
|
if len(payload) % 2:
|
|
pcm_remainder = payload[-1:]
|
|
payload = payload[:-1]
|
|
else:
|
|
pcm_remainder = b""
|
|
if not payload:
|
|
return b"", resample_state, pcm_remainder
|
|
converted, resample_state = audioop.ratecv(
|
|
payload,
|
|
2,
|
|
1,
|
|
self._provider_sample_rate_hz,
|
|
self._target_sample_rate_hz,
|
|
resample_state,
|
|
)
|
|
return converted, resample_state, pcm_remainder
|
|
|
|
@staticmethod
|
|
def _is_error_payload(payload: dict[str, object]) -> bool:
|
|
message_type = str(payload.get("message_type") or payload.get("type") or "").strip().lower()
|
|
return message_type.endswith("error") or "error" in payload
|
|
|
|
@staticmethod
|
|
def _format_error_payload(payload: dict[str, object]) -> str:
|
|
detail = str(payload.get("message") or payload.get("detail") or payload.get("error") or "").strip()
|
|
if detail:
|
|
return f"ElevenLabs TTS WebSocket error: {detail}"
|
|
return "ElevenLabs TTS WebSocket error"
|
|
|
|
@staticmethod
|
|
def _read_bool_env(name: str, default: bool) -> bool:
|
|
raw = os.getenv(name)
|
|
if raw is None:
|
|
return default
|
|
return str(raw).strip().lower() in {"1", "true", "yes", "on"}
|
|
|
|
@staticmethod
|
|
def _read_float_env(name: str, default: float) -> float:
|
|
raw = os.getenv(name)
|
|
if raw is None:
|
|
return default
|
|
try:
|
|
return float(str(raw).strip())
|
|
except ValueError:
|
|
return default
|