From 641d17007fc2f68d79b308327c3f33ba22e421a5 Mon Sep 17 00:00:00 2001 From: Neil Ruaro Date: Mon, 12 Jan 2026 22:20:42 +0900 Subject: [PATCH] Clean up Camb TTS service and tests --- .../07zb-interruptible-camb-local.py | 4 + src/pipecat/services/camb/tts.py | 111 ++++------ tests/test_camb_tts.py | 199 +++++++++++++----- 3 files changed, 185 insertions(+), 129 deletions(-) diff --git a/examples/foundational/07zb-interruptible-camb-local.py b/examples/foundational/07zb-interruptible-camb-local.py index 6ddea3db0..83e4ce0dc 100644 --- a/examples/foundational/07zb-interruptible-camb-local.py +++ b/examples/foundational/07zb-interruptible-camb-local.py @@ -34,6 +34,8 @@ 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.metrics.metrics import TTFBMetricsData +from pipecat.observers.loggers.metrics_log_observer import MetricsLogObserver from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -104,6 +106,7 @@ they will be spoken aloud. Avoid special characters, emojis, or bullet points."" # Create pipeline task # Use 24kHz sample rate to match Camb.ai TTS output + # Add MetricsLogObserver to track TTFB metrics task = PipelineTask( pipeline, params=PipelineParams( @@ -111,6 +114,7 @@ they will be spoken aloud. Avoid special characters, emojis, or bullet points."" enable_metrics=True, enable_usage_metrics=True, ), + observers=[MetricsLogObserver(include_metrics={TTFBMetricsData})], ) # Start the conversation when the pipeline is ready diff --git a/src/pipecat/services/camb/tts.py b/src/pipecat/services/camb/tts.py index efe9cbb8d..d5d800034 100644 --- a/src/pipecat/services/camb/tts.py +++ b/src/pipecat/services/camb/tts.py @@ -46,6 +46,9 @@ DEFAULT_TIMEOUT = 60.0 # Seconds (minimum recommended by Camb.ai) MIN_TEXT_LENGTH = 3 MAX_TEXT_LENGTH = 3000 +# Gender mapping for voice listing +GENDER_MAP = {0: "Not Specified", 1: "Male", 2: "Female", 9: "Not Applicable"} + def language_to_camb_language(language: Language) -> Optional[str]: """Convert a Pipecat Language enum to Camb.ai language code. @@ -132,31 +135,21 @@ class CambTTSService(TTSService): Example:: - # Using API key (creates internal client) + # Basic usage with defaults + tts = CambTTSService(api_key="your-api-key") + + # With custom voice and model tts = CambTTSService( api_key="your-api-key", - voice_id=147320, - model="mars-flash", - 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", + voice_id=12345, + model="mars-pro", ) # For mars-instruct with custom instructions: - tts_instruct = CambTTSService( + tts = CambTTSService( api_key="your-api-key", - voice_id=147320, model="mars-instruct", params=CambTTSService.InputParams( - language=Language.EN, user_instructions="Speak with excitement and energy" ) ) @@ -182,10 +175,10 @@ class CambTTSService(TTSService): def __init__( self, *, - api_key: Optional[str] = None, - client: Optional[AsyncCambAI] = None, + api_key: str, voice_id: int = DEFAULT_VOICE_ID, model: str = DEFAULT_MODEL, + timeout: float = DEFAULT_TIMEOUT, sample_rate: Optional[int] = None, params: Optional[InputParams] = None, **kwargs, @@ -193,29 +186,20 @@ class CambTTSService(TTSService): """Initialize the Camb.ai TTS service. Args: - 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. + api_key: Camb.ai API key for authentication. + voice_id: Voice ID to use. Defaults to DEFAULT_VOICE_ID. model: TTS model to use. Options: "mars-flash", "mars-pro", "mars-instruct". - Defaults to "mars-flash" (fastest). - sample_rate: Audio sample rate in Hz. If None, uses Camb.ai default (24000). + Defaults to DEFAULT_MODEL (mars-flash, fastest). + timeout: Request timeout in seconds. Defaults to DEFAULT_TIMEOUT (60s). + sample_rate: Audio sample rate in Hz. If None, uses DEFAULT_SAMPLE_RATE (24kHz). 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() - # 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 + self._client = AsyncCambAI(api_key=api_key, timeout=timeout) # Build settings self._settings = { @@ -344,60 +328,43 @@ class CambTTSService(TTSService): yield TTSStoppedFrame() @staticmethod - async def list_voices( - api_key: Optional[str] = None, - client: Optional[AsyncCambAI] = None, - ) -> List[Dict[str, Any]]: + async def list_voices(api_key: str) -> List[Dict[str, Any]]: """Fetch available voices from Camb.ai API. Args: - 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. + api_key: Camb.ai API key for authentication. 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:: - # Using API key voices = await CambTTSService.list_voices(api_key="your-api-key") for voice in voices: print(f"{voice['id']}: {voice['name']}") - - # 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") + client = AsyncCambAI(api_key=api_key) + voice_list = await client.voice_cloning.list_voices() - gender_map = { - 0: "Not Specified", - 1: "Male", - 2: "Female", - 9: "Not Applicable", - } + voices = [] + for voice in voice_list: + voice_id = voice.get("id") + # Skip voices without an ID + if voice_id is None: + continue - # Use provided client or create a temporary one - if client is not None: - sdk_client = client - else: - sdk_client = AsyncCambAI(api_key=api_key) + gender_int = voice.get("gender") + gender = GENDER_MAP.get(gender_int) if gender_int is not None else None - 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 - ] + voices.append({ + "id": voice_id, + "name": voice.get("voice_name", ""), + "gender": gender, + "age": voice.get("age"), + "language": voice.get("language"), + }) + + return voices diff --git a/tests/test_camb_tts.py b/tests/test_camb_tts.py index 571cf67d8..44319db32 100644 --- a/tests/test_camb_tts.py +++ b/tests/test_camb_tts.py @@ -9,6 +9,7 @@ These tests mock the Camb.ai SDK client to test the service behavior. """ +import asyncio from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -22,7 +23,12 @@ from pipecat.frames.frames import ( TTSStoppedFrame, TTSTextFrame, ) -from pipecat.services.camb.tts import CambTTSService, language_to_camb_language +from pipecat.services.camb.tts import ( + CambTTSService, + DEFAULT_SAMPLE_RATE, + DEFAULT_VOICE_ID, + language_to_camb_language, +) from pipecat.tests.utils import run_test from pipecat.transcriptions.language import Language @@ -49,14 +55,10 @@ async def test_run_camb_tts_success(): mock_client.text_to_speech.tts = mock_tts_stream MockAsyncCambAI.return_value = mock_client - tts_service = CambTTSService( - api_key="test-api-key", - voice_id=2681, - model="mars-flash", - ) + tts_service = CambTTSService(api_key="test-api-key") # Manually set sample rate (normally done by StartFrame) - tts_service._sample_rate = 24000 + tts_service._sample_rate = DEFAULT_SAMPLE_RATE # Test run_tts directly to avoid frame count variability text = "Hello world, this is a test." @@ -73,9 +75,9 @@ async def test_run_camb_tts_success(): 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) + # Verify sample rate matches Camb.ai's output for a_frame in audio_frames: - assert a_frame.sample_rate == 24000, "Sample rate should be 24000 Hz" + assert a_frame.sample_rate == DEFAULT_SAMPLE_RATE assert a_frame.num_channels == 1, "Should be mono audio" @@ -87,10 +89,7 @@ async def test_run_camb_tts_error(): mock_client.text_to_speech.tts = mock_tts_stream_error MockAsyncCambAI.return_value = mock_client - tts_service = CambTTSService( - api_key="invalid-key", - voice_id=147320, - ) + tts_service = CambTTSService(api_key="invalid-key") frames_to_send = [ TTSSpeakFrame(text="This should fail."), @@ -114,25 +113,26 @@ async def test_run_camb_tts_error(): @pytest.mark.asyncio async def test_list_voices(): - """Test voice listing endpoint.""" + """Test voice listing endpoint with dict responses.""" 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 - - mock_voice2 = MagicMock() - mock_voice2.id = 2682 - mock_voice2.voice_name = "Cellar" - mock_voice2.gender = 2 - mock_voice2.age = 30 - mock_voice2.language = 1 - - return [mock_voice1, mock_voice2] + # Return mock voice dicts (as returned by the API) + return [ + { + "id": 2681, + "voice_name": "Attic", + "gender": 1, + "age": 25, + "language": None, + }, + { + "id": 2682, + "voice_name": "Cellar", + "gender": 2, + "age": 30, + "language": "en-us", + }, + ] with patch("pipecat.services.camb.tts.AsyncCambAI") as MockAsyncCambAI: mock_client = MagicMock() @@ -151,6 +151,50 @@ async def test_list_voices(): assert attic_voice["age"] == 25 +@pytest.mark.asyncio +async def test_list_voices_skips_none_id(): + """Test that voices without an ID are skipped.""" + + async def mock_list_voices(*args, **kwargs): + return [ + {"id": 123, "voice_name": "Valid", "gender": 1, "age": 25, "language": None}, + {"id": None, "voice_name": "NoID", "gender": 2, "age": 30, "language": None}, + {"voice_name": "MissingID", "gender": 1, "age": 35, "language": None}, + ] + + 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 only return the voice with a valid ID + assert len(voices) == 1 + assert voices[0]["id"] == 123 + assert voices[0]["name"] == "Valid" + + +@pytest.mark.asyncio +async def test_list_voices_handles_none_gender(): + """Test that None gender is handled correctly.""" + + async def mock_list_voices(*args, **kwargs): + return [ + {"id": 123, "voice_name": "NoGender", "gender": None, "age": 25, "language": None}, + ] + + 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") + + assert len(voices) == 1 + assert voices[0]["gender"] is None + + @pytest.mark.asyncio async def test_text_length_validation_too_short(): """Test that text shorter than 3 characters is handled gracefully.""" @@ -160,10 +204,7 @@ async def test_text_length_validation_too_short(): mock_client.text_to_speech.tts = AsyncMock(side_effect=AssertionError("TTS should not be called")) MockAsyncCambAI.return_value = mock_client - tts_service = CambTTSService( - api_key="test-api-key", - voice_id=147320, - ) + tts_service = CambTTSService(api_key="test-api-key") frames_to_send = [ TTSSpeakFrame(text="Hi"), # Only 2 characters @@ -260,39 +301,83 @@ async def test_mars_instruct_model(): @pytest.mark.asyncio -async def test_client_initialization_with_api_key(): - """Test that client is created when api_key is provided.""" +async def test_client_initialization(): + """Test that client is created with correct parameters.""" with patch("pipecat.services.camb.tts.AsyncCambAI") as MockAsyncCambAI: mock_client = MagicMock() MockAsyncCambAI.return_value = mock_client - tts = CambTTSService( - api_key="test-key", - voice_id=147320, - ) + CambTTSService(api_key="test-key", timeout=120.0) - # Should have created a client - MockAsyncCambAI.assert_called_once() - assert tts._owns_client is True + # Should have created a client with correct params + MockAsyncCambAI.assert_called_once_with(api_key="test-key", timeout=120.0) @pytest.mark.asyncio -async def test_client_initialization_with_existing_client(): - """Test that existing client is used when provided.""" - mock_client = MagicMock() +async def test_default_voice_id_used(): + """Test that DEFAULT_VOICE_ID is used when not specified.""" + with patch("pipecat.services.camb.tts.AsyncCambAI") as MockAsyncCambAI: + mock_client = MagicMock() + MockAsyncCambAI.return_value = mock_client - tts = CambTTSService( - client=mock_client, - voice_id=147320, - ) + tts = CambTTSService(api_key="test-key") - # Should use the provided client - assert tts._client is mock_client - assert tts._owns_client is False + assert tts._voice_id_int == DEFAULT_VOICE_ID @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) +async def test_ttfb_metrics_tracked(): + """Test that TTFB metrics are properly tracked during TTS generation.""" + import time + + ttfb_start_called = False + ttfb_stop_called = False + start_time = None + stop_time = None + + async def mock_tts_with_delay(*args, **kwargs): + # Simulate some network delay + await asyncio.sleep(0.05) + yield b"\x00\x01" * 4800 + + with patch("pipecat.services.camb.tts.AsyncCambAI") as MockAsyncCambAI: + mock_client = MagicMock() + mock_client.text_to_speech.tts = mock_tts_with_delay + MockAsyncCambAI.return_value = mock_client + + tts_service = CambTTSService(api_key="test-api-key") + tts_service._sample_rate = DEFAULT_SAMPLE_RATE + + # Patch the metrics methods to track calls + original_start_ttfb = tts_service.start_ttfb_metrics + original_stop_ttfb = tts_service.stop_ttfb_metrics + + async def patched_start_ttfb(): + nonlocal ttfb_start_called, start_time + ttfb_start_called = True + start_time = time.time() + await original_start_ttfb() + + async def patched_stop_ttfb(): + nonlocal ttfb_stop_called, stop_time + if not ttfb_stop_called: # Only record first stop + ttfb_stop_called = True + stop_time = time.time() + await original_stop_ttfb() + + tts_service.start_ttfb_metrics = patched_start_ttfb + tts_service.stop_ttfb_metrics = patched_stop_ttfb + + # Run TTS + frames = [] + async for frame in tts_service.run_tts("Hello, this is a TTFB test."): + frames.append(frame) + + # Verify TTFB tracking was called + assert ttfb_start_called, "start_ttfb_metrics should be called" + assert ttfb_stop_called, "stop_ttfb_metrics should be called" + assert start_time is not None and stop_time is not None + + # TTFB should be >= simulated delay + ttfb = stop_time - start_time + assert ttfb >= 0.05, f"TTFB ({ttfb:.3f}s) should be >= 0.05s (simulated delay)"