wehub-resource-sync
621 lines
22 KiB
Python
621 lines
22 KiB
Python
# Copyright 2025 LiveKit, Inc.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
|
|
"""Text-to-Speech implementation for Gnani Vachana
|
|
|
|
This module provides a TTS implementation that uses the Gnani Vachana API,
|
|
supporting three synthesis modes via ``synthesize()``:
|
|
- REST (RESTChunkedStream) — single-request batch synthesis
|
|
- SSE (SSEChunkedStream) — streaming via Server-Sent Events
|
|
- WebSocket (WebSocketChunkedStream) — lowest-latency via WebSocket
|
|
|
|
Additionally, ``stream()`` uses WebSocket (SynthesizeStream) for real-time
|
|
token-by-token streaming input.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import base64
|
|
import json
|
|
import os
|
|
from dataclasses import dataclass, replace
|
|
from typing import Any, Literal
|
|
|
|
import aiohttp
|
|
|
|
from livekit.agents import (
|
|
DEFAULT_API_CONNECT_OPTIONS,
|
|
APIConnectionError,
|
|
APIConnectOptions,
|
|
APIStatusError,
|
|
APITimeoutError,
|
|
tts,
|
|
utils,
|
|
)
|
|
|
|
from .log import logger
|
|
|
|
GNANI_TTS_BASE_URL = "https://api.vachana.ai"
|
|
|
|
GnaniTTSVoices = Literal[
|
|
"Karan",
|
|
"Simran",
|
|
"Nara",
|
|
"Riya",
|
|
"Viraj",
|
|
"Raju",
|
|
]
|
|
|
|
SUPPORTED_VOICES: set[str] = {"Karan", "Simran", "Nara", "Riya", "Viraj", "Raju"}
|
|
|
|
GnaniTTSEncodings = Literal["linear_pcm", "oggopus"]
|
|
GnaniTTSContainers = Literal["raw", "mp3", "wav", "mulaw", "ogg"]
|
|
GnaniTTSBitrates = Literal["96k", "128k", "192k"]
|
|
GnaniTTSSynthesizeMethod = Literal["rest", "sse", "websocket"]
|
|
|
|
SUPPORTED_SAMPLE_RATES = (8000, 16000, 22050, 44100)
|
|
|
|
_WAV_HEADER_SIZE = 44
|
|
|
|
|
|
@dataclass
|
|
class GnaniTTSOptions:
|
|
api_key: str
|
|
voice: str = "Karan"
|
|
model: str = "vachana-voice-v3"
|
|
sample_rate: int = 16000
|
|
encoding: str = "linear_pcm"
|
|
container: str = "wav"
|
|
num_channels: int = 1
|
|
sample_width: int = 2
|
|
bitrate: str | None = None
|
|
base_url: str = GNANI_TTS_BASE_URL
|
|
synthesize_method: str = "rest"
|
|
|
|
|
|
_DEPRECATED_TTS_KWARGS = frozenset(("language", "http_session"))
|
|
|
|
|
|
def _check_deprecated_tts_args(kwargs: dict[str, Any], *, caller: str = "TTS.__init__") -> None:
|
|
"""Warn about deprecated kwargs and raise on truly unknown ones."""
|
|
for name in _DEPRECATED_TTS_KWARGS:
|
|
if name in kwargs:
|
|
logger.warning(f"`{name}` is deprecated and no longer used")
|
|
|
|
unknown = set(kwargs) - _DEPRECATED_TTS_KWARGS
|
|
if unknown:
|
|
raise TypeError(
|
|
f"{caller}() got unexpected keyword argument(s): {', '.join(sorted(unknown))}"
|
|
)
|
|
|
|
|
|
class TTS(tts.TTS):
|
|
"""Gnani Vachana Text-to-Speech implementation.
|
|
|
|
Provides text-to-speech functionality using Gnani's Vachana platform.
|
|
Supports REST, SSE, and WebSocket synthesis modes.
|
|
|
|
Args:
|
|
voice: Voice to use for synthesis (Karan, Simran, Riya, etc.).
|
|
model: TTS model name (default: vachana-voice-v3).
|
|
sample_rate: Audio output sample rate (8000-44100).
|
|
encoding: Audio encoding (linear_pcm or oggopus).
|
|
container: Audio container format (raw, mp3, wav, mulaw, ogg).
|
|
api_key: Gnani API key (falls back to GNANI_API_KEY env var).
|
|
base_url: Vachana API base URL.
|
|
synthesize_method: Synthesis mode — "rest", "sse", or "websocket".
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
voice: GnaniTTSVoices | str = "Karan",
|
|
model: str = "vachana-voice-v3",
|
|
sample_rate: int = 16000,
|
|
num_channels: int = 1,
|
|
encoding: GnaniTTSEncodings | str = "linear_pcm",
|
|
container: GnaniTTSContainers | str = "wav",
|
|
bitrate: GnaniTTSBitrates | str | None = None,
|
|
api_key: str | None = None,
|
|
base_url: str = GNANI_TTS_BASE_URL,
|
|
synthesize_method: GnaniTTSSynthesizeMethod = "rest",
|
|
**kwargs: Any,
|
|
) -> None:
|
|
_check_deprecated_tts_args(kwargs)
|
|
|
|
if sample_rate not in SUPPORTED_SAMPLE_RATES:
|
|
raise ValueError(
|
|
f"sample_rate must be one of {SUPPORTED_SAMPLE_RATES}, got {sample_rate}"
|
|
)
|
|
|
|
super().__init__(
|
|
capabilities=tts.TTSCapabilities(streaming=True),
|
|
sample_rate=sample_rate,
|
|
num_channels=num_channels,
|
|
)
|
|
|
|
self._api_key = api_key or os.environ.get("GNANI_API_KEY")
|
|
if not self._api_key:
|
|
raise ValueError(
|
|
"Gnani API key is required. "
|
|
"Provide it directly or set GNANI_API_KEY environment variable."
|
|
)
|
|
|
|
if voice not in SUPPORTED_VOICES:
|
|
raise ValueError(
|
|
f"Voice '{voice}' not supported. "
|
|
f"Supported voices: {', '.join(sorted(SUPPORTED_VOICES))}"
|
|
)
|
|
|
|
self._opts = GnaniTTSOptions(
|
|
api_key=self._api_key,
|
|
voice=voice,
|
|
model=model,
|
|
sample_rate=sample_rate,
|
|
encoding=encoding,
|
|
container=container,
|
|
num_channels=num_channels,
|
|
bitrate=bitrate,
|
|
base_url=base_url,
|
|
synthesize_method=synthesize_method,
|
|
)
|
|
self._session: aiohttp.ClientSession | None = None
|
|
|
|
@property
|
|
def model(self) -> str:
|
|
return self._opts.model
|
|
|
|
@property
|
|
def provider(self) -> str:
|
|
return "Gnani"
|
|
|
|
def _ensure_session(self) -> aiohttp.ClientSession:
|
|
if not self._session:
|
|
self._session = utils.http_context.http_session()
|
|
return self._session
|
|
|
|
def synthesize(
|
|
self, text: str, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS
|
|
) -> tts.ChunkedStream:
|
|
if self._opts.synthesize_method == "sse":
|
|
return SSEChunkedStream(tts=self, input_text=text, conn_options=conn_options)
|
|
if self._opts.synthesize_method == "websocket":
|
|
return WebSocketChunkedStream(tts=self, input_text=text, conn_options=conn_options)
|
|
return RESTChunkedStream(tts=self, input_text=text, conn_options=conn_options)
|
|
|
|
def stream(
|
|
self, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS
|
|
) -> SynthesizeStream:
|
|
return SynthesizeStream(tts=self, conn_options=conn_options)
|
|
|
|
def update_options(
|
|
self,
|
|
*,
|
|
voice: str | None = None,
|
|
model: str | None = None,
|
|
**kwargs: Any,
|
|
) -> None:
|
|
_check_deprecated_tts_args(kwargs, caller="TTS.update_options")
|
|
|
|
if voice is not None:
|
|
if voice not in SUPPORTED_VOICES:
|
|
raise ValueError(
|
|
f"Voice '{voice}' not supported. "
|
|
f"Supported voices: {', '.join(sorted(SUPPORTED_VOICES))}"
|
|
)
|
|
self._opts.voice = voice
|
|
if model is not None:
|
|
self._opts.model = model
|
|
|
|
async def aclose(self) -> None:
|
|
pass
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _build_payload(opts: GnaniTTSOptions, text: str) -> dict:
|
|
audio_config: dict = {
|
|
"sample_rate": opts.sample_rate,
|
|
"encoding": opts.encoding,
|
|
"num_channels": opts.num_channels,
|
|
"sample_width": opts.sample_width,
|
|
"container": opts.container,
|
|
}
|
|
if opts.bitrate is not None:
|
|
audio_config["bitrate"] = opts.bitrate
|
|
return {
|
|
"text": text,
|
|
"voice": opts.voice,
|
|
"model": opts.model,
|
|
"audio_config": audio_config,
|
|
}
|
|
|
|
|
|
def _build_headers(opts: GnaniTTSOptions) -> dict[str, str]:
|
|
return {
|
|
"X-API-Key-ID": opts.api_key,
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
|
|
def _mime_type(opts: GnaniTTSOptions) -> str:
|
|
if opts.container == "raw":
|
|
return "audio/pcm"
|
|
return f"audio/{opts.container}"
|
|
|
|
|
|
def _strip_wav_header(data: bytes) -> bytes:
|
|
"""Strip the RIFF/WAV header if present, returning only PCM samples."""
|
|
if len(data) > _WAV_HEADER_SIZE and data[:4] == b"RIFF" and data[8:12] == b"WAVE":
|
|
return data[_WAV_HEADER_SIZE:]
|
|
return data
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REST ChunkedStream
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class RESTChunkedStream(tts.ChunkedStream):
|
|
"""REST-based chunked TTS — POST /api/v1/tts/inference."""
|
|
|
|
def __init__(self, *, tts: TTS, input_text: str, conn_options: APIConnectOptions) -> None:
|
|
super().__init__(tts=tts, input_text=input_text, conn_options=conn_options)
|
|
self._tts: TTS = tts
|
|
self._opts = replace(tts._opts)
|
|
|
|
async def _run(self, output_emitter: tts.AudioEmitter) -> None:
|
|
try:
|
|
async with self._tts._ensure_session().post(
|
|
url=f"{self._opts.base_url}/api/v1/tts/inference",
|
|
json=_build_payload(self._opts, self._input_text),
|
|
headers=_build_headers(self._opts),
|
|
timeout=aiohttp.ClientTimeout(
|
|
total=self._conn_options.timeout,
|
|
sock_connect=self._conn_options.timeout,
|
|
),
|
|
) as res:
|
|
if res.status != 200:
|
|
error_text = await res.text()
|
|
logger.error("Gnani TTS REST error: %s - %s", res.status, error_text)
|
|
raise APIStatusError(
|
|
message=f"Gnani TTS API Error ({res.status}): {error_text}",
|
|
status_code=res.status,
|
|
body=error_text,
|
|
)
|
|
|
|
audio_bytes = await res.read()
|
|
|
|
output_emitter.initialize(
|
|
request_id=utils.shortuuid(),
|
|
sample_rate=self._tts.sample_rate,
|
|
num_channels=self._tts.num_channels,
|
|
mime_type=_mime_type(self._opts),
|
|
)
|
|
output_emitter.push(audio_bytes)
|
|
output_emitter.flush()
|
|
|
|
except asyncio.TimeoutError as e:
|
|
raise APITimeoutError("Gnani TTS REST request timed out") from e
|
|
except (APIStatusError, APIConnectionError, APITimeoutError):
|
|
raise
|
|
except Exception as e:
|
|
raise APIConnectionError(f"Gnani TTS REST error: {e}") from e
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SSE ChunkedStream
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class SSEChunkedStream(tts.ChunkedStream):
|
|
"""SSE-based chunked TTS — POST /api/v1/tts/sse.
|
|
|
|
Each SSE chunk decodes to a complete WAV file. This class strips per-chunk
|
|
WAV headers and emits only raw PCM so the LiveKit pipeline receives a
|
|
single contiguous audio stream.
|
|
"""
|
|
|
|
def __init__(self, *, tts: TTS, input_text: str, conn_options: APIConnectOptions) -> None:
|
|
super().__init__(tts=tts, input_text=input_text, conn_options=conn_options)
|
|
self._tts: TTS = tts
|
|
self._opts = replace(tts._opts)
|
|
|
|
async def _run(self, output_emitter: tts.AudioEmitter) -> None:
|
|
try:
|
|
async with self._tts._ensure_session().post(
|
|
url=f"{self._opts.base_url}/api/v1/tts/sse",
|
|
json=_build_payload(self._opts, self._input_text),
|
|
headers=_build_headers(self._opts),
|
|
timeout=aiohttp.ClientTimeout(
|
|
total=self._conn_options.timeout,
|
|
sock_connect=self._conn_options.timeout,
|
|
),
|
|
read_bufsize=10 * 1024 * 1024,
|
|
) as res:
|
|
if res.status != 200:
|
|
error_text = await res.text()
|
|
logger.error("Gnani TTS SSE error: %s - %s", res.status, error_text)
|
|
raise APIStatusError(
|
|
message=f"Gnani TTS SSE Error ({res.status}): {error_text}",
|
|
status_code=res.status,
|
|
body=error_text,
|
|
)
|
|
|
|
output_emitter.initialize(
|
|
request_id=utils.shortuuid(),
|
|
sample_rate=self._tts.sample_rate,
|
|
num_channels=self._tts.num_channels,
|
|
mime_type="audio/pcm",
|
|
)
|
|
|
|
data_buf = ""
|
|
while True:
|
|
line_bytes = await res.content.readline()
|
|
if not line_bytes:
|
|
break
|
|
line = line_bytes.decode("utf-8").rstrip("\r\n")
|
|
|
|
if not line:
|
|
if not data_buf:
|
|
continue
|
|
try:
|
|
payload = json.loads(data_buf)
|
|
except json.JSONDecodeError:
|
|
data_buf = ""
|
|
continue
|
|
data_buf = ""
|
|
|
|
if payload.get("status") == "error" or "error" in payload:
|
|
raise APIStatusError(
|
|
message=payload.get("message", json.dumps(payload)),
|
|
status_code=500,
|
|
body=payload,
|
|
)
|
|
if payload.get("status") == "streaming_started":
|
|
continue
|
|
if payload.get("is_final", False):
|
|
audio_b64 = payload.get("audio", "")
|
|
if audio_b64:
|
|
output_emitter.push(_strip_wav_header(base64.b64decode(audio_b64)))
|
|
break
|
|
|
|
audio_b64 = payload.get("audio", "")
|
|
if audio_b64:
|
|
output_emitter.push(_strip_wav_header(base64.b64decode(audio_b64)))
|
|
continue
|
|
|
|
if line.startswith(":"):
|
|
continue
|
|
if line.startswith("event:"):
|
|
continue
|
|
if line.startswith("id:") or line.startswith("retry:"):
|
|
continue
|
|
if line.startswith("data:"):
|
|
data_buf += line[5:].strip()
|
|
else:
|
|
data_buf += line.strip()
|
|
|
|
output_emitter.flush()
|
|
|
|
except asyncio.TimeoutError as e:
|
|
raise APITimeoutError("Gnani TTS SSE request timed out") from e
|
|
except (APIStatusError, APIConnectionError, APITimeoutError):
|
|
raise
|
|
except Exception as e:
|
|
raise APIConnectionError(f"Gnani TTS SSE error: {e}") from e
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# WebSocket ChunkedStream (for synthesize_method="websocket")
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class WebSocketChunkedStream(tts.ChunkedStream):
|
|
"""WebSocket-based chunked TTS — wss://api.vachana.ai/api/v1/tts.
|
|
|
|
Wraps the WebSocket endpoint into the ChunkedStream interface so that
|
|
``synthesize()`` can use it when ``synthesize_method="websocket"``.
|
|
Each received chunk's WAV header is stripped; only raw PCM is emitted.
|
|
"""
|
|
|
|
def __init__(self, *, tts: TTS, input_text: str, conn_options: APIConnectOptions) -> None:
|
|
super().__init__(tts=tts, input_text=input_text, conn_options=conn_options)
|
|
self._tts: TTS = tts
|
|
self._opts = replace(tts._opts)
|
|
|
|
def _build_ws_url(self) -> str:
|
|
base = self._opts.base_url
|
|
if base.startswith("https://"):
|
|
ws_base = "wss://" + base[len("https://") :]
|
|
elif base.startswith("http://"):
|
|
ws_base = "ws://" + base[len("http://") :]
|
|
else:
|
|
ws_base = "wss://" + base
|
|
return f"{ws_base}/api/v1/tts"
|
|
|
|
async def _run(self, output_emitter: tts.AudioEmitter) -> None:
|
|
import websockets
|
|
|
|
try:
|
|
ws_url = self._build_ws_url()
|
|
async with websockets.connect(
|
|
ws_url,
|
|
additional_headers=_build_headers(self._opts),
|
|
ping_interval=20,
|
|
ping_timeout=20,
|
|
close_timeout=10,
|
|
) as ws:
|
|
request_body = _build_payload(self._opts, self._input_text)
|
|
await ws.send(json.dumps(request_body))
|
|
|
|
output_emitter.initialize(
|
|
request_id=utils.shortuuid(),
|
|
sample_rate=self._tts.sample_rate,
|
|
num_channels=self._tts.num_channels,
|
|
mime_type="audio/pcm",
|
|
)
|
|
|
|
async for msg in ws:
|
|
if isinstance(msg, bytes):
|
|
output_emitter.push(_strip_wav_header(msg))
|
|
continue
|
|
|
|
payload = json.loads(msg)
|
|
msg_type = payload.get("type", "")
|
|
|
|
if msg_type == "audio":
|
|
inner = payload.get("data", {})
|
|
audio_b64 = inner.get("audio", "")
|
|
if audio_b64:
|
|
output_emitter.push(_strip_wav_header(base64.b64decode(audio_b64)))
|
|
|
|
elif msg_type == "complete":
|
|
inner = payload.get("data")
|
|
if inner is not None:
|
|
audio_b64 = inner.get("audio", "")
|
|
if audio_b64:
|
|
output_emitter.push(_strip_wav_header(base64.b64decode(audio_b64)))
|
|
break
|
|
|
|
elif msg_type == "error":
|
|
error_msg = payload.get("message", "Unknown error")
|
|
logger.error("Gnani TTS WS error: %s", error_msg)
|
|
raise APIStatusError(
|
|
message=f"Gnani TTS stream error: {error_msg}",
|
|
status_code=500,
|
|
body=error_msg,
|
|
)
|
|
|
|
output_emitter.flush()
|
|
|
|
except websockets.exceptions.ConnectionClosed as e:
|
|
raise APIConnectionError(f"Gnani TTS WebSocket closed: {e}") from e
|
|
except asyncio.TimeoutError as e:
|
|
raise APITimeoutError("Gnani TTS WebSocket timed out") from e
|
|
except (APIStatusError, APIConnectionError, APITimeoutError):
|
|
raise
|
|
except Exception as e:
|
|
raise APIConnectionError(f"Gnani TTS WebSocket error: {e}") from e
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# WebSocket SynthesizeStream (for stream())
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class SynthesizeStream(tts.SynthesizeStream):
|
|
"""WebSocket-based streaming TTS — wss://api.vachana.ai/api/v1/tts."""
|
|
|
|
def __init__(self, *, tts: TTS, conn_options: APIConnectOptions):
|
|
super().__init__(tts=tts, conn_options=conn_options)
|
|
self._tts: TTS = tts
|
|
self._opts = replace(tts._opts)
|
|
|
|
def _build_ws_url(self) -> str:
|
|
base = self._opts.base_url
|
|
if base.startswith("https://"):
|
|
ws_base = "wss://" + base[len("https://") :]
|
|
elif base.startswith("http://"):
|
|
ws_base = "ws://" + base[len("http://") :]
|
|
else:
|
|
ws_base = "wss://" + base
|
|
return f"{ws_base}/api/v1/tts"
|
|
|
|
async def _run(self, output_emitter: tts.AudioEmitter) -> None:
|
|
import websockets
|
|
|
|
request_id = utils.shortuuid()
|
|
output_emitter.initialize(
|
|
request_id=request_id,
|
|
sample_rate=self._tts.sample_rate,
|
|
num_channels=self._tts.num_channels,
|
|
mime_type="audio/pcm",
|
|
stream=True,
|
|
)
|
|
|
|
text_parts: list[str] = []
|
|
async for data in self._input_ch:
|
|
if isinstance(data, str):
|
|
text_parts.append(data)
|
|
elif isinstance(data, self._FlushSentinel):
|
|
break
|
|
|
|
full_text = "".join(text_parts).strip()
|
|
if not full_text:
|
|
return
|
|
|
|
segment_id = utils.shortuuid()
|
|
output_emitter.start_segment(segment_id=segment_id)
|
|
|
|
try:
|
|
ws_url = self._build_ws_url()
|
|
async with websockets.connect(
|
|
ws_url,
|
|
additional_headers=_build_headers(self._opts),
|
|
ping_interval=20,
|
|
ping_timeout=20,
|
|
close_timeout=10,
|
|
) as ws:
|
|
request_body = _build_payload(self._opts, full_text)
|
|
await ws.send(json.dumps(request_body))
|
|
|
|
self._mark_started()
|
|
|
|
async for msg in ws:
|
|
if isinstance(msg, bytes):
|
|
output_emitter.push(_strip_wav_header(msg))
|
|
continue
|
|
|
|
payload = json.loads(msg)
|
|
msg_type = payload.get("type", "")
|
|
|
|
if msg_type == "audio":
|
|
inner = payload.get("data", {})
|
|
audio_b64 = inner.get("audio", "")
|
|
if audio_b64:
|
|
output_emitter.push(_strip_wav_header(base64.b64decode(audio_b64)))
|
|
|
|
elif msg_type == "complete":
|
|
inner = payload.get("data")
|
|
if inner is not None:
|
|
audio_b64 = inner.get("audio", "")
|
|
if audio_b64:
|
|
output_emitter.push(_strip_wav_header(base64.b64decode(audio_b64)))
|
|
break
|
|
|
|
elif msg_type == "error":
|
|
error_msg = payload.get("message", "Unknown error")
|
|
logger.error("Gnani TTS WS error: %s", error_msg)
|
|
raise APIStatusError(
|
|
message=f"Gnani TTS stream error: {error_msg}",
|
|
status_code=500,
|
|
body=error_msg,
|
|
)
|
|
|
|
except websockets.exceptions.ConnectionClosed as e:
|
|
raise APIConnectionError(f"Gnani TTS WebSocket closed: {e}") from e
|
|
except asyncio.TimeoutError as e:
|
|
raise APITimeoutError("Gnani TTS WebSocket timed out") from e
|
|
except (APIStatusError, APIConnectionError, APITimeoutError):
|
|
raise
|
|
except Exception as e:
|
|
raise APIConnectionError(f"Gnani TTS WebSocket error: {e}") from e
|
|
|
|
output_emitter.end_segment()
|