wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Address PR feedback: add --voice-id arg, remove test script

Browse Source 
- Add --voice-id CLI argument to example (default: 2681)
- Remove test_camb_quick.py from examples/ (tests belong in tests/)
- Update docstring with new usage
This commit is contained in:
Neil Ruaro
2026-01-05 20:34:05 +08:00
parent 54933bea2a
commit a3d7e9eafe
2 changed files with 15 additions and 185 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
examples/foundational/07zb-interruptible-camb-local.py
View File

@@ -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))

181
examples/foundational/test_camb_quick.py
View File

@@ -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)