diff --git a/src/pipecat/services/cartesia/tts.py b/src/pipecat/services/cartesia/tts.py index 22493f229..402ea27a0 100644 --- a/src/pipecat/services/cartesia/tts.py +++ b/src/pipecat/services/cartesia/tts.py @@ -4,6 +4,8 @@ # SPDX-License-Identifier: BSD 2-Clause License # +"""Cartesia text-to-speech service implementations.""" + import base64 import json import uuid @@ -42,6 +44,14 @@ except ModuleNotFoundError as e: def language_to_cartesia_language(language: Language) -> Optional[str]: + """Convert a Language enum to Cartesia language code. + + Args: + language: The Language enum value to convert. + + Returns: + The corresponding Cartesia language code, or None if not supported. + """ BASE_LANGUAGES = { Language.DE: "de", Language.EN: "en", @@ -74,7 +84,35 @@ def language_to_cartesia_language(language: Language) -> Optional[str]: class CartesiaTTSService(AudioContextWordTTSService): + """Cartesia TTS service with WebSocket streaming and word timestamps. + + Provides text-to-speech using Cartesia's streaming WebSocket API. + Supports word-level timestamps, audio context management, and various voice + customization options including speed and emotion controls. + + Args: + api_key: Cartesia API key for authentication. + voice_id: ID of the voice to use for synthesis. + cartesia_version: API version string for Cartesia service. + url: WebSocket URL for Cartesia TTS API. + model: TTS model to use (e.g., "sonic-2"). + sample_rate: Audio sample rate. If None, uses default. + encoding: Audio encoding format. + container: Audio container format. + params: Additional input parameters for voice customization. + text_aggregator: Custom text aggregator for processing input text. + **kwargs: Additional arguments passed to the parent service. + """ + class InputParams(BaseModel): + """Input parameters for Cartesia TTS configuration. + + Parameters: + language: Language to use for synthesis. + speed: Voice speed control (string or float). + emotion: List of emotion controls (deprecated). + """ + language: Optional[Language] = Language.EN speed: Optional[Union[str, float]] = "" emotion: Optional[List[str]] = [] @@ -137,14 +175,32 @@ class CartesiaTTSService(AudioContextWordTTSService): self._receive_task = None def can_generate_metrics(self) -> bool: + """Check if this service can generate processing metrics. + + Returns: + True, as Cartesia service supports metrics generation. + """ return True async def set_model(self, model: str): + """Set the TTS model. + + Args: + model: The model name to use for synthesis. + """ self._model_id = model await super().set_model(model) logger.info(f"Switching TTS model to: [{model}]") def language_to_service_language(self, language: Language) -> Optional[str]: + """Convert a Language enum to Cartesia language format. + + Args: + language: The language to convert. + + Returns: + The Cartesia-specific language code, or None if not supported. + """ return language_to_cartesia_language(language) def _build_msg( @@ -182,15 +238,30 @@ class CartesiaTTSService(AudioContextWordTTSService): return json.dumps(msg) async def start(self, frame: StartFrame): + """Start the Cartesia TTS service. + + Args: + frame: The start frame containing initialization parameters. + """ await super().start(frame) self._settings["output_format"]["sample_rate"] = self.sample_rate await self._connect() async def stop(self, frame: EndFrame): + """Stop the Cartesia TTS service. + + Args: + frame: The end frame. + """ await super().stop(frame) await self._disconnect() async def cancel(self, frame: CancelFrame): + """Stop the Cartesia TTS service. + + Args: + frame: The end frame. + """ await super().cancel(frame) await self._disconnect() @@ -247,6 +318,7 @@ class CartesiaTTSService(AudioContextWordTTSService): self._context_id = None async def flush_audio(self): + """Flush any pending audio and finalize the current context.""" if not self._context_id or not self._websocket: return logger.trace(f"{self}: flushing audio") @@ -287,6 +359,14 @@ class CartesiaTTSService(AudioContextWordTTSService): @traced_tts async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + """Generate speech from text using Cartesia's streaming API. + + Args: + text: The text to synthesize into speech. + + Yields: + Frame: Audio frames containing the synthesized speech. + """ logger.debug(f"{self}: Generating TTS [{text}]") try: @@ -316,7 +396,34 @@ class CartesiaTTSService(AudioContextWordTTSService): class CartesiaHttpTTSService(TTSService): + """Cartesia HTTP-based TTS service. + + Provides text-to-speech using Cartesia's HTTP API for simpler, non-streaming + synthesis. Suitable for use cases where streaming is not required and simpler + integration is preferred. + + Args: + api_key: Cartesia API key for authentication. + voice_id: ID of the voice to use for synthesis. + model: TTS model to use (e.g., "sonic-2"). + base_url: Base URL for Cartesia HTTP API. + cartesia_version: API version string for Cartesia service. + sample_rate: Audio sample rate. If None, uses default. + encoding: Audio encoding format. + container: Audio container format. + params: Additional input parameters for voice customization. + **kwargs: Additional arguments passed to the parent TTSService. + """ + class InputParams(BaseModel): + """Input parameters for Cartesia HTTP TTS configuration. + + Parameters: + language: Language to use for synthesis. + speed: Voice speed control (string or float). + emotion: List of emotion controls (deprecated). + """ + language: Optional[Language] = Language.EN speed: Optional[Union[str, float]] = "" emotion: Optional[List[str]] = Field(default_factory=list) @@ -363,25 +470,61 @@ class CartesiaHttpTTSService(TTSService): ) def can_generate_metrics(self) -> bool: + """Check if this service can generate processing metrics. + + Returns: + True, as Cartesia HTTP service supports metrics generation. + """ return True def language_to_service_language(self, language: Language) -> Optional[str]: + """Convert a Language enum to Cartesia language format. + + Args: + language: The language to convert. + + Returns: + The Cartesia-specific language code, or None if not supported. + """ return language_to_cartesia_language(language) async def start(self, frame: StartFrame): + """Start the Cartesia HTTP TTS service. + + Args: + frame: The start frame containing initialization parameters. + """ await super().start(frame) self._settings["output_format"]["sample_rate"] = self.sample_rate async def stop(self, frame: EndFrame): + """Stop the Cartesia HTTP TTS service. + + Args: + frame: The end frame. + """ await super().stop(frame) await self._client.close() async def cancel(self, frame: CancelFrame): + """Cancel the Cartesia HTTP TTS service. + + Args: + frame: The cancel frame. + """ await super().cancel(frame) await self._client.close() @traced_tts async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + """Generate speech from text using Cartesia's HTTP API. + + Args: + text: The text to synthesize into speech. + + Yields: + Frame: Audio frames containing the synthesized speech. + """ logger.debug(f"{self}: Generating TTS [{text}]") try: