services(cartesia): added CartesiaHttpTTSService

CHANGELOG.md

@@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Added `CartesiaHttpTTSService`. This is a synchronous frame processor
+  (i.e. given an input text frame it will wait for the whole output before
+  returning).
+
 - A clock can now be specified to `PipelineTask` (defaults to
   `SystemClock`). This clock will be passed to each frame processor via the
   `StartFrame`.

pyproject.toml

@@ -36,7 +36,7 @@ Website = "https://pipecat.ai"
 [project.optional-dependencies]
 anthropic = [ "anthropic~=0.34.0" ]
 azure = [ "azure-cognitiveservices-speech~=1.40.0" ]
-cartesia = [ "websockets~=12.0" ]
+cartesia = [ "cartesia~=1.0.13", "websockets~=12.0" ]
 daily = [ "daily-python~=0.10.1" ]
 deepgram = [ "deepgram-sdk~=3.5.0" ]
 elevenlabs = [ "websockets~=12.0" ]

src/pipecat/services/cartesia.py

@@ -8,7 +8,6 @@ import json
 import uuid
 import base64
 import asyncio
-import time
 
 from typing import AsyncGenerator
 
@@ -22,17 +21,17 @@ from pipecat.frames.frames import (
     EndFrame,
     TTSStartedFrame,
     TTSStoppedFrame,
-    TextFrame,
     LLMFullResponseEndFrame
 )
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.transcriptions.language import Language
-from pipecat.services.ai_services import AsyncWordTTSService
+from pipecat.services.ai_services import AsyncWordTTSService, TTSService
 
 from loguru import logger
 
 # See .env.example for Cartesia configuration needed
 try:
+    from cartesia import AsyncCartesia
     import websockets
 except ModuleNotFoundError as e:
     logger.error(f"Exception: {e}")
@@ -165,7 +164,7 @@ class CartesiaTTSService(AsyncWordTTSService):
     async def flush_audio(self):
         if not self._context_id or not self._websocket:
             return
-        logger.debug("Flushing audio")
+        logger.trace("Flushing audio")
         msg = {
             "transcript": "",
             "continue": False,
@@ -257,3 +256,84 @@ class CartesiaTTSService(AsyncWordTTSService):
             yield None
         except Exception as e:
             logger.error(f"{self} exception: {e}")
+
+
+class CartesiaHttpTTSService(TTSService):
+
+    def __init__(
+            self,
+            *,
+            api_key: str,
+            voice_id: str,
+            model_id: str = "sonic-english",
+            base_url: str = "https://api.cartesia.ai",
+            encoding: str = "pcm_s16le",
+            sample_rate: int = 16000,
+            language: str = "en",
+            **kwargs):
+        super().__init__(**kwargs)
+
+        self._api_key = api_key
+        self._voice_id = voice_id
+        self._model_id = model_id
+        self._output_format = {
+            "container": "raw",
+            "encoding": encoding,
+            "sample_rate": sample_rate,
+        }
+        self._language = language
+
+        self._client = AsyncCartesia(api_key=api_key, base_url=base_url)
+
+    def can_generate_metrics(self) -> bool:
+        return True
+
+    async def set_model(self, model: str):
+        logger.debug(f"Switching TTS model to: [{model}]")
+        self._model_id = model
+
+    async def set_voice(self, voice: str):
+        logger.debug(f"Switching TTS voice to: [{voice}]")
+        self._voice_id = voice
+
+    async def set_language(self, language: Language):
+        logger.debug(f"Switching TTS language to: [{language}]")
+        self._language = language_to_cartesia_language(language)
+
+    async def stop(self, frame: EndFrame):
+        await super().stop(frame)
+        await self._client.close()
+
+    async def cancel(self, frame: CancelFrame):
+        await super().cancel(frame)
+        await self._client.close()
+
+    async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
+        logger.debug(f"Generating TTS: [{text}]")
+
+        await self.push_frame(TTSStartedFrame())
+        await self.start_ttfb_metrics()
+
+        try:
+            output = await self._client.tts.sse(
+                model_id=self._model_id,
+                transcript=text,
+                voice_id=self._voice_id,
+                output_format=self._output_format,
+                language=self._language,
+                stream=False
+            )
+
+            await self.stop_ttfb_metrics()
+
+            frame = AudioRawFrame(
+                audio=output["audio"],
+                sample_rate=self._output_format["sample_rate"],
+                num_channels=1
+            )
+            yield frame
+        except Exception as e:
+            logger.error(f"{self} exception: {e}")
+
+        await self.start_tts_usage_metrics(text)
+        await self.push_frame(TTSStoppedFrame())