added cambai tts integration

Author: eRuaro

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

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)

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]" ]