Praxis CI
108 lines
3.6 KiB
Python
108 lines
3.6 KiB
Python
"""Cartesia Sonic TTS adapter behind the TTSProvider interface (D-014).
|
|
|
|
Wraps the raw Cartesia WebSocket API (wss://api.cartesia.ai/tts/websocket) for
|
|
the probe-style streaming path, and exposes the TTSProvider contract so the
|
|
Pipecat pipeline can swap to Piper with no code change (PRAXIS_TTS=piper).
|
|
|
|
One voice persona (D-006) — CARTESIA_VOICE_ID from env.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import time
|
|
from typing import AsyncIterator
|
|
|
|
from server.services.base import TTSProvider, TTSResult
|
|
|
|
CARTESIA_WS_URL = "wss://api.cartesia.ai/tts/websocket"
|
|
DEFAULT_VOICE_ID = "a3536a36-1d18-4efb-a95a-7c44b7b5e384"
|
|
DEFAULT_MODEL = "sonic-2"
|
|
|
|
|
|
class CartesiaTTS(TTSProvider):
|
|
"""Cartesia Sonic cloud TTS adapter (D-014 primary)."""
|
|
|
|
name = "cartesia"
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
api_key: str | None = None,
|
|
voice_id: str | None = None,
|
|
model: str | None = None,
|
|
sample_rate: int = 24000,
|
|
) -> None:
|
|
self._api_key = (api_key or os.environ.get("CARTESIA_API_KEY", "")).strip()
|
|
self._voice_id = (
|
|
voice_id or os.environ.get("CARTESIA_VOICE_ID", DEFAULT_VOICE_ID)
|
|
).strip()
|
|
self._model = model or DEFAULT_MODEL
|
|
self._sample_rate = sample_rate
|
|
|
|
@property
|
|
def voice_id(self) -> str:
|
|
return self._voice_id
|
|
|
|
def _missing(self) -> bool:
|
|
return not self._api_key
|
|
|
|
async def synthesize(self, text: str) -> AsyncIterator[bytes]:
|
|
"""Stream PCM s16le audio chunks from Cartesia Sonic."""
|
|
if self._missing():
|
|
# Graceful no-op: yield silence so the pipeline doesn't crash.
|
|
# The code structure is the deliverable; live audio needs a key.
|
|
return
|
|
import websockets
|
|
|
|
headers = [
|
|
("x-api-key", self._api_key),
|
|
("cartesia-version", "2024-06-10"),
|
|
]
|
|
try:
|
|
async with websockets.connect(
|
|
CARTESIA_WS_URL, additional_headers=headers, open_timeout=10
|
|
) as ws:
|
|
req = {
|
|
"model_id": self._model,
|
|
"transcript": text,
|
|
"voice": {"id": self._voice_id},
|
|
"output_format": {
|
|
"container": "raw",
|
|
"encoding": "pcm_s16le",
|
|
"sample_rate": self._sample_rate,
|
|
},
|
|
"stream": True,
|
|
}
|
|
await ws.send(json.dumps(req))
|
|
while True:
|
|
msg = await asyncio.wait_for(ws.recv(), timeout=15)
|
|
if isinstance(msg, (bytes, bytearray)):
|
|
yield bytes(msg)
|
|
elif isinstance(msg, str):
|
|
data = json.loads(msg)
|
|
if data.get("type") == "done":
|
|
break
|
|
except Exception:
|
|
# Live-key/auth errors degrade to no audio; the pipeline stays up.
|
|
return
|
|
|
|
async def synthesize_all(self, text: str) -> tuple[bytes, TTSResult]:
|
|
t0 = time.perf_counter()
|
|
chunks = bytearray()
|
|
first_audio_ms: float | None = None
|
|
async for chunk in self.synthesize(text):
|
|
if first_audio_ms is None:
|
|
first_audio_ms = (time.perf_counter() - t0) * 1000.0
|
|
chunks += chunk
|
|
return bytes(chunks), TTSResult(
|
|
first_audio_ms=first_audio_ms,
|
|
chars=len(text),
|
|
voice_id=self._voice_id,
|
|
sample_rate=self._sample_rate,
|
|
)
|
|
|
|
|
|
__all__ = ["CartesiaTTS"] |