From 7ae0d651d6bf426b54afda2db881688326fa8948 Mon Sep 17 00:00:00 2001 From: Neil Ruaro Date: Mon, 29 Dec 2025 21:08:40 +0800 Subject: [PATCH] added cambai tts integration --- .../foundational/07za-interruptible-camb.py | 207 ++++++++ examples/foundational/test_camb_quick.py | 181 +++++++ pyproject.toml | 1 + src/pipecat/services/camb/__init__.py | 8 + src/pipecat/services/camb/tts.py | 467 ++++++++++++++++++ tests/test_camb_tts.py | 431 ++++++++++++++++ uv.lock | 6 +- 7 files changed, 1300 insertions(+), 1 deletion(-) create mode 100644 examples/foundational/07za-interruptible-camb.py create mode 100644 examples/foundational/test_camb_quick.py create mode 100644 src/pipecat/services/camb/__init__.py create mode 100644 src/pipecat/services/camb/tts.py create mode 100644 tests/test_camb_tts.py diff --git a/examples/foundational/07za-interruptible-camb.py b/examples/foundational/07za-interruptible-camb.py new file mode 100644 index 000000000..4266e6f17 --- /dev/null +++ b/examples/foundational/07za-interruptible-camb.py @@ -0,0 +1,207 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Camb.ai MARS-8 TTS example with interruption handling. + +This example demonstrates: +- Basic TTS synthesis with Camb.ai MARS-8 +- Voice selection +- Speed control +- Handling interruptions + +Requirements: +- CAMB_API_KEY environment variable +- OPENAI_API_KEY environment variable (for LLM) +- DEEPGRAM_API_KEY environment variable (for STT) + +Usage: + export CAMB_API_KEY=your_camb_api_key + export OPENAI_API_KEY=your_openai_api_key + export DEEPGRAM_API_KEY=your_deepgram_api_key + python 07za-interruptible-camb.py --transport daily + +For more information: +- Camb.ai API docs: https://camb.mintlify.app/ +- Pipecat docs: https://docs.pipecat.ai/ +""" + +import os + +import aiohttp +from dotenv import load_dotenv +from loguru import logger + +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.audio.vad.vad_analyzer import VADParams +from pipecat.frames.frames import LLMRunFrame +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.aggregators.llm_response_universal import ( + LLMContextAggregatorPair, +) +from pipecat.runner.types import RunnerArguments +from pipecat.runner.utils import create_transport +from pipecat.services.camb.tts import CambTTSService +from pipecat.services.deepgram.stt import DeepgramSTTService +from pipecat.services.openai.llm import OpenAILLMService +from pipecat.transports.base_transport import BaseTransport, TransportParams +from pipecat.transports.daily.transport import DailyParams +from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams + +load_dotenv(override=True) + + +# Transport configuration for different platforms +transport_params = { + "daily": lambda: DailyParams( + audio_in_enabled=True, + audio_out_enabled=True, + vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)), + ), + "twilio": lambda: FastAPIWebsocketParams( + audio_in_enabled=True, + audio_out_enabled=True, + vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)), + ), + "webrtc": lambda: TransportParams( + audio_in_enabled=True, + audio_out_enabled=True, + vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)), + ), +} + + +async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): + """Run the bot with Camb.ai TTS. + + Args: + transport: The transport to use for audio I/O. + runner_args: Runner arguments from the CLI. + """ + logger.info("Starting Camb.ai TTS bot") + + # Create an HTTP session for the TTS service + async with aiohttp.ClientSession() as session: + # Initialize Deepgram STT for speech recognition + stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + + # Initialize Camb.ai TTS with MARS-8-flash model (fastest) + tts = CambTTSService( + api_key=os.getenv("CAMB_API_KEY"), + aiohttp_session=session, + voice_id=2681, # Attic voice (default) + model="mars-8-flash", # Fast inference model + params=CambTTSService.InputParams( + speed=1.0, # Normal speed (0.5-2.0 range) + ), + ) + + # Initialize OpenAI LLM + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) + + # System prompt for the assistant + messages = [ + { + "role": "system", + "content": """You are a helpful voice assistant powered by Camb.ai's MARS-8 +text-to-speech technology. Your goal is to have natural conversations and demonstrate +high-quality speech synthesis. Keep your responses concise and conversational since +they will be spoken aloud. Avoid special characters, emojis, or bullet points that +can't easily be spoken.""", + }, + ] + + # Set up context management + context = LLMContext(messages) + context_aggregator = LLMContextAggregatorPair(context) + + # Build the pipeline + pipeline = Pipeline( + [ + transport.input(), # Transport user input + stt, # Speech-to-text + context_aggregator.user(), # User context aggregation + llm, # Language model + tts, # Camb.ai TTS + transport.output(), # Transport bot output + context_aggregator.assistant(), # Assistant context aggregation + ] + ) + + # Create the pipeline task + task = PipelineTask( + pipeline, + params=PipelineParams( + enable_metrics=True, + enable_usage_metrics=True, + ), + idle_timeout_secs=runner_args.pipeline_idle_timeout_secs, + ) + + @transport.event_handler("on_client_connected") + async def on_client_connected(transport, client): + logger.info("Client connected") + # Start the conversation with a greeting + messages.append( + { + "role": "system", + "content": "Please introduce yourself briefly and ask how you can help.", + } + ) + await task.queue_frames([LLMRunFrame()]) + + @transport.event_handler("on_client_disconnected") + async def on_client_disconnected(transport, client): + logger.info("Client disconnected") + await task.cancel() + + # Run the pipeline + runner = PipelineRunner(handle_sigint=runner_args.handle_sigint) + await runner.run(task) + + +async def bot(runner_args: RunnerArguments): + """Main bot entry point compatible with Pipecat Cloud. + + Args: + runner_args: Arguments passed from the runner. + """ + transport = await create_transport(runner_args, transport_params) + await run_bot(transport, runner_args) + + +async def list_available_voices(): + """Helper function to list available Camb.ai voices. + + Run this to see what voices are available for your API key. + """ + async with aiohttp.ClientSession() as session: + voices = await CambTTSService.list_voices( + api_key=os.getenv("CAMB_API_KEY"), + aiohttp_session=session, + ) + print("\nAvailable Camb.ai voices:") + print("-" * 50) + for voice in voices: + print(f" ID: {voice['id']}, Name: {voice['name']}, Gender: {voice['gender']}") + print("-" * 50) + print(f"Total: {len(voices)} voices\n") + + +if __name__ == "__main__": + import sys + + # If --list-voices flag is passed, list voices and exit + if "--list-voices" in sys.argv: + import asyncio + + asyncio.run(list_available_voices()) + else: + from pipecat.runner.run import main + + main() diff --git a/examples/foundational/test_camb_quick.py b/examples/foundational/test_camb_quick.py new file mode 100644 index 000000000..d178b2b7c --- /dev/null +++ b/examples/foundational/test_camb_quick.py @@ -0,0 +1,181 @@ +#!/usr/bin/env python3 +"""Quick test script to verify Camb.ai TTS integration works. + +Usage: + export CAMB_API_KEY=your_api_key + python test_camb_quick.py +""" + +import asyncio +import os +import sys + +# Add the src directory to the path so we can import the module +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "src")) + +import aiohttp +from dotenv import load_dotenv + +load_dotenv() + + +async def test_list_voices(): + """Test listing available voices.""" + from pipecat.services.camb.tts import CambTTSService + + api_key = os.getenv("CAMB_API_KEY") + if not api_key: + print("ERROR: CAMB_API_KEY environment variable not set!") + return False + + print("\n1. Testing list_voices()...") + async with aiohttp.ClientSession() as session: + try: + voices = await CambTTSService.list_voices( + api_key=api_key, + aiohttp_session=session, + ) + print(f" SUCCESS: Found {len(voices)} voices") + if voices: + print(f" First voice: ID={voices[0]['id']}, Name={voices[0]['name']}") + return True + except Exception as e: + print(f" FAILED: {e}") + import traceback + traceback.print_exc() + return False + + +async def test_tts_synthesis(): + """Test basic TTS synthesis.""" + from pipecat.services.camb.tts import CambTTSService + from pipecat.frames.frames import TTSAudioRawFrame, TTSStartedFrame, TTSStoppedFrame, ErrorFrame + + api_key = os.getenv("CAMB_API_KEY") + if not api_key: + print("ERROR: CAMB_API_KEY environment variable not set!") + return False + + print("\n2. Testing TTS synthesis...") + async with aiohttp.ClientSession() as session: + tts = CambTTSService( + api_key=api_key, + aiohttp_session=session, + voice_id=2681, # Attic voice + model="mars-8-flash", + ) + + # Manually set sample rate (normally done by StartFrame) + tts._sample_rate = 24000 + + text = "Hello! This is a test of the Camb.ai text to speech integration." + print(f" Synthesizing: '{text}'") + + audio_bytes = 0 + frames_received = [] + + try: + async for frame in tts.run_tts(text): + frames_received.append(type(frame).__name__) + if isinstance(frame, TTSAudioRawFrame): + audio_bytes += len(frame.audio) + elif isinstance(frame, ErrorFrame): + print(f" FAILED: {frame.error}") + return False + + print(f" Frames received: {frames_received}") + print(f" Audio bytes received: {audio_bytes}") + + if audio_bytes > 0: + print(" SUCCESS: TTS synthesis works!") + + # Optionally save and play audio + save_audio = input("\n Save audio to test_output.wav? (y/n): ").strip().lower() + if save_audio == 'y': + await save_audio_to_file(tts, text) + # Try to play the audio + play_audio = input(" Play the audio? (y/n): ").strip().lower() + if play_audio == 'y': + play_wav_file("test_output.wav") + + return True + else: + print(" FAILED: No audio received") + return False + + except Exception as e: + print(f" FAILED: {e}") + import traceback + traceback.print_exc() + return False + + +async def save_audio_to_file(tts, text): + """Save synthesized audio to a WAV file.""" + import wave + from pipecat.frames.frames import TTSAudioRawFrame + + audio_data = bytearray() + async for frame in tts.run_tts(text): + if isinstance(frame, TTSAudioRawFrame): + audio_data.extend(frame.audio) + + if audio_data: + with wave.open("test_output.wav", "wb") as wav_file: + wav_file.setnchannels(1) # Mono + wav_file.setsampwidth(2) # 16-bit + wav_file.setframerate(24000) # 24kHz + wav_file.writeframes(bytes(audio_data)) + print(" Saved to test_output.wav") + + +def play_wav_file(filepath): + """Play a WAV file using the system's default player.""" + import subprocess + import platform + + system = platform.system() + try: + if system == "Darwin": # macOS + subprocess.run(["afplay", filepath], check=True) + elif system == "Linux": + subprocess.run(["aplay", filepath], check=True) + elif system == "Windows": + subprocess.run(["powershell", "-c", f"(New-Object Media.SoundPlayer '{filepath}').PlaySync()"], check=True) + else: + print(f" Unsupported platform: {system}. Please play {filepath} manually.") + except Exception as e: + print(f" Could not play audio: {e}") + + +async def main(): + print("=" * 50) + print("Camb.ai TTS Integration Test") + print("=" * 50) + + results = [] + + # Test 1: List voices + results.append(await test_list_voices()) + + # Test 2: TTS synthesis + results.append(await test_tts_synthesis()) + + # Summary + print("\n" + "=" * 50) + print("Summary:") + print(f" List voices: {'PASS' if results[0] else 'FAIL'}") + print(f" TTS synthesis: {'PASS' if results[1] else 'FAIL'}") + print("=" * 50) + + if all(results): + print("\nAll tests passed!") + return 0 + else: + print("\nSome tests failed!") + return 1 + + +if __name__ == "__main__": + exit_code = asyncio.run(main()) + sys.exit(exit_code) diff --git a/pyproject.toml b/pyproject.toml index 7f50189de..f640e9fb1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -53,6 +53,7 @@ aws = [ "aioboto3~=15.5.0", "pipecat-ai[websockets-base]" ] aws-nova-sonic = [ "aws_sdk_bedrock_runtime~=0.2.0; python_version>='3.12'" ] azure = [ "azure-cognitiveservices-speech~=1.44.0"] cartesia = [ "cartesia~=2.0.3", "pipecat-ai[websockets-base]" ] +camb = [ "pipecat-ai[websockets-base]" ] cerebras = [] daily = [ "daily-python~=0.23.0" ] deepgram = [ "deepgram-sdk~=4.7.0", "pipecat-ai[websockets-base]" ] diff --git a/src/pipecat/services/camb/__init__.py b/src/pipecat/services/camb/__init__.py new file mode 100644 index 000000000..0d444e949 --- /dev/null +++ b/src/pipecat/services/camb/__init__.py @@ -0,0 +1,8 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + + +from .tts import * diff --git a/src/pipecat/services/camb/tts.py b/src/pipecat/services/camb/tts.py new file mode 100644 index 000000000..a6bcfb9e1 --- /dev/null +++ b/src/pipecat/services/camb/tts.py @@ -0,0 +1,467 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Camb.ai MARS-8 text-to-speech service implementation. + +This module provides TTS functionality using Camb.ai's MARS-8 model family, +offering high-quality text-to-speech synthesis with HTTP streaming support. + +Features: + - MARS-8 models: mars-8, mars-8-flash, mars-8-instruct + - 140+ languages supported + - Real-time HTTP streaming + - 24kHz audio output + - Voice customization (speed, instructions) +""" + +import asyncio +from typing import Any, AsyncGenerator, Dict, List, Mapping, Optional + +import aiohttp +from loguru import logger +from pydantic import BaseModel, Field + +from pipecat.frames.frames import ( + ErrorFrame, + Frame, + StartFrame, + TTSAudioRawFrame, + TTSStartedFrame, + TTSStoppedFrame, +) +from pipecat.services.tts_service import TTSService +from pipecat.transcriptions.language import Language, resolve_language +from pipecat.utils.tracing.service_decorators import traced_tts + + +# Default configuration +DEFAULT_VOICE_ID = 2681 # Attic voice (publicly available) +DEFAULT_LANGUAGE = "en-us" +DEFAULT_MODEL = "mars-8-flash" # Faster inference +DEFAULT_BASE_URL = "https://client.camb.ai/apis" +DEFAULT_SAMPLE_RATE = 24000 # 24kHz +DEFAULT_TIMEOUT = 60.0 # Seconds (minimum recommended by Camb.ai) +MIN_TEXT_LENGTH = 3 +MAX_TEXT_LENGTH = 3000 + + +def language_to_camb_language(language: Language) -> Optional[str]: + """Convert a Pipecat Language enum to Camb.ai language code. + + Args: + language: The Language enum value to convert. + + Returns: + The corresponding Camb.ai language code (BCP-47 format), or None if not supported. + """ + LANGUAGE_MAP = { + Language.EN: "en-us", + Language.EN_US: "en-us", + Language.EN_GB: "en-gb", + Language.EN_AU: "en-au", + Language.ES: "es-es", + Language.ES_ES: "es-es", + Language.ES_MX: "es-mx", + Language.FR: "fr-fr", + Language.FR_FR: "fr-fr", + Language.FR_CA: "fr-ca", + Language.DE: "de-de", + Language.DE_DE: "de-de", + Language.IT: "it-it", + Language.PT: "pt-pt", + Language.PT_BR: "pt-br", + Language.PT_PT: "pt-pt", + Language.NL: "nl-nl", + Language.PL: "pl-pl", + Language.RU: "ru-ru", + Language.JA: "ja-jp", + Language.KO: "ko-kr", + Language.ZH: "zh-cn", + Language.ZH_CN: "zh-cn", + Language.ZH_TW: "zh-tw", + Language.AR: "ar-sa", + Language.HI: "hi-in", + Language.TR: "tr-tr", + Language.VI: "vi-vn", + Language.TH: "th-th", + Language.ID: "id-id", + Language.MS: "ms-my", + Language.SV: "sv-se", + Language.DA: "da-dk", + Language.NO: "no-no", + Language.FI: "fi-fi", + Language.CS: "cs-cz", + Language.EL: "el-gr", + Language.HE: "he-il", + Language.HU: "hu-hu", + Language.RO: "ro-ro", + Language.SK: "sk-sk", + Language.UK: "uk-ua", + Language.BG: "bg-bg", + Language.HR: "hr-hr", + Language.SR: "sr-rs", + Language.SL: "sl-si", + Language.CA: "ca-es", + Language.EU: "eu-es", + Language.GL: "gl-es", + Language.AF: "af-za", + Language.SW: "sw-ke", + Language.TA: "ta-in", + Language.TE: "te-in", + Language.BN: "bn-in", + Language.MR: "mr-in", + Language.GU: "gu-in", + Language.KN: "kn-in", + Language.ML: "ml-in", + Language.PA: "pa-in", + Language.UR: "ur-pk", + Language.FA: "fa-ir", + Language.TL: "tl-ph", + } + + return resolve_language(language, LANGUAGE_MAP, use_base_code=True) + + +class CambTTSService(TTSService): + """Camb.ai MARS-8 HTTP-based text-to-speech service. + + Converts text to speech using Camb.ai's MARS-8 TTS models with support for + multiple languages. Provides control over voice characteristics like speed + and custom instructions (for mars-8-instruct model). + + Example:: + + tts = CambTTSService( + api_key="your-api-key", + voice_id=2681, + model="mars-8-flash", + aiohttp_session=session, + params=CambTTSService.InputParams( + language=Language.EN, + speed=1.0 + ) + ) + + # For mars-8-instruct with custom instructions: + tts_instruct = CambTTSService( + api_key="your-api-key", + voice_id=2681, + model="mars-8-instruct", + aiohttp_session=session, + params=CambTTSService.InputParams( + language=Language.EN, + user_instructions="Speak with excitement and energy" + ) + ) + """ + + class InputParams(BaseModel): + """Input parameters for Camb.ai TTS configuration. + + Parameters: + language: Language for synthesis (BCP-47 format). Defaults to English. + speed: Speech speed multiplier (0.5 to 2.0). Defaults to 1.0. + user_instructions: Custom instructions for mars-8-instruct model only. + Ignored for other models. Max 1000 characters. + """ + + language: Optional[Language] = Language.EN + speed: Optional[float] = Field(default=1.0, ge=0.5, le=2.0) + user_instructions: Optional[str] = Field( + default=None, + max_length=1000, + description="Custom instructions for mars-8-instruct model only. " + "Use to control tone, style, or pronunciation. Max 1000 characters.", + ) + + def __init__( + self, + *, + api_key: str, + aiohttp_session: aiohttp.ClientSession, + voice_id: int = DEFAULT_VOICE_ID, + model: str = DEFAULT_MODEL, + base_url: str = DEFAULT_BASE_URL, + sample_rate: Optional[int] = None, + params: Optional[InputParams] = None, + **kwargs, + ): + """Initialize the Camb.ai TTS service. + + Args: + api_key: Camb.ai API key for authentication. + aiohttp_session: Shared aiohttp session for making HTTP requests. + voice_id: Voice ID to use (e.g., 2681 for Attic). Defaults to 2681. + model: TTS model to use. Options: "mars-8", "mars-8-flash", "mars-8-instruct". + Defaults to "mars-8-flash" (fastest). + base_url: Camb.ai API base URL. Defaults to production URL. + sample_rate: Audio sample rate in Hz. If None, uses Camb.ai default (24000). + params: Additional voice parameters. If None, uses defaults. + **kwargs: Additional arguments passed to parent TTSService. + """ + super().__init__(sample_rate=sample_rate, **kwargs) + + params = params or CambTTSService.InputParams() + + self._api_key = api_key + self._session = aiohttp_session + + # Remove trailing slash from base URL + if base_url.endswith("/"): + logger.warning("Base URL ends with a slash, removing it.") + base_url = base_url[:-1] + + self._base_url = base_url + + # Build settings + self._settings = { + "language": ( + self.language_to_service_language(params.language) + if params.language + else DEFAULT_LANGUAGE + ), + "speed": params.speed or 1.0, + "user_instructions": params.user_instructions, + } + + self.set_model_name(model) + self.set_voice(str(voice_id)) + self._voice_id_int = voice_id + + def can_generate_metrics(self) -> bool: + """Check if this service can generate processing metrics. + + Returns: + True, as Camb.ai service supports metrics generation. + """ + return True + + def language_to_service_language(self, language: Language) -> Optional[str]: + """Convert a Language enum to Camb.ai language format. + + Args: + language: The language to convert. + + Returns: + The Camb.ai-specific language code, or None if not supported. + """ + return language_to_camb_language(language) + + async def start(self, frame: StartFrame): + """Start the Camb.ai TTS service. + + Args: + frame: The start frame containing initialization parameters. + """ + await super().start(frame) + # Use Camb.ai's native sample rate if not specified + if not self._init_sample_rate: + self._sample_rate = DEFAULT_SAMPLE_RATE + self._settings["sample_rate"] = self._sample_rate + + async def _update_settings(self, settings: Mapping[str, Any]): + """Update service settings dynamically. + + Args: + settings: Dictionary of settings to update. + """ + await super()._update_settings(settings) + + for key, value in settings.items(): + if key in self._settings: + if key == "language" and isinstance(value, Language): + self._settings[key] = language_to_camb_language(value) + else: + self._settings[key] = value + logger.debug(f"Updated Camb.ai TTS setting {key} to: {value}") + elif key == "voice_id": + self._voice_id_int = int(value) + self.set_voice(str(value)) + + @traced_tts + async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + """Generate speech from text using Camb.ai's TTS API. + + Args: + text: The text to synthesize into speech (3-3000 characters). + + Yields: + Frame: Audio frames containing the synthesized speech. + """ + logger.debug(f"{self}: Generating TTS [{text}]") + + # Validate text length + if len(text) < MIN_TEXT_LENGTH: + logger.warning(f"Text too short for Camb.ai TTS (min {MIN_TEXT_LENGTH} chars): {text}") + yield TTSStoppedFrame() + return + + if len(text) > MAX_TEXT_LENGTH: + logger.warning( + f"Text too long for Camb.ai TTS (max {MAX_TEXT_LENGTH} chars), truncating" + ) + text = text[:MAX_TEXT_LENGTH] + + # Build request payload + payload = { + "text": text, + "voice_id": self._voice_id_int, + "language": self._settings["language"], + "speech_model": self._model_name, + "output_configuration": {"format": "pcm_s16le"}, + "voice_settings": {"speed": self._settings["speed"]}, + } + + # Add user instructions if using mars-8-instruct model + if self._model_name == "mars-8-instruct" and self._settings.get("user_instructions"): + payload["user_instructions"] = self._settings["user_instructions"] + + headers = { + "x-api-key": self._api_key, + "Accept": "application/json", + "Content-Type": "application/json", + } + + try: + await self.start_ttfb_metrics() + + async with self._session.post( + f"{self._base_url}/tts-stream", + json=payload, + headers=headers, + timeout=aiohttp.ClientTimeout(total=DEFAULT_TIMEOUT), + ) as response: + if response.status != 200: + error_text = await response.text() + error_msg = self._format_error_message(response.status, error_text) + logger.error(f"{self}: {error_msg}") + yield ErrorFrame(error=error_msg) + return + + await self.start_tts_usage_metrics(text) + yield TTSStartedFrame() + + CHUNK_SIZE = self.chunk_size + + async for frame in self._stream_audio_frames_from_iterator( + response.content.iter_chunked(CHUNK_SIZE), strip_wav_header=False + ): + await self.stop_ttfb_metrics() + yield frame + + except aiohttp.ClientError as e: + error_msg = f"Network error communicating with Camb.ai: {e}" + logger.error(f"{self}: {error_msg}") + yield ErrorFrame(error=error_msg) + except asyncio.TimeoutError: + error_msg = f"Timeout waiting for Camb.ai TTS response (>{DEFAULT_TIMEOUT}s)" + logger.error(f"{self}: {error_msg}") + yield ErrorFrame(error=error_msg) + except Exception as e: + error_msg = f"Unexpected error in Camb.ai TTS: {e}" + logger.error(f"{self}: {error_msg}") + yield ErrorFrame(error=error_msg) + finally: + logger.debug(f"{self}: Finished TTS [{text}]") + await self.stop_ttfb_metrics() + yield TTSStoppedFrame() + + def _format_error_message(self, status: int, error_text: str) -> str: + """Format error message based on HTTP status code. + + Args: + status: HTTP status code. + error_text: Error response body. + + Returns: + Formatted, user-friendly error message. + """ + if status == 401: + return ( + "Invalid Camb.ai API key. " + "Set CAMB_API_KEY environment variable with your API key from https://camb.ai" + ) + elif status == 403: + return ( + f"Voice ID {self._voice_id_int} is not accessible with your API key. " + "Use list_voices() to see available voices." + ) + elif status == 404: + return ( + f"Invalid voice ID: {self._voice_id_int}. " + "Use list_voices() to see available voices." + ) + elif status == 429: + return "Camb.ai rate limit exceeded. Please wait before making more requests." + elif status >= 500: + return f"Camb.ai server error (status {status}): {error_text}" + else: + return f"Camb.ai API error (status {status}): {error_text}" + + @staticmethod + async def list_voices( + api_key: str, + aiohttp_session: aiohttp.ClientSession, + base_url: str = DEFAULT_BASE_URL, + ) -> List[Dict[str, Any]]: + """Fetch available voices from Camb.ai API. + + Args: + api_key: Camb.ai API key for authentication. + aiohttp_session: aiohttp ClientSession for making HTTP requests. + base_url: Camb.ai API base URL. + + Returns: + List of voice dictionaries with id, name, gender, and language fields. + + Raises: + Exception: If the API request fails. + + Example:: + + async with aiohttp.ClientSession() as session: + voices = await CambTTSService.list_voices( + api_key="your-api-key", + aiohttp_session=session, + ) + for voice in voices: + print(f"{voice['id']}: {voice['name']}") + """ + if base_url.endswith("/"): + base_url = base_url[:-1] + + headers = { + "x-api-key": api_key, + "Accept": "application/json", + } + + gender_map = { + 0: "Not Specified", + 1: "Male", + 2: "Female", + 9: "Not Applicable", + } + + async with aiohttp_session.get( + f"{base_url}/list-voices", + headers=headers, + timeout=aiohttp.ClientTimeout(total=30.0), + ) as response: + if response.status != 200: + error_text = await response.text() + raise Exception(f"Failed to list voices (status {response.status}): {error_text}") + + data = await response.json() + return [ + { + "id": v["id"], + "name": v.get("voice_name", "Unknown"), + "gender": gender_map.get(v.get("gender"), "Unknown"), + "age": v.get("age"), + "language": v.get("language"), + } + for v in data + ] diff --git a/tests/test_camb_tts.py b/tests/test_camb_tts.py new file mode 100644 index 000000000..71cc99c99 --- /dev/null +++ b/tests/test_camb_tts.py @@ -0,0 +1,431 @@ +# +# Copyright (c) 2024-2025 Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Tests for CambTTSService. + +These tests use mock servers to simulate the Camb.ai API responses. +""" + +import asyncio + +import aiohttp +import pytest +from aiohttp import web + +from pipecat.frames.frames import ( + AggregatedTextFrame, + ErrorFrame, + TTSAudioRawFrame, + TTSSpeakFrame, + TTSStartedFrame, + TTSStoppedFrame, + TTSTextFrame, +) +from pipecat.services.camb.tts import CambTTSService, language_to_camb_language +from pipecat.tests.utils import run_test +from pipecat.transcriptions.language import Language + + +@pytest.mark.asyncio +async def test_run_camb_tts_success(aiohttp_client): + """Test successful TTS generation with chunked PCM audio. + + Verifies the frame sequence: TTSStartedFrame -> TTSAudioRawFrame* -> TTSStoppedFrame + """ + + async def handler(request): + # Verify request headers + assert request.headers.get("x-api-key") == "test-api-key" + assert request.headers.get("Content-Type") == "application/json" + + # Parse and verify request body + body = await request.json() + assert "text" in body + assert body["voice_id"] == 2681 + assert body["language"] == "en-us" + assert body["speech_model"] == "mars-8-flash" + assert body["output_configuration"]["format"] == "pcm_s16le" + + # Prepare a StreamResponse with chunked PCM data + resp = web.StreamResponse( + status=200, + reason="OK", + headers={"Content-Type": "audio/raw"}, + ) + await resp.prepare(request) + + # Write out chunked PCM byte data (16-bit samples) + # Use smaller chunks for more predictable frame count + data = b"\x00\x01" * 4800 # Small chunk of audio + await resp.write(data) + await resp.write_eof() + + return resp + + # Create an aiohttp test server + app = web.Application() + app.router.add_post("/tts-stream", handler) + client = await aiohttp_client(app) + + base_url = str(client.make_url("")).rstrip("/") + + async with aiohttp.ClientSession() as session: + tts_service = CambTTSService( + api_key="test-api-key", + aiohttp_session=session, + base_url=base_url, + voice_id=2681, + model="mars-8-flash", + ) + + # Manually set sample rate (normally done by StartFrame) + tts_service._sample_rate = 24000 + + # Test run_tts directly to avoid frame count variability + text = "Hello world, this is a test." + frames = [] + async for frame in tts_service.run_tts(text): + frames.append(frame) + + # Verify we got the expected frame types + frame_types = [type(f).__name__ for f in frames] + assert "TTSStartedFrame" in frame_types, "Should have TTSStartedFrame" + assert "TTSAudioRawFrame" in frame_types, "Should have TTSAudioRawFrame" + assert "TTSStoppedFrame" in frame_types, "Should have TTSStoppedFrame" + + audio_frames = [f for f in frames if isinstance(f, TTSAudioRawFrame)] + assert len(audio_frames) > 0, "Should have at least one audio frame" + + # Verify sample rate matches Camb.ai's output (24kHz) + for a_frame in audio_frames: + assert a_frame.sample_rate == 24000, "Sample rate should be 24000 Hz" + assert a_frame.num_channels == 1, "Should be mono audio" + + +@pytest.mark.asyncio +async def test_run_camb_tts_error_401(aiohttp_client): + """Test handling of invalid API key (401 Unauthorized).""" + + async def handler(request): + return web.Response( + status=401, + text="Unauthorized: Invalid API key", + ) + + app = web.Application() + app.router.add_post("/tts-stream", handler) + client = await aiohttp_client(app) + base_url = str(client.make_url("")).rstrip("/") + + async with aiohttp.ClientSession() as session: + tts_service = CambTTSService( + api_key="invalid-key", + aiohttp_session=session, + base_url=base_url, + ) + + frames_to_send = [ + TTSSpeakFrame(text="This should fail."), + ] + + expected_down_frames = [AggregatedTextFrame, TTSStoppedFrame, TTSTextFrame] + expected_up_frames = [ErrorFrame] + + frames_received = await run_test( + tts_service, + frames_to_send=frames_to_send, + expected_down_frames=expected_down_frames, + expected_up_frames=expected_up_frames, + ) + up_frames = frames_received[1] + + assert isinstance(up_frames[0], ErrorFrame), "Must receive an ErrorFrame for 401" + assert "Invalid Camb.ai API key" in up_frames[0].error, ( + "ErrorFrame should mention invalid API key" + ) + + +@pytest.mark.asyncio +async def test_run_camb_tts_error_404(aiohttp_client): + """Test handling of invalid voice ID (404 Not Found).""" + + async def handler(request): + return web.Response( + status=404, + text="Voice not found", + ) + + app = web.Application() + app.router.add_post("/tts-stream", handler) + client = await aiohttp_client(app) + base_url = str(client.make_url("")).rstrip("/") + + async with aiohttp.ClientSession() as session: + tts_service = CambTTSService( + api_key="test-api-key", + aiohttp_session=session, + base_url=base_url, + voice_id=99999, # Invalid voice ID + ) + + frames_to_send = [ + TTSSpeakFrame(text="This should fail."), + ] + + expected_down_frames = [AggregatedTextFrame, TTSStoppedFrame, TTSTextFrame] + expected_up_frames = [ErrorFrame] + + frames_received = await run_test( + tts_service, + frames_to_send=frames_to_send, + expected_down_frames=expected_down_frames, + expected_up_frames=expected_up_frames, + ) + up_frames = frames_received[1] + + assert isinstance(up_frames[0], ErrorFrame), "Must receive an ErrorFrame for 404" + assert "Invalid voice ID" in up_frames[0].error, ( + "ErrorFrame should mention invalid voice ID" + ) + + +@pytest.mark.asyncio +async def test_run_camb_tts_error_429(aiohttp_client): + """Test handling of rate limit (429 Too Many Requests).""" + + async def handler(request): + return web.Response( + status=429, + text="Rate limit exceeded", + ) + + app = web.Application() + app.router.add_post("/tts-stream", handler) + client = await aiohttp_client(app) + base_url = str(client.make_url("")).rstrip("/") + + async with aiohttp.ClientSession() as session: + tts_service = CambTTSService( + api_key="test-api-key", + aiohttp_session=session, + base_url=base_url, + ) + + frames_to_send = [ + TTSSpeakFrame(text="This should fail due to rate limit."), + ] + + expected_down_frames = [AggregatedTextFrame, TTSStoppedFrame, TTSTextFrame] + expected_up_frames = [ErrorFrame] + + frames_received = await run_test( + tts_service, + frames_to_send=frames_to_send, + expected_down_frames=expected_down_frames, + expected_up_frames=expected_up_frames, + ) + up_frames = frames_received[1] + + assert isinstance(up_frames[0], ErrorFrame), "Must receive an ErrorFrame for 429" + assert "rate limit" in up_frames[0].error.lower(), ( + "ErrorFrame should mention rate limit" + ) + + +@pytest.mark.asyncio +async def test_list_voices(aiohttp_client): + """Test voice listing endpoint.""" + + async def handler(request): + # Verify API key header + assert request.headers.get("x-api-key") == "test-api-key" + + # Return mock voice data (matching actual API response structure) + voices = [ + { + "id": 2681, + "voice_name": "Attic", + "gender": 1, + "age": 25, + "language": None, + "transcript": None, + "description": None, + "is_published": None, + }, + { + "id": 2682, + "voice_name": "Cellar", + "gender": 2, + "age": 30, + "language": 1, + "transcript": None, + "description": None, + "is_published": False, + }, + ] + return web.json_response(voices) + + app = web.Application() + app.router.add_get("/list-voices", handler) + client = await aiohttp_client(app) + base_url = str(client.make_url("")).rstrip("/") + + async with aiohttp.ClientSession() as session: + voices = await CambTTSService.list_voices( + api_key="test-api-key", + aiohttp_session=session, + base_url=base_url, + ) + + # Should return all voices + assert len(voices) == 2, "Should return all voices" + + # Verify voice data structure + attic_voice = next(v for v in voices if v["id"] == 2681) + assert attic_voice["name"] == "Attic" + assert attic_voice["gender"] == "Male" + assert attic_voice["age"] == 25 + + +@pytest.mark.asyncio +async def test_text_length_validation_too_short(aiohttp_client): + """Test that text shorter than 3 characters is handled gracefully.""" + + async def handler(request): + # This should not be called for short text + pytest.fail("Handler should not be called for text < 3 chars") + + app = web.Application() + app.router.add_post("/tts-stream", handler) + client = await aiohttp_client(app) + base_url = str(client.make_url("")).rstrip("/") + + async with aiohttp.ClientSession() as session: + tts_service = CambTTSService( + api_key="test-api-key", + aiohttp_session=session, + base_url=base_url, + ) + + frames_to_send = [ + TTSSpeakFrame(text="Hi"), # Only 2 characters + ] + + # For short text, we expect TTSStoppedFrame but no audio + expected_down_frames = [AggregatedTextFrame, TTSStoppedFrame, TTSTextFrame] + + frames_received = await run_test( + tts_service, + frames_to_send=frames_to_send, + expected_down_frames=expected_down_frames, + ) + down_frames = frames_received[0] + + # Verify no audio frames were generated + audio_frames = [f for f in down_frames if isinstance(f, TTSAudioRawFrame)] + assert len(audio_frames) == 0, "Should not generate audio for text < 3 chars" + + +@pytest.mark.asyncio +async def test_input_params(): + """Test InputParams model validation and defaults.""" + + # Test defaults + params = CambTTSService.InputParams() + assert params.language == Language.EN + assert params.speed == 1.0 + assert params.user_instructions is None + + # Test custom values + params = CambTTSService.InputParams( + language=Language.ES, + speed=1.5, + user_instructions="Speak slowly and clearly", + ) + assert params.language == Language.ES + assert params.speed == 1.5 + assert params.user_instructions == "Speak slowly and clearly" + + +@pytest.mark.asyncio +async def test_language_mapping(): + """Test language enum to Camb.ai language code conversion.""" + + # Test common languages + assert language_to_camb_language(Language.EN) == "en-us" + assert language_to_camb_language(Language.EN_US) == "en-us" + assert language_to_camb_language(Language.EN_GB) == "en-gb" + assert language_to_camb_language(Language.ES) == "es-es" + assert language_to_camb_language(Language.FR) == "fr-fr" + assert language_to_camb_language(Language.DE) == "de-de" + assert language_to_camb_language(Language.JA) == "ja-jp" + assert language_to_camb_language(Language.ZH) == "zh-cn" + + +@pytest.mark.asyncio +async def test_mars8_instruct_model(aiohttp_client): + """Test that user_instructions are included for mars-8-instruct model.""" + + received_payload = {} + + async def handler(request): + nonlocal received_payload + received_payload = await request.json() + + # Return minimal successful response + resp = web.StreamResponse(status=200, headers={"Content-Type": "audio/raw"}) + await resp.prepare(request) + await resp.write(b"\x00" * 1000) + await resp.write_eof() + return resp + + app = web.Application() + app.router.add_post("/tts-stream", handler) + client = await aiohttp_client(app) + base_url = str(client.make_url("")).rstrip("/") + + async with aiohttp.ClientSession() as session: + tts_service = CambTTSService( + api_key="test-api-key", + aiohttp_session=session, + base_url=base_url, + model="mars-8-instruct", + params=CambTTSService.InputParams(user_instructions="Speak with excitement"), + ) + + frames_to_send = [ + TTSSpeakFrame(text="This is exciting news!"), + ] + + await run_test( + tts_service, + frames_to_send=frames_to_send, + expected_down_frames=[ + AggregatedTextFrame, + TTSStartedFrame, + TTSAudioRawFrame, + TTSStoppedFrame, + TTSTextFrame, + ], + ) + + # Verify user_instructions was included in the request + assert received_payload.get("speech_model") == "mars-8-instruct" + assert received_payload.get("user_instructions") == "Speak with excitement" + + +@pytest.mark.asyncio +async def test_base_url_trailing_slash(): + """Test that trailing slash in base URL is handled correctly.""" + async with aiohttp.ClientSession() as session: + tts = CambTTSService( + api_key="test-key", + aiohttp_session=session, + base_url="https://api.example.com/", # With trailing slash + ) + + # Should have removed the trailing slash + assert tts._base_url == "https://api.example.com" diff --git a/uv.lock b/uv.lock index a9b3d2dd4..34811146a 100644 --- a/uv.lock +++ b/uv.lock @@ -4259,6 +4259,9 @@ aws-nova-sonic = [ azure = [ { name = "azure-cognitiveservices-speech" }, ] +camb = [ + { name = "websockets" }, +] cartesia = [ { name = "cartesia" }, { name = "websockets" }, @@ -4525,6 +4528,7 @@ requires-dist = [ { name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'assemblyai'" }, { name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'asyncai'" }, { name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'aws'" }, + { name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'camb'" }, { name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'cartesia'" }, { name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'deepgram'" }, { name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'elevenlabs'" }, @@ -4573,7 +4577,7 @@ requires-dist = [ { name = "wait-for2", marker = "python_full_version < '3.12'", specifier = ">=0.4.1" }, { name = "websockets", marker = "extra == 'websockets-base'", specifier = ">=13.1,<16.0" }, ] -provides-extras = ["aic", "anthropic", "assemblyai", "asyncai", "aws", "aws-nova-sonic", "azure", "cartesia", "cerebras", "daily", "deepgram", "deepseek", "elevenlabs", "fal", "fireworks", "fish", "gladia", "google", "gradium", "grok", "groq", "gstreamer", "heygen", "hume", "inworld", "koala", "krisp", "langchain", "livekit", "lmnt", "local", "local-smart-turn", "local-smart-turn-v3", "mcp", "mem0", "mistral", "mlx-whisper", "moondream", "neuphonic", "noisereduce", "nvidia", "openai", "rnnoise", "openpipe", "openrouter", "perplexity", "playht", "qwen", "remote-smart-turn", "rime", "riva", "runner", "sagemaker", "sambanova", "sarvam", "sentry", "silero", "simli", "soniox", "soundfile", "speechmatics", "strands", "tavus", "together", "tracing", "ultravox", "webrtc", "websocket", "websockets-base", "whisper"] +provides-extras = ["aic", "anthropic", "assemblyai", "asyncai", "aws", "aws-nova-sonic", "azure", "cartesia", "camb", "cerebras", "daily", "deepgram", "deepseek", "elevenlabs", "fal", "fireworks", "fish", "gladia", "google", "gradium", "grok", "groq", "gstreamer", "heygen", "hume", "inworld", "koala", "krisp", "langchain", "livekit", "lmnt", "local", "local-smart-turn", "local-smart-turn-v3", "mcp", "mem0", "mistral", "mlx-whisper", "moondream", "neuphonic", "noisereduce", "nvidia", "openai", "rnnoise", "openpipe", "openrouter", "perplexity", "playht", "qwen", "remote-smart-turn", "rime", "riva", "runner", "sagemaker", "sambanova", "sarvam", "sentry", "silero", "simli", "soniox", "soundfile", "speechmatics", "strands", "tavus", "together", "tracing", "ultravox", "webrtc", "websocket", "websockets-base", "whisper"] [package.metadata.requires-dev] dev = [