diff --git a/examples/foundational/07zb-interruptible-camb-local.py b/examples/foundational/07zb-interruptible-camb-local.py index fa77cd2a3..de7af4488 100644 --- a/examples/foundational/07zb-interruptible-camb-local.py +++ b/examples/foundational/07zb-interruptible-camb-local.py @@ -20,9 +20,10 @@ 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 07zb-interruptible-camb-local.py + python 07zb-interruptible-camb-local.py [--voice-id VOICE_ID] """ +import argparse import asyncio import os import sys @@ -52,7 +53,7 @@ logger.remove(0) logger.add(sys.stderr, level="DEBUG") -async def main(): +async def main(voice_id: int): # Local audio transport - uses your microphone and speakers transport = LocalAudioTransport( LocalAudioTransportParams( @@ -71,7 +72,7 @@ async def main(): tts = CambTTSService( api_key=os.getenv("CAMB_API_KEY"), aiohttp_session=session, - voice_id=2681, # Attic voice + voice_id=voice_id, model="mars-8-flash", params=CambTTSService.InputParams( speed=1.0, @@ -109,9 +110,11 @@ 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 task = PipelineTask( pipeline, params=PipelineParams( + audio_out_sample_rate=24000, enable_metrics=True, enable_usage_metrics=True, ), @@ -141,4 +144,12 @@ they will be spoken aloud. Avoid special characters, emojis, or bullet points."" if __name__ == "__main__": - asyncio.run(main()) + parser = argparse.ArgumentParser(description="Camb.ai TTS example with local audio") + parser.add_argument( + "--voice-id", + type=int, + default=2681, + help="Camb.ai voice ID to use (default: 2681 - Attic voice)", + ) + args = parser.parse_args() + asyncio.run(main(args.voice_id)) diff --git a/examples/foundational/test_camb_quick.py b/examples/foundational/test_camb_quick.py deleted file mode 100644 index d178b2b7c..000000000 --- a/examples/foundational/test_camb_quick.py +++ /dev/null @@ -1,181 +0,0 @@ -#!/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)