From 9293b5f24a3b08b40eee767be38a36dd3786f6aa Mon Sep 17 00:00:00 2001 From: Neil Ruaro Date: Mon, 12 Jan 2026 19:23:43 +0900 Subject: [PATCH] Migrate Camb TTS service from raw HTTP to official SDK - Replace aiohttp with camb SDK (AsyncCambAI client) - Add support for passing existing SDK client instance - Simplify API: no longer requires aiohttp_session parameter - Update example to use simplified initialization - Rewrite tests to mock SDK client instead of HTTP servers --- .../07zb-interruptible-camb-local.py | 114 +++--- src/pipecat/services/camb/tts.py | 229 +++++------- tests/test_camb_tts.py | 334 ++++++------------ 3 files changed, 242 insertions(+), 435 deletions(-) diff --git a/examples/foundational/07zb-interruptible-camb-local.py b/examples/foundational/07zb-interruptible-camb-local.py index 19ef0bbed..6ddea3db0 100644 --- a/examples/foundational/07zb-interruptible-camb-local.py +++ b/examples/foundational/07zb-interruptible-camb-local.py @@ -28,7 +28,6 @@ import asyncio import os import sys -import aiohttp from dotenv import load_dotenv from loguru import logger @@ -66,73 +65,70 @@ async def main(voice_id: int): # Deepgram STT for speech recognition stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) - # Create HTTP session for Camb.ai TTS - async with aiohttp.ClientSession() as session: - # Camb.ai TTS with MARS-flash model - tts = CambTTSService( - api_key=os.getenv("CAMB_API_KEY"), - aiohttp_session=session, - voice_id=voice_id, - model="mars-flash", - ) + # Camb.ai TTS with MARS-flash model (uses official SDK) + tts = CambTTSService( + api_key=os.getenv("CAMB_API_KEY"), + voice_id=voice_id, + model="mars-flash", + ) - # OpenAI LLM - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) + # OpenAI LLM + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) - # System prompt - messages = [ - { - "role": "system", - "content": """You are a helpful voice assistant powered by Camb.ai's MARS + # System prompt + messages = [ + { + "role": "system", + "content": """You are a helpful voice assistant powered by Camb.ai's MARS text-to-speech technology. Keep your responses concise and conversational since they will be spoken aloud. Avoid special characters, emojis, or bullet points.""", - }, + }, + ] + + # Context management + context = LLMContext(messages) + context_aggregator = LLMContextAggregatorPair(context) + + # Build the pipeline + pipeline = Pipeline( + [ + transport.input(), # Microphone input + stt, # Speech-to-text + context_aggregator.user(), # User context + llm, # Language model + tts, # Camb.ai TTS + transport.output(), # Speaker output + context_aggregator.assistant(), # Assistant context ] + ) - # Context management - context = LLMContext(messages) - context_aggregator = LLMContextAggregatorPair(context) + # Create pipeline task + # Use 24kHz sample rate to match Camb.ai TTS output + task = PipelineTask( + pipeline, + params=PipelineParams( + audio_out_sample_rate=24000, + enable_metrics=True, + enable_usage_metrics=True, + ), + ) - # Build the pipeline - pipeline = Pipeline( - [ - transport.input(), # Microphone input - stt, # Speech-to-text - context_aggregator.user(), # User context - llm, # Language model - tts, # Camb.ai TTS - transport.output(), # Speaker output - context_aggregator.assistant(), # Assistant context - ] + # Start the conversation when the pipeline is ready + @task.event_handler("on_pipeline_started") + async def on_pipeline_started(task, frame): + messages.append( + { + "role": "system", + "content": "Please introduce yourself briefly and ask how you can help.", + } ) + await task.queue_frames([LLMRunFrame()]) - # Create pipeline task - # Use 24kHz sample rate to match Camb.ai TTS output - task = PipelineTask( - pipeline, - params=PipelineParams( - audio_out_sample_rate=24000, - enable_metrics=True, - enable_usage_metrics=True, - ), - ) - - # Start the conversation when the pipeline is ready - @task.event_handler("on_pipeline_started") - async def on_pipeline_started(task, frame): - messages.append( - { - "role": "system", - "content": "Please introduce yourself briefly and ask how you can help.", - } - ) - await task.queue_frames([LLMRunFrame()]) - - # Run the pipeline - runner = PipelineRunner() - logger.info("Starting Camb.ai TTS bot with local audio...") - logger.info("Speak into your microphone to interact with the bot.") - await runner.run(task) + # Run the pipeline + runner = PipelineRunner() + logger.info("Starting Camb.ai TTS bot with local audio...") + logger.info("Speak into your microphone to interact with the bot.") + await runner.run(task) if __name__ == "__main__": diff --git a/src/pipecat/services/camb/tts.py b/src/pipecat/services/camb/tts.py index d0471d63d..efe9cbb8d 100644 --- a/src/pipecat/services/camb/tts.py +++ b/src/pipecat/services/camb/tts.py @@ -7,20 +7,20 @@ """Camb.ai MARS text-to-speech service implementation. This module provides TTS functionality using Camb.ai's MARS model family, -offering high-quality text-to-speech synthesis with HTTP streaming support. +offering high-quality text-to-speech synthesis with streaming support. Features: - MARS models: mars-flash, mars-pro, mars-instruct - 140+ languages supported - - Real-time HTTP streaming + - Real-time streaming via official SDK - 24kHz audio output - Voice customization (instructions for mars-instruct) """ -import asyncio from typing import Any, AsyncGenerator, Dict, List, Mapping, Optional -import aiohttp +from camb.client import AsyncCambAI +from camb import StreamTtsOutputConfiguration from loguru import logger from pydantic import BaseModel, Field @@ -41,7 +41,6 @@ from pipecat.utils.tracing.service_decorators import traced_tts DEFAULT_VOICE_ID = 147320 DEFAULT_LANGUAGE = "en-us" DEFAULT_MODEL = "mars-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 @@ -126,29 +125,36 @@ def language_to_camb_language(language: Language) -> Optional[str]: class CambTTSService(TTSService): - """Camb.ai MARS HTTP-based text-to-speech service. + """Camb.ai MARS text-to-speech service using the official SDK. Converts text to speech using Camb.ai's MARS TTS models with support for multiple languages. Provides custom instructions support for the mars-instruct model. Example:: + # Using API key (creates internal client) tts = CambTTSService( api_key="your-api-key", voice_id=147320, model="mars-flash", - aiohttp_session=session, params=CambTTSService.InputParams( language=Language.EN ) ) + # Using existing SDK client + client = AsyncCambAI(api_key="your-api-key") + tts = CambTTSService( + client=client, + voice_id=147320, + model="mars-flash", + ) + # For mars-instruct with custom instructions: tts_instruct = CambTTSService( api_key="your-api-key", voice_id=147320, model="mars-instruct", - aiohttp_session=session, params=CambTTSService.InputParams( language=Language.EN, user_instructions="Speak with excitement and energy" @@ -176,11 +182,10 @@ class CambTTSService(TTSService): def __init__( self, *, - api_key: str, - aiohttp_session: aiohttp.ClientSession, + api_key: Optional[str] = None, + client: Optional[AsyncCambAI] = None, 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, @@ -188,29 +193,29 @@ class CambTTSService(TTSService): """Initialize the Camb.ai TTS service. Args: - api_key: Camb.ai API key for authentication. - aiohttp_session: Shared aiohttp session for making HTTP requests. + api_key: Camb.ai API key for authentication. Required if client is not provided. + client: Existing AsyncCambAI client instance. If provided, api_key is ignored. voice_id: Voice ID to use. Defaults to 147320. model: TTS model to use. Options: "mars-flash", "mars-pro", "mars-instruct". Defaults to "mars-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) + if client is None and api_key is None: + raise ValueError("Either 'api_key' or 'client' must be provided") + 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 + # Use provided client or create one + if client is not None: + self._client = client + self._owns_client = False + else: + self._client = AsyncCambAI(api_key=api_key, timeout=DEFAULT_TIMEOUT) + self._owns_client = True # Build settings self._settings = { @@ -300,63 +305,37 @@ class CambTTSService(TTSService): ) 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"}, - } - - # Add user instructions if using mars-instruct model - if self._model_name == "mars-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 + # Build SDK parameters + tts_kwargs: Dict[str, Any] = { + "text": text, + "voice_id": self._voice_id_int, + "language": self._settings["language"], + "speech_model": self._model_name, + "output_configuration": StreamTtsOutputConfiguration(format="pcm_s16le"), + } - await self.start_tts_usage_metrics(text) - yield TTSStartedFrame() + # Add user instructions if using mars-instruct model + if self._model_name == "mars-instruct" and self._settings.get("user_instructions"): + tts_kwargs["user_instructions"] = self._settings["user_instructions"] - async for chunk in response.content.iter_chunked(self.chunk_size): - if chunk: - await self.stop_ttfb_metrics() - yield TTSAudioRawFrame( - audio=chunk, - sample_rate=self.sample_rate, - num_channels=1, - ) + await self.start_tts_usage_metrics(text) + yield TTSStartedFrame() + + # Stream audio chunks from SDK + async for chunk in self._client.text_to_speech.tts(**tts_kwargs): + if chunk: + await self.stop_ttfb_metrics() + yield TTSAudioRawFrame( + audio=chunk, + sample_rate=self.sample_rate, + num_channels=1, + ) - 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}" + error_msg = f"Camb.ai TTS error: {e}" logger.error(f"{self}: {error_msg}") yield ErrorFrame(error=error_msg) finally: @@ -364,74 +343,37 @@ class CambTTSService(TTSService): 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, + api_key: Optional[str] = None, + client: Optional[AsyncCambAI] = None, ) -> 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. + api_key: Camb.ai API key for authentication. Required if client is not provided. + client: Existing AsyncCambAI client instance. If provided, api_key is ignored. Returns: List of voice dictionaries with id, name, gender, and language fields. Raises: + ValueError: If neither api_key nor client is provided. 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] + # Using API key + voices = await CambTTSService.list_voices(api_key="your-api-key") + for voice in voices: + print(f"{voice['id']}: {voice['name']}") - headers = { - "x-api-key": api_key, - "Accept": "application/json", - } + # Using existing client + client = AsyncCambAI(api_key="your-api-key") + voices = await CambTTSService.list_voices(client=client) + """ + if client is None and api_key is None: + raise ValueError("Either 'api_key' or 'client' must be provided") gender_map = { 0: "Not Specified", @@ -440,23 +382,22 @@ class CambTTSService(TTSService): 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}") + # Use provided client or create a temporary one + if client is not None: + sdk_client = client + else: + sdk_client = AsyncCambAI(api_key=api_key) - 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 - ] + voices = await sdk_client.voice_cloning.list_voices() + return [ + { + "id": v.id if hasattr(v, "id") else v.get("id"), + "name": v.voice_name if hasattr(v, "voice_name") else v.get("voice_name", "Unknown"), + "gender": gender_map.get( + v.gender if hasattr(v, "gender") else v.get("gender"), "Unknown" + ), + "age": v.age if hasattr(v, "age") else v.get("age"), + "language": v.language if hasattr(v, "language") else v.get("language"), + } + for v in voices + ] diff --git a/tests/test_camb_tts.py b/tests/test_camb_tts.py index 03b4bee1f..571cf67d8 100644 --- a/tests/test_camb_tts.py +++ b/tests/test_camb_tts.py @@ -6,14 +6,12 @@ """Tests for CambTTSService. -These tests use mock servers to simulate the Camb.ai API responses. +These tests mock the Camb.ai SDK client to test the service behavior. """ -import asyncio +from unittest.mock import AsyncMock, MagicMock, patch -import aiohttp import pytest -from aiohttp import web from pipecat.frames.frames import ( AggregatedTextFrame, @@ -29,54 +27,30 @@ from pipecat.tests.utils import run_test from pipecat.transcriptions.language import Language +async def mock_tts_stream(*args, **kwargs): + """Mock TTS stream that yields audio chunks.""" + yield b"\x00\x01" * 4800 # Small chunk of PCM audio + + +async def mock_tts_stream_error(*args, **kwargs): + """Mock TTS stream that raises an error.""" + raise Exception("API error: Invalid API key") + yield # Make this a generator + + @pytest.mark.asyncio -async def test_run_camb_tts_success(aiohttp_client): +async def test_run_camb_tts_success(): """Test successful TTS generation with chunked PCM audio. Verifies the frame sequence: TTSStartedFrame -> TTSAudioRawFrame* -> TTSStoppedFrame """ + with patch("pipecat.services.camb.tts.AsyncCambAI") as MockAsyncCambAI: + mock_client = MagicMock() + mock_client.text_to_speech.tts = mock_tts_stream + MockAsyncCambAI.return_value = mock_client - 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-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-flash", ) @@ -106,32 +80,24 @@ async def test_run_camb_tts_success(aiohttp_client): @pytest.mark.asyncio -async def test_run_camb_tts_error_401(aiohttp_client): - """Test handling of invalid API key (401 Unauthorized).""" +async def test_run_camb_tts_error(): + """Test handling of TTS API errors.""" + with patch("pipecat.services.camb.tts.AsyncCambAI") as MockAsyncCambAI: + mock_client = MagicMock() + mock_client.text_to_speech.tts = mock_tts_stream_error + MockAsyncCambAI.return_value = mock_client - 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, + voice_id=147320, ) frames_to_send = [ TTSSpeakFrame(text="This should fail."), ] - expected_down_frames = [AggregatedTextFrame, TTSStoppedFrame, TTSTextFrame] + # TTSStartedFrame is emitted before we attempt to iterate the stream + expected_down_frames = [AggregatedTextFrame, TTSStartedFrame, TTSStoppedFrame, TTSTextFrame] expected_up_frames = [ErrorFrame] frames_received = await run_test( @@ -142,143 +108,38 @@ async def test_run_camb_tts_error_401(aiohttp_client): ) 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" - ) + assert isinstance(up_frames[0], ErrorFrame), "Must receive an ErrorFrame" + assert "error" in up_frames[0].error.lower(), "ErrorFrame should contain error message" @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): +async def test_list_voices(): """Test voice listing endpoint.""" - async def handler(request): - # Verify API key header - assert request.headers.get("x-api-key") == "test-api-key" + async def mock_list_voices(*args, **kwargs): + # Return mock Voice objects + mock_voice1 = MagicMock() + mock_voice1.id = 2681 + mock_voice1.voice_name = "Attic" + mock_voice1.gender = 1 + mock_voice1.age = 25 + mock_voice1.language = None - # 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) + mock_voice2 = MagicMock() + mock_voice2.id = 2682 + mock_voice2.voice_name = "Cellar" + mock_voice2.gender = 2 + mock_voice2.age = 30 + mock_voice2.language = 1 - app = web.Application() - app.router.add_get("/list-voices", handler) - client = await aiohttp_client(app) - base_url = str(client.make_url("")).rstrip("/") + return [mock_voice1, mock_voice2] - async with aiohttp.ClientSession() as session: - voices = await CambTTSService.list_voices( - api_key="test-api-key", - aiohttp_session=session, - base_url=base_url, - ) + with patch("pipecat.services.camb.tts.AsyncCambAI") as MockAsyncCambAI: + mock_client = MagicMock() + mock_client.voice_cloning.list_voices = mock_list_voices + MockAsyncCambAI.return_value = mock_client + + voices = await CambTTSService.list_voices(api_key="test-api-key") # Should return all voices assert len(voices) == 2, "Should return all voices" @@ -291,23 +152,17 @@ async def test_list_voices(aiohttp_client): @pytest.mark.asyncio -async def test_text_length_validation_too_short(aiohttp_client): +async def test_text_length_validation_too_short(): """Test that text shorter than 3 characters is handled gracefully.""" + with patch("pipecat.services.camb.tts.AsyncCambAI") as MockAsyncCambAI: + mock_client = MagicMock() + # TTS should not be called for short text + mock_client.text_to_speech.tts = AsyncMock(side_effect=AssertionError("TTS should not be called")) + MockAsyncCambAI.return_value = mock_client - 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, + voice_id=147320, ) frames_to_send = [ @@ -363,32 +218,22 @@ async def test_language_mapping(): @pytest.mark.asyncio -async def test_mars_instruct_model(aiohttp_client): +async def test_mars_instruct_model(): """Test that user_instructions are included for mars-instruct model.""" + received_kwargs = {} - received_payload = {} + async def mock_tts_with_capture(*args, **kwargs): + nonlocal received_kwargs + received_kwargs = kwargs + yield b"\x00" * 1000 - async def handler(request): - nonlocal received_payload - received_payload = await request.json() + with patch("pipecat.services.camb.tts.AsyncCambAI") as MockAsyncCambAI: + mock_client = MagicMock() + mock_client.text_to_speech.tts = mock_tts_with_capture + MockAsyncCambAI.return_value = mock_client - # 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-instruct", params=CambTTSService.InputParams(user_instructions="Speak with excitement"), ) @@ -410,19 +255,44 @@ async def test_mars_instruct_model(aiohttp_client): ) # Verify user_instructions was included in the request - assert received_payload.get("speech_model") == "mars-instruct" - assert received_payload.get("user_instructions") == "Speak with excitement" + assert received_kwargs.get("speech_model") == "mars-instruct" + assert received_kwargs.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: +async def test_client_initialization_with_api_key(): + """Test that client is created when api_key is provided.""" + with patch("pipecat.services.camb.tts.AsyncCambAI") as MockAsyncCambAI: + mock_client = MagicMock() + MockAsyncCambAI.return_value = mock_client + tts = CambTTSService( api_key="test-key", - aiohttp_session=session, - base_url="https://api.example.com/", # With trailing slash + voice_id=147320, ) - # Should have removed the trailing slash - assert tts._base_url == "https://api.example.com" + # Should have created a client + MockAsyncCambAI.assert_called_once() + assert tts._owns_client is True + + +@pytest.mark.asyncio +async def test_client_initialization_with_existing_client(): + """Test that existing client is used when provided.""" + mock_client = MagicMock() + + tts = CambTTSService( + client=mock_client, + voice_id=147320, + ) + + # Should use the provided client + assert tts._client is mock_client + assert tts._owns_client is False + + +@pytest.mark.asyncio +async def test_client_initialization_requires_api_key_or_client(): + """Test that ValueError is raised when neither api_key nor client is provided.""" + with pytest.raises(ValueError, match="Either 'api_key' or 'client' must be provided"): + CambTTSService(voice_id=147320)