diff --git a/CHANGELOG.md b/CHANGELOG.md index 3fd1a86d5..26187a4bb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added new fields to `PipelineParams` to control audio input and output sample + rates for the whole pipeline. This allows controlling sample rates from a + single place instead of having to specify sample rates in each + service. Setting a sample rate to a service is still possible and will + override the value from `PipelineParams`. + - Introduce audio resamplers (`BaseAudioResampler`). This is just a base class to implement audio resamplers. Currently, two implementations are provided `SOXRAudioResampler` and `ResampyResampler`. A new diff --git a/examples/bot-ready-signalling/server/signalling_bot.py b/examples/bot-ready-signalling/server/signalling_bot.py index 23d407c05..4d4d287b5 100644 --- a/examples/bot-ready-signalling/server/signalling_bot.py +++ b/examples/bot-ready-signalling/server/signalling_bot.py @@ -17,7 +17,7 @@ from runner import configure from pipecat.frames.frames import AudioRawFrame, EndFrame, OutputAudioRawFrame, TTSSpeakFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner -from pipecat.pipeline.task import PipelineTask +from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.services.cartesia import CartesiaTTSService from pipecat.transports.services.daily import DailyParams, DailyTransport @@ -31,16 +31,15 @@ logger.add(sys.stderr, level="DEBUG") class SilenceFrame(OutputAudioRawFrame): def __init__( self, - audio: bytes = None, - sample_rate: int = 16000, - num_channels: int = 1, - duration: float = 0.1, + *, + sample_rate: int, + duration: float, ): # Initialize the parent class with the silent frame's data super().__init__( - audio=self.create_silent_audio_frame(sample_rate, num_channels, duration).audio, + audio=self.create_silent_audio_frame(sample_rate, 1, duration).audio, sample_rate=sample_rate, - num_channels=num_channels, + num_channels=1, ) @staticmethod @@ -80,7 +79,10 @@ async def main(): return await task.queue_frames( [ - SilenceFrame(duration=0.5), + SilenceFrame( + sample_rate=task.params.audio_out_sample_rate, + duration=0.5, + ), TTSSpeakFrame(f"Hello there, how are you doing today ?"), EndFrame(), ] diff --git a/examples/foundational/07g-interruptible-openai-tts.py b/examples/foundational/07g-interruptible-openai-tts.py index 78a23b65f..ee3946cc7 100644 --- a/examples/foundational/07g-interruptible-openai-tts.py +++ b/examples/foundational/07g-interruptible-openai-tts.py @@ -37,7 +37,6 @@ async def main(): "Respond bot", DailyParams( audio_out_enabled=True, - audio_out_sample_rate=24000, transcription_enabled=True, vad_enabled=True, vad_analyzer=SileroVADAnalyzer(), diff --git a/examples/foundational/07k-interruptible-lmnt.py b/examples/foundational/07k-interruptible-lmnt.py index 4cfd76f24..a6fd3aff4 100644 --- a/examples/foundational/07k-interruptible-lmnt.py +++ b/examples/foundational/07k-interruptible-lmnt.py @@ -38,7 +38,6 @@ async def main(): "Respond bot", DailyParams( audio_out_enabled=True, - audio_out_sample_rate=24000, transcription_enabled=True, vad_enabled=True, vad_analyzer=SileroVADAnalyzer(), diff --git a/examples/foundational/07n-interruptible-google.py b/examples/foundational/07n-interruptible-google.py index 6c83825aa..cbeea755d 100644 --- a/examples/foundational/07n-interruptible-google.py +++ b/examples/foundational/07n-interruptible-google.py @@ -40,7 +40,6 @@ async def main(): "Respond bot", DailyParams( audio_out_enabled=True, - audio_out_sample_rate=24000, vad_enabled=True, vad_analyzer=SileroVADAnalyzer(), vad_audio_passthrough=True, diff --git a/examples/foundational/09-mirror.py b/examples/foundational/09-mirror.py index 3befece1b..669562742 100644 --- a/examples/foundational/09-mirror.py +++ b/examples/foundational/09-mirror.py @@ -21,7 +21,7 @@ from pipecat.frames.frames import ( ) from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner -from pipecat.pipeline.task import PipelineTask +from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.transports.services.daily import DailyParams, DailyTransport @@ -61,7 +61,6 @@ async def main(): "Test", DailyParams( audio_in_enabled=True, - audio_in_sample_rate=24000, audio_out_enabled=True, camera_out_enabled=True, camera_out_is_live=True, @@ -78,7 +77,9 @@ async def main(): runner = PipelineRunner() - task = PipelineTask(pipeline) + task = PipelineTask( + pipeline, PipelineParams(audio_in_sample_rate=24000, audio_out_sample_rate=24000) + ) await runner.run(task) diff --git a/examples/foundational/09a-local-mirror.py b/examples/foundational/09a-local-mirror.py index dd024d9bb..937ab4867 100644 --- a/examples/foundational/09a-local-mirror.py +++ b/examples/foundational/09a-local-mirror.py @@ -22,7 +22,7 @@ from pipecat.frames.frames import ( ) from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner -from pipecat.pipeline.task import PipelineTask +from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.transports.base_transport import TransportParams from pipecat.transports.local.tk import TkLocalTransport @@ -62,7 +62,7 @@ async def main(): tk_root.title("Local Mirror") daily_transport = DailyTransport( - room_url, token, "Test", DailyParams(audio_in_enabled=True, audio_in_sample_rate=24000) + room_url, token, "Test", DailyParams(audio_in_enabled=True) ) tk_transport = TkLocalTransport( @@ -82,7 +82,9 @@ async def main(): pipeline = Pipeline([daily_transport.input(), MirrorProcessor(), tk_transport.output()]) - task = PipelineTask(pipeline) + task = PipelineTask( + pipeline, PipelineParams(audio_in_sample_rate=24000, audio_out_sample_rate=24000) + ) async def run_tk(): while not task.has_finished(): diff --git a/examples/foundational/18-gstreamer-filesrc.py b/examples/foundational/18-gstreamer-filesrc.py index c61b3f6cb..b74846d53 100644 --- a/examples/foundational/18-gstreamer-filesrc.py +++ b/examples/foundational/18-gstreamer-filesrc.py @@ -51,8 +51,6 @@ async def main(): out_params=GStreamerPipelineSource.OutputParams( video_width=1280, video_height=720, - audio_sample_rate=24000, - audio_channels=1, ), ) diff --git a/examples/foundational/19-openai-realtime-beta.py b/examples/foundational/19-openai-realtime-beta.py index 9337c726e..609b5f036 100644 --- a/examples/foundational/19-openai-realtime-beta.py +++ b/examples/foundational/19-openai-realtime-beta.py @@ -80,9 +80,7 @@ async def main(): "Respond bot", DailyParams( audio_in_enabled=True, - audio_in_sample_rate=24000, audio_out_enabled=True, - audio_out_sample_rate=24000, transcription_enabled=False, vad_enabled=True, vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.8)), diff --git a/examples/foundational/20b-persistent-context-openai-realtime.py b/examples/foundational/20b-persistent-context-openai-realtime.py index e70543de2..3a1c43f47 100644 --- a/examples/foundational/20b-persistent-context-openai-realtime.py +++ b/examples/foundational/20b-persistent-context-openai-realtime.py @@ -177,9 +177,7 @@ async def main(): "Respond bot", DailyParams( audio_in_enabled=True, - audio_in_sample_rate=24000, audio_out_enabled=True, - audio_out_sample_rate=24000, transcription_enabled=False, vad_enabled=True, vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.8)), diff --git a/examples/foundational/21-tavus-layer.py b/examples/foundational/21-tavus-layer.py index 158c485d2..0cbc065a0 100644 --- a/examples/foundational/21-tavus-layer.py +++ b/examples/foundational/21-tavus-layer.py @@ -88,6 +88,10 @@ async def main(): task = PipelineTask( pipeline, PipelineParams( + # We just use 16000 because that's what Tavus is expecting and + # we avoid resampling. + audio_in_sample_rate=16000, + audio_out_sample_rate=16000, allow_interruptions=True, enable_metrics=True, enable_usage_metrics=True, diff --git a/examples/foundational/22d-natural-conversation-gemini-audio.py b/examples/foundational/22d-natural-conversation-gemini-audio.py index e21f93896..ffd7ca77c 100644 --- a/examples/foundational/22d-natural-conversation-gemini-audio.py +++ b/examples/foundational/22d-natural-conversation-gemini-audio.py @@ -639,7 +639,6 @@ async def main(): vad_enabled=True, vad_analyzer=SileroVADAnalyzer(), vad_audio_passthrough=True, - audio_in_sample_rate=16000, ), ) diff --git a/examples/foundational/26-gemini-multimodal-live.py b/examples/foundational/26-gemini-multimodal-live.py index 5f41833c8..62ea5d5f0 100644 --- a/examples/foundational/26-gemini-multimodal-live.py +++ b/examples/foundational/26-gemini-multimodal-live.py @@ -37,8 +37,6 @@ async def main(): token, "Respond bot", DailyParams( - audio_in_sample_rate=16000, - audio_out_sample_rate=24000, audio_out_enabled=True, vad_enabled=True, vad_audio_passthrough=True, diff --git a/examples/foundational/26a-gemini-multimodal-live-transcription.py b/examples/foundational/26a-gemini-multimodal-live-transcription.py index aae69e82c..ae48ce31b 100644 --- a/examples/foundational/26a-gemini-multimodal-live-transcription.py +++ b/examples/foundational/26a-gemini-multimodal-live-transcription.py @@ -37,8 +37,6 @@ async def main(): token, "Respond bot", DailyParams( - audio_in_sample_rate=16000, - audio_out_sample_rate=24000, audio_out_enabled=True, vad_enabled=True, vad_audio_passthrough=True, diff --git a/examples/foundational/26b-gemini-multimodal-live-function-calling.py b/examples/foundational/26b-gemini-multimodal-live-function-calling.py index 3ba19a2d5..03e3096a3 100644 --- a/examples/foundational/26b-gemini-multimodal-live-function-calling.py +++ b/examples/foundational/26b-gemini-multimodal-live-function-calling.py @@ -84,8 +84,6 @@ async def main(): token, "Respond bot", DailyParams( - audio_in_sample_rate=16000, - audio_out_sample_rate=24000, audio_out_enabled=True, vad_enabled=True, vad_audio_passthrough=True, diff --git a/examples/foundational/26c-gemini-multimodal-live-video.py b/examples/foundational/26c-gemini-multimodal-live-video.py index f9bc2a8ce..0beac44c9 100644 --- a/examples/foundational/26c-gemini-multimodal-live-video.py +++ b/examples/foundational/26c-gemini-multimodal-live-video.py @@ -37,8 +37,6 @@ async def main(): token, "Respond bot", DailyParams( - audio_in_sample_rate=16000, - audio_out_sample_rate=24000, audio_out_enabled=True, vad_enabled=True, vad_audio_passthrough=True, @@ -47,8 +45,6 @@ async def main(): # matter because we can only use the Multimodal Live API's phrase # endpointing, for now. vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.5)), - start_audio_paused=True, - start_video_paused=True, ), ) diff --git a/examples/foundational/26d-gemini-multimodal-live-text.py b/examples/foundational/26d-gemini-multimodal-live-text.py index aa72cae53..b243399e9 100644 --- a/examples/foundational/26d-gemini-multimodal-live-text.py +++ b/examples/foundational/26d-gemini-multimodal-live-text.py @@ -52,8 +52,6 @@ async def main(): token, "Respond bot", DailyParams( - audio_in_sample_rate=16000, - audio_out_sample_rate=24000, audio_out_enabled=True, vad_enabled=True, vad_audio_passthrough=True, diff --git a/examples/foundational/29-livekit-audio-chat.py b/examples/foundational/29-livekit-audio-chat.py index a3f58ece3..2ad02c296 100644 --- a/examples/foundational/29-livekit-audio-chat.py +++ b/examples/foundational/29-livekit-audio-chat.py @@ -38,8 +38,6 @@ load_dotenv(override=True) logger.remove(0) logger.add(sys.stderr, level="DEBUG") -DESIRED_SAMPLE_RATE = 16000 - def generate_token(room_name: str, participant_name: str, api_key: str, api_secret: str) -> str: token = api.AccessToken(api_key, api_secret) @@ -114,11 +112,8 @@ async def main(): token=token, room_name=room_name, params=LiveKitParams( - audio_in_channels=1, audio_in_enabled=True, audio_out_enabled=True, - audio_in_sample_rate=DESIRED_SAMPLE_RATE, - audio_out_sample_rate=DESIRED_SAMPLE_RATE, vad_analyzer=SileroVADAnalyzer(), vad_enabled=True, vad_audio_passthrough=True, @@ -128,7 +123,6 @@ async def main(): stt = DeepgramSTTService( api_key=os.getenv("DEEPGRAM_API_KEY"), live_options=LiveOptions( - sample_rate=DESIRED_SAMPLE_RATE, vad_events=True, ), ) @@ -138,7 +132,6 @@ async def main(): tts = CartesiaTTSService( api_key=os.getenv("CARTESIA_API_KEY"), voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady - sample_rate=DESIRED_SAMPLE_RATE, ) messages = [ diff --git a/examples/simple-chatbot/server/bot-gemini.py b/examples/simple-chatbot/server/bot-gemini.py index 7999b1bfc..a7e7717ce 100644 --- a/examples/simple-chatbot/server/bot-gemini.py +++ b/examples/simple-chatbot/server/bot-gemini.py @@ -121,8 +121,6 @@ async def main(): token, "Chatbot", DailyParams( - audio_in_sample_rate=16000, - audio_out_sample_rate=24000, audio_out_enabled=True, camera_out_enabled=True, camera_out_width=1024, diff --git a/examples/studypal/studypal.py b/examples/studypal/studypal.py index b4691fc5f..0b1496e3d 100644 --- a/examples/studypal/studypal.py +++ b/examples/studypal/studypal.py @@ -112,7 +112,6 @@ async def main(): token, "studypal", DailyParams( - audio_out_sample_rate=44100, audio_out_enabled=True, transcription_enabled=True, vad_enabled=True, @@ -124,7 +123,6 @@ async def main(): api_key=os.getenv("CARTESIA_API_KEY"), voice_id=os.getenv("CARTESIA_VOICE_ID", "4d2fd738-3b3d-4368-957a-bb4805275bd9"), # British Narration Lady: 4d2fd738-3b3d-4368-957a-bb4805275bd9 - sample_rate=44100, ) llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o-mini") @@ -155,7 +153,12 @@ Your task is to help the user understand and learn from this article in 2 senten ] ) - task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True, enable_metrics=True)) + task = PipelineTask( + pipeline, + PipelineParams( + audio_out_sample_rate=44100, allow_interruptions=True, enable_metrics=True + ), + ) @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): diff --git a/examples/twilio-chatbot/bot.py b/examples/twilio-chatbot/bot.py index 64c3c9cb5..6d5953954 100644 --- a/examples/twilio-chatbot/bot.py +++ b/examples/twilio-chatbot/bot.py @@ -11,7 +11,6 @@ import sys import wave import aiofiles -from deepgram import LiveOptions from dotenv import load_dotenv from fastapi import WebSocket from loguru import logger @@ -36,8 +35,6 @@ load_dotenv(override=True) logger.remove(0) logger.add(sys.stderr, level="DEBUG") -SAMPLE_RATE = 8000 - async def save_audio(server_name: str, audio: bytes, sample_rate: int, num_channels: int): if len(audio) > 0: @@ -63,29 +60,21 @@ async def run_bot(websocket_client: WebSocket, stream_sid: str, testing: bool): params=FastAPIWebsocketParams( audio_in_enabled=True, audio_out_enabled=True, - audio_out_sample_rate=SAMPLE_RATE, add_wav_header=False, vad_enabled=True, - vad_analyzer=SileroVADAnalyzer(sample_rate=SAMPLE_RATE), + vad_analyzer=SileroVADAnalyzer(), vad_audio_passthrough=True, - serializer=TwilioFrameSerializer( - stream_sid, TwilioFrameSerializer.InputParams(sample_rate=SAMPLE_RATE) - ), + serializer=TwilioFrameSerializer(stream_sid), ), ) llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") - stt = DeepgramSTTService( - api_key=os.getenv("DEEPGRAM_API_KEY"), - live_options=LiveOptions(sample_rate=SAMPLE_RATE), - audio_passthrough=True, - ) + stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"), audio_passthrough=True) tts = CartesiaTTSService( api_key=os.getenv("CARTESIA_API_KEY"), voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady - sample_rate=SAMPLE_RATE, push_silence_after_stop=testing, ) @@ -101,7 +90,7 @@ async def run_bot(websocket_client: WebSocket, stream_sid: str, testing: bool): # NOTE: Watch out! This will save all the conversation in memory. You can # pass `buffer_size` to get periodic callbacks. - audiobuffer = AudioBufferProcessor(sample_rate=SAMPLE_RATE) + audiobuffer = AudioBufferProcessor() pipeline = Pipeline( [ @@ -116,7 +105,12 @@ async def run_bot(websocket_client: WebSocket, stream_sid: str, testing: bool): ] ) - task = PipelineTask(pipeline, params=PipelineParams(allow_interruptions=True)) + task = PipelineTask( + pipeline, + params=PipelineParams( + audio_in_sample_rate=8000, audio_out_sample_rate=8000, allow_interruptions=True + ), + ) @transport.event_handler("on_client_connected") async def on_client_connected(transport, client): diff --git a/examples/twilio-chatbot/client.py b/examples/twilio-chatbot/client.py index fa710773b..7cd90e2c8 100644 --- a/examples/twilio-chatbot/client.py +++ b/examples/twilio-chatbot/client.py @@ -16,7 +16,6 @@ from uuid import uuid4 import aiofiles import aiohttp -from deepgram import LiveOptions from dotenv import load_dotenv from loguru import logger @@ -44,7 +43,6 @@ logger.add(sys.stderr, level="DEBUG") DEFAULT_CLIENT_DURATION = 30 -SAMPLE_RATE = 8000 async def download_twiml(server_url: str) -> str: @@ -92,15 +90,10 @@ async def run_client(client_name: str, server_url: str, duration_secs: int): params=WebsocketClientParams( audio_in_enabled=True, audio_out_enabled=True, - audio_out_sample_rate=SAMPLE_RATE, add_wav_header=False, - serializer=TwilioFrameSerializer( - stream_sid, params=TwilioFrameSerializer.InputParams(sample_rate=SAMPLE_RATE) - ), + serializer=TwilioFrameSerializer(stream_sid), vad_enabled=True, - vad_analyzer=SileroVADAnalyzer( - params=VADParams(stop_secs=1.5), sample_rate=SAMPLE_RATE - ), + vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=1.5)), vad_audio_passthrough=True, ), ) @@ -110,14 +103,12 @@ async def run_client(client_name: str, server_url: str, duration_secs: int): # We let the audio passthrough so we can record the conversation. stt = DeepgramSTTService( api_key=os.getenv("DEEPGRAM_API_KEY"), - live_options=LiveOptions(sample_rate=SAMPLE_RATE), audio_passthrough=True, ) tts = CartesiaTTSService( api_key=os.getenv("CARTESIA_API_KEY"), voice_id="e13cae5c-ec59-4f71-b0a6-266df3c9bb8e", # Madame Mischief - sample_rate=SAMPLE_RATE, push_silence_after_stop=True, ) @@ -133,7 +124,7 @@ async def run_client(client_name: str, server_url: str, duration_secs: int): # NOTE: Watch out! This will save all the conversation in memory. You can # pass `buffer_size` to get periodic callbacks. - audiobuffer = AudioBufferProcessor(sample_rate=SAMPLE_RATE) + audiobuffer = AudioBufferProcessor() pipeline = Pipeline( [ @@ -148,7 +139,12 @@ async def run_client(client_name: str, server_url: str, duration_secs: int): ] ) - task = PipelineTask(pipeline, params=PipelineParams(allow_interruptions=True)) + task = PipelineTask( + pipeline, + params=PipelineParams( + audio_in_sample_rate=8000, audio_out_sample_rate=8000, allow_interruptions=True + ), + ) @transport.event_handler("on_connected") async def on_connected(transport: WebsocketClientTransport, client): diff --git a/examples/websocket-server/bot.py b/examples/websocket-server/bot.py index a36339fff..9a0de15c6 100644 --- a/examples/websocket-server/bot.py +++ b/examples/websocket-server/bot.py @@ -17,6 +17,7 @@ from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.serializers.protobuf import ProtobufFrameSerializer from pipecat.services.cartesia import CartesiaTTSService from pipecat.services.deepgram import DeepgramSTTService from pipecat.services.openai import OpenAILLMService @@ -80,7 +81,7 @@ class SessionTimeoutHandler: async def main(): transport = WebsocketServerTransport( params=WebsocketServerParams( - audio_out_sample_rate=16000, + serializer=ProtobufFrameSerializer(), audio_out_enabled=True, add_wav_header=True, vad_enabled=True, @@ -97,7 +98,6 @@ async def main(): tts = CartesiaTTSService( api_key=os.getenv("CARTESIA_API_KEY"), voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady - sample_rate=16000, ) messages = [ @@ -122,7 +122,12 @@ async def main(): ] ) - task = PipelineTask(pipeline, params=PipelineParams(allow_interruptions=True)) + task = PipelineTask( + pipeline, + params=PipelineParams( + audio_in_sample_rate=16000, audio_out_sample_rate=16000, allow_interruptions=True + ), + ) @transport.event_handler("on_client_connected") async def on_client_connected(transport, client): diff --git a/src/pipecat/audio/vad/silero.py b/src/pipecat/audio/vad/silero.py index 00a1358a5..a5db2a0ae 100644 --- a/src/pipecat/audio/vad/silero.py +++ b/src/pipecat/audio/vad/silero.py @@ -5,6 +5,7 @@ # import time +from typing import Optional import numpy as np from loguru import logger @@ -104,11 +105,8 @@ class SileroOnnxModel: class SileroVADAnalyzer(VADAnalyzer): - def __init__(self, *, sample_rate: int = 16000, params: VADParams = VADParams()): - super().__init__(sample_rate=sample_rate, num_channels=1, params=params) - - if sample_rate != 16000 and sample_rate != 8000: - raise ValueError("Silero VAD sample rate needs to be 16000 or 8000") + def __init__(self, *, sample_rate: Optional[int] = None, params: VADParams = VADParams()): + super().__init__(sample_rate=sample_rate, params=params) logger.debug("Loading Silero VAD model...") @@ -138,6 +136,12 @@ class SileroVADAnalyzer(VADAnalyzer): # VADAnalyzer # + def set_sample_rate(self, sample_rate: int): + if sample_rate != 16000 and sample_rate != 8000: + raise ValueError("Silero VAD sample rate needs to be 16000 or 8000") + + super().set_sample_rate(sample_rate) + def num_frames_required(self) -> int: return 512 if self.sample_rate == 16000 else 256 diff --git a/src/pipecat/audio/vad/vad_analyzer.py b/src/pipecat/audio/vad/vad_analyzer.py index a6c4f70a1..b5d9b0ba2 100644 --- a/src/pipecat/audio/vad/vad_analyzer.py +++ b/src/pipecat/audio/vad/vad_analyzer.py @@ -6,6 +6,7 @@ from abc import abstractmethod from enum import Enum +from typing import Optional from loguru import logger from pydantic import BaseModel @@ -33,11 +34,11 @@ class VADParams(BaseModel): class VADAnalyzer: - def __init__(self, *, sample_rate: int, num_channels: int, params: VADParams): - self._sample_rate = sample_rate - self._num_channels = num_channels - - self.set_params(params) + def __init__(self, *, sample_rate: Optional[int] = None, params: VADParams): + self._init_sample_rate = sample_rate + self._sample_rate = 0 + self._params = params + self._num_channels = 1 self._vad_buffer = b"" @@ -65,13 +66,17 @@ class VADAnalyzer: def voice_confidence(self, buffer) -> float: pass + def set_sample_rate(self, sample_rate: int): + self._sample_rate = self._init_sample_rate or sample_rate + self.set_params(self._params) + def set_params(self, params: VADParams): logger.info(f"Setting VAD params to: {params}") self._params = params self._vad_frames = self.num_frames_required() self._vad_frames_num_bytes = self._vad_frames * self._num_channels * 2 - vad_frames_per_sec = self._vad_frames / self._sample_rate + vad_frames_per_sec = self._vad_frames / self.sample_rate self._vad_start_frames = round(self._params.start_secs / vad_frames_per_sec) self._vad_stop_frames = round(self._params.stop_secs / vad_frames_per_sec) @@ -80,7 +85,7 @@ class VADAnalyzer: self._vad_state: VADState = VADState.QUIET def _get_smoothed_volume(self, audio: bytes) -> float: - volume = calculate_audio_volume(audio, self._sample_rate) + volume = calculate_audio_volume(audio, self.sample_rate) return exp_smoothing(volume, self._prev_volume, self._smoothing_factor) def analyze_audio(self, buffer) -> VADState: diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 0038fcea9..d34f56208 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -428,6 +428,8 @@ class StartFrame(SystemFrame): clock: BaseClock task_manager: TaskManager + audio_in_sample_rate: int = 16000 + audio_out_sample_rate: int = 24000 allow_interruptions: bool = False enable_metrics: bool = False enable_usage_metrics: bool = False diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 6f9c95bb0..da300100d 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -40,6 +40,8 @@ HEARTBEAT_MONITOR_SECONDS = HEARTBEAT_SECONDS * 5 class PipelineParams(BaseModel): model_config = ConfigDict(arbitrary_types_allowed=True) + audio_in_sample_rate: int = 16000 + audio_out_sample_rate: int = 24000 allow_interruptions: bool = False enable_heartbeats: bool = False enable_metrics: bool = False @@ -136,6 +138,11 @@ class PipelineTask(BaseTask): """Returns the name of this task.""" return self._name + @property + def params(self) -> PipelineParams: + """Returns the pipeline parameters of this task.""" + return self._params + def set_event_loop(self, loop: asyncio.AbstractEventLoop): self._task_manager.set_event_loop(loop) @@ -275,6 +282,8 @@ class PipelineTask(BaseTask): enable_usage_metrics=self._params.enable_usage_metrics, report_only_initial_ttfb=self._params.report_only_initial_ttfb, observer=self._observer, + audio_in_sample_rate=self._params.audio_in_sample_rate, + audio_out_sample_rate=self._params.audio_out_sample_rate, ) await self._source.queue_frame(start_frame, FrameDirection.DOWNSTREAM) diff --git a/src/pipecat/processors/audio/audio_buffer_processor.py b/src/pipecat/processors/audio/audio_buffer_processor.py index d56604330..a1f213180 100644 --- a/src/pipecat/processors/audio/audio_buffer_processor.py +++ b/src/pipecat/processors/audio/audio_buffer_processor.py @@ -5,6 +5,7 @@ # import time +from typing import Optional from pipecat.audio.utils import create_default_resampler, interleave_stereo_audio, mix_audio from pipecat.frames.frames import ( @@ -14,6 +15,7 @@ from pipecat.frames.frames import ( Frame, InputAudioRawFrame, OutputAudioRawFrame, + StartFrame, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor @@ -33,10 +35,16 @@ class AudioBufferProcessor(FrameProcessor): """ def __init__( - self, *, sample_rate: int = 24000, num_channels: int = 1, buffer_size: int = 0, **kwargs + self, + *, + sample_rate: Optional[int] = None, + num_channels: int = 1, + buffer_size: int = 0, + **kwargs, ): super().__init__(**kwargs) - self._sample_rate = sample_rate + self._init_sample_rate = sample_rate + self._sample_rate = 0 self._num_channels = num_channels self._buffer_size = buffer_size @@ -86,6 +94,10 @@ class AudioBufferProcessor(FrameProcessor): async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) + # Update output sample rate if necessary. + if isinstance(frame, StartFrame): + self._update_sample_rate(frame) + if self._recording and isinstance(frame, InputAudioRawFrame): # Add silence if we need to. silence = self._compute_silence(self._last_user_frame_at) @@ -113,6 +125,9 @@ class AudioBufferProcessor(FrameProcessor): await self.push_frame(frame, direction) + def _update_sample_rate(self, frame: StartFrame): + self._sample_rate = self._init_sample_rate or frame.audio_out_sample_rate + async def _call_on_audio_data_handler(self): if not self.has_audio() or not self._recording: return diff --git a/src/pipecat/processors/audio/vad/silero.py b/src/pipecat/processors/audio/vad/silero.py index 2d6619a0c..edfe484ba 100644 --- a/src/pipecat/processors/audio/vad/silero.py +++ b/src/pipecat/processors/audio/vad/silero.py @@ -4,6 +4,8 @@ # SPDX-License-Identifier: BSD 2-Clause License # +from typing import Optional + from loguru import logger from pipecat.audio.vad.silero import SileroVADAnalyzer @@ -11,6 +13,7 @@ from pipecat.audio.vad.vad_analyzer import VADParams, VADState from pipecat.frames.frames import ( AudioRawFrame, Frame, + StartFrame, StartInterruptionFrame, StopInterruptionFrame, UserStartedSpeakingFrame, @@ -23,7 +26,7 @@ class SileroVAD(FrameProcessor): def __init__( self, *, - sample_rate: int = 16000, + sample_rate: Optional[int] = None, vad_params: VADParams = VADParams(), audio_passthrough: bool = False, ): @@ -41,6 +44,9 @@ class SileroVAD(FrameProcessor): async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) + if isinstance(frame, StartFrame): + self._vad_analyzer.set_sample_rate(frame.audio_in_sample_rate) + if isinstance(frame, AudioRawFrame): await self._analyze_audio(frame) if self._audio_passthrough: diff --git a/src/pipecat/processors/gstreamer/pipeline_source.py b/src/pipecat/processors/gstreamer/pipeline_source.py index 125f46d76..12b9cef67 100644 --- a/src/pipecat/processors/gstreamer/pipeline_source.py +++ b/src/pipecat/processors/gstreamer/pipeline_source.py @@ -5,6 +5,7 @@ # import asyncio +from typing import Optional from loguru import logger from pydantic import BaseModel @@ -38,7 +39,7 @@ class GStreamerPipelineSource(FrameProcessor): class OutputParams(BaseModel): video_width: int = 1280 video_height: int = 720 - audio_sample_rate: int = 24000 + audio_sample_rate: Optional[int] = None audio_channels: int = 1 clock_sync: bool = True @@ -46,6 +47,7 @@ class GStreamerPipelineSource(FrameProcessor): super().__init__(**kwargs) self._out_params = out_params + self._sample_rate = 0 Gst.init() @@ -90,6 +92,7 @@ class GStreamerPipelineSource(FrameProcessor): await self.push_frame(frame, direction) async def _start(self, frame: StartFrame): + self._sample_rate = self._out_params.audio_sample_rate or frame.audio_out_sample_rate self._player.set_state(Gst.State.PLAYING) async def _stop(self, frame: EndFrame): @@ -122,7 +125,7 @@ class GStreamerPipelineSource(FrameProcessor): audioresample = Gst.ElementFactory.make("audioresample", None) audiocapsfilter = Gst.ElementFactory.make("capsfilter", None) audiocaps = Gst.Caps.from_string( - f"audio/x-raw,format=S16LE,rate={self._out_params.audio_sample_rate},channels={self._out_params.audio_channels},layout=interleaved" + f"audio/x-raw,format=S16LE,rate={self._sample_rate},channels={self._out_params.audio_channels},layout=interleaved" ) audiocapsfilter.set_property("caps", audiocaps) appsink_audio = Gst.ElementFactory.make("appsink", None) @@ -188,7 +191,7 @@ class GStreamerPipelineSource(FrameProcessor): (_, info) = buffer.map(Gst.MapFlags.READ) frame = OutputAudioRawFrame( audio=info.data, - sample_rate=self._out_params.audio_sample_rate, + sample_rate=self._sample_rate, num_channels=self._out_params.audio_channels, ) asyncio.run_coroutine_threadsafe(self.push_frame(frame), self.get_event_loop()) diff --git a/src/pipecat/serializers/base_serializer.py b/src/pipecat/serializers/base_serializer.py index 8a0efc49b..c70b1408e 100644 --- a/src/pipecat/serializers/base_serializer.py +++ b/src/pipecat/serializers/base_serializer.py @@ -7,7 +7,7 @@ from abc import ABC, abstractmethod from enum import Enum -from pipecat.frames.frames import Frame +from pipecat.frames.frames import Frame, StartFrame class FrameSerializerType(Enum): @@ -21,6 +21,9 @@ class FrameSerializer(ABC): def type(self) -> FrameSerializerType: pass + async def setup(self, frame: StartFrame): + pass + @abstractmethod async def serialize(self, frame: Frame) -> str | bytes | None: pass diff --git a/src/pipecat/serializers/telnyx.py b/src/pipecat/serializers/telnyx.py index 79c6f42ff..9bff7535f 100644 --- a/src/pipecat/serializers/telnyx.py +++ b/src/pipecat/serializers/telnyx.py @@ -6,6 +6,7 @@ import base64 import json +from typing import Optional from pydantic import BaseModel @@ -22,6 +23,7 @@ from pipecat.frames.frames import ( InputAudioRawFrame, InputDTMFFrame, KeypadEntry, + StartFrame, StartInterruptionFrame, ) from pipecat.serializers.base_serializer import FrameSerializer, FrameSerializerType @@ -29,8 +31,8 @@ from pipecat.serializers.base_serializer import FrameSerializer, FrameSerializer class TelnyxFrameSerializer(FrameSerializer): class InputParams(BaseModel): - telnyx_sample_rate: int = 8000 - sample_rate: int = 16000 + telnyx_sample_rate: Optional[int] = None + sample_rate: Optional[int] = None inbound_encoding: str = "PCMU" outbound_encoding: str = "PCMU" @@ -52,17 +54,21 @@ class TelnyxFrameSerializer(FrameSerializer): def type(self) -> FrameSerializerType: return FrameSerializerType.TEXT + async def setup(self, frame: StartFrame): + self._telnyx_sample_rate = self._params.telnyx_sample_rate or frame.audio_in_sample_rate + self._sample_rate = self._params.sample_rate or frame.audio_out_sample_rate + async def serialize(self, frame: Frame) -> str | bytes | None: if isinstance(frame, AudioRawFrame): data = frame.audio if self._params.inbound_encoding == "PCMU": serialized_data = await pcm_to_ulaw( - data, frame.sample_rate, self._params.telnyx_sample_rate, self._resampler + data, frame.sample_rate, self._telnyx_sample_rate, self._resampler ) elif self._params.inbound_encoding == "PCMA": serialized_data = await pcm_to_alaw( - data, frame.sample_rate, self._params.telnyx_sample_rate, self._resampler + data, frame.sample_rate, self._telnyx_sample_rate, self._resampler ) else: raise ValueError(f"Unsupported encoding: {self._params.inbound_encoding}") @@ -89,22 +95,22 @@ class TelnyxFrameSerializer(FrameSerializer): if self._params.outbound_encoding == "PCMU": deserialized_data = await ulaw_to_pcm( payload, - self._params.telnyx_sample_rate, - self._params.sample_rate, + self._telnyx_sample_rate, + self._sample_rate, self._resampler, ) elif self._params.outbound_encoding == "PCMA": deserialized_data = await alaw_to_pcm( payload, - self._params.telnyx_sample_rate, - self._params.sample_rate, + self._telnyx_sample_rate, + self._sample_rate, self._resampler, ) else: raise ValueError(f"Unsupported encoding: {self._params.outbound_encoding}") audio_frame = InputAudioRawFrame( - audio=deserialized_data, num_channels=1, sample_rate=self._params.sample_rate + audio=deserialized_data, num_channels=1, sample_rate=self._sample_rate ) return audio_frame elif message["event"] == "dtmf": diff --git a/src/pipecat/serializers/twilio.py b/src/pipecat/serializers/twilio.py index 7862fa0db..0ca37c221 100644 --- a/src/pipecat/serializers/twilio.py +++ b/src/pipecat/serializers/twilio.py @@ -6,6 +6,7 @@ import base64 import json +from typing import Optional from pydantic import BaseModel @@ -16,6 +17,7 @@ from pipecat.frames.frames import ( InputAudioRawFrame, InputDTMFFrame, KeypadEntry, + StartFrame, StartInterruptionFrame, TransportMessageFrame, TransportMessageUrgentFrame, @@ -25,19 +27,26 @@ from pipecat.serializers.base_serializer import FrameSerializer, FrameSerializer class TwilioFrameSerializer(FrameSerializer): class InputParams(BaseModel): - twilio_sample_rate: int = 8000 - sample_rate: int = 16000 + twilio_sample_rate: Optional[int] = None + sample_rate: Optional[int] = None def __init__(self, stream_sid: str, params: InputParams = InputParams()): self._stream_sid = stream_sid self._params = params + self._twilio_sample_rate = 0 + self._sample_rate = 0 + self._resampler = create_default_resampler() @property def type(self) -> FrameSerializerType: return FrameSerializerType.TEXT + async def setup(self, frame: StartFrame): + self._twilio_sample_rate = self._params.twilio_sample_rate or frame.audio_in_sample_rate + self._sample_rate = self._params.sample_rate or frame.audio_out_sample_rate + async def serialize(self, frame: Frame) -> str | bytes | None: if isinstance(frame, StartInterruptionFrame): answer = {"event": "clear", "streamSid": self._stream_sid} @@ -46,7 +55,7 @@ class TwilioFrameSerializer(FrameSerializer): data = frame.audio serialized_data = await pcm_to_ulaw( - data, frame.sample_rate, self._params.twilio_sample_rate, self._resampler + data, frame.sample_rate, self._twilio_sample_rate, self._resampler ) payload = base64.b64encode(serialized_data).decode("utf-8") answer = { @@ -67,10 +76,10 @@ class TwilioFrameSerializer(FrameSerializer): payload = base64.b64decode(payload_base64) deserialized_data = await ulaw_to_pcm( - payload, self._params.twilio_sample_rate, self._params.sample_rate, self._resampler + payload, self._twilio_sample_rate, self._sample_rate, self._resampler ) audio_frame = InputAudioRawFrame( - audio=deserialized_data, num_channels=1, sample_rate=self._params.sample_rate + audio=deserialized_data, num_channels=1, sample_rate=self._sample_rate ) return audio_frame elif message["event"] == "dtmf": diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index 5044c7d92..1bbdb4340 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -213,7 +213,7 @@ class TTSService(AIService): # if push_silence_after_stop is True, send this amount of audio silence silence_time_s: float = 2.0, # TTS output sample rate - sample_rate: int = 24000, + sample_rate: Optional[int] = None, text_filter: Optional[BaseTextFilter] = None, **kwargs, ): @@ -224,7 +224,8 @@ class TTSService(AIService): self._stop_frame_timeout_s: float = stop_frame_timeout_s self._push_silence_after_stop: bool = push_silence_after_stop self._silence_time_s: float = silence_time_s - self._sample_rate: int = sample_rate + self._init_sample_rate = sample_rate + self._sample_rate = 0 self._voice_id: str = "" self._settings: Dict[str, Any] = {} self._text_filter: Optional[BaseTextFilter] = text_filter @@ -248,16 +249,20 @@ class TTSService(AIService): async def flush_audio(self): pass - def language_to_service_language(self, language: Language) -> str | None: - return Language(language) - # Converts the text to audio. @abstractmethod async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: pass + def language_to_service_language(self, language: Language) -> str | None: + return Language(language) + + async def update_setting(self, key: str, value: Any): + pass + async def start(self, frame: StartFrame): await super().start(frame) + self._sample_rate = self._init_sample_rate or frame.audio_out_sample_rate if self._push_stop_frames: self._stop_frame_task = self.create_task(self._stop_frame_handler()) @@ -467,9 +472,17 @@ class WordTTSService(TTSService): class STTService(AIService): """STTService is a base class for speech-to-text services.""" - def __init__(self, audio_passthrough=False, **kwargs): + def __init__( + self, + audio_passthrough=False, + # STT input sample rate + sample_rate: Optional[int] = None, + **kwargs, + ): super().__init__(**kwargs) self._audio_passthrough = audio_passthrough + self._init_sample_rate = sample_rate + self._sample_rate = 0 self._settings: Dict[str, Any] = {} self._muted: bool = False @@ -478,6 +491,10 @@ class STTService(AIService): """Returns whether the STT service is currently muted.""" return self._muted + @property + def sample_rate(self) -> int: + return self._sample_rate + @abstractmethod async def set_model(self, model: str): self.set_model_name(model) @@ -491,6 +508,10 @@ class STTService(AIService): """Returns transcript as a string""" pass + async def start(self, frame: StartFrame): + await super().start(frame) + self._sample_rate = self._init_sample_rate or frame.audio_in_sample_rate + async def _update_settings(self, settings: Mapping[str, Any]): logger.info(f"Updating STT settings: {self._settings}") for key, value in settings.items(): @@ -540,17 +561,15 @@ class SegmentedSTTService(STTService): min_volume: float = 0.6, max_silence_secs: float = 0.3, max_buffer_secs: float = 1.5, - sample_rate: int = 24000, - num_channels: int = 1, + sample_rate: Optional[int] = None, **kwargs, ): - super().__init__(**kwargs) + super().__init__(sample_rate=sample_rate, **kwargs) self._min_volume = min_volume self._max_silence_secs = max_silence_secs self._max_buffer_secs = max_buffer_secs - self._sample_rate = sample_rate - self._num_channels = num_channels - (self._content, self._wave) = self._new_wave() + self._content = None + self._wave = None self._silence_num_frames = 0 # Volume exponential smoothing self._smoothing_factor = 0.2 @@ -569,8 +588,8 @@ class SegmentedSTTService(STTService): # If buffer is not empty and we have enough data or there's been a long # silence, transcribe the audio gathered so far. - silence_secs = self._silence_num_frames / self._sample_rate - buffer_secs = self._wave.getnframes() / self._sample_rate + silence_secs = self._silence_num_frames / self.sample_rate + buffer_secs = self._wave.getnframes() / self.sample_rate if self._content.tell() > 0 and ( buffer_secs > self._max_buffer_secs or silence_secs > self._max_silence_secs ): @@ -580,18 +599,24 @@ class SegmentedSTTService(STTService): await self.process_generator(self.run_stt(self._content.read())) (self._content, self._wave) = self._new_wave() + async def start(self, frame: StartFrame): + await super().start(frame) + (self._content, self._wave) = self._new_wave() + async def stop(self, frame: EndFrame): + await super().stop(frame) self._wave.close() async def cancel(self, frame: CancelFrame): + await super().cancel(frame) self._wave.close() def _new_wave(self): content = io.BytesIO() ww = wave.open(content, "wb") ww.setsampwidth(2) - ww.setnchannels(self._num_channels) - ww.setframerate(self._sample_rate) + ww.setnchannels(1) + ww.setframerate(self.sample_rate) return (content, ww) def _get_smoothed_volume(self, frame: AudioRawFrame) -> float: diff --git a/src/pipecat/services/assemblyai.py b/src/pipecat/services/assemblyai.py index 78783f8b4..7abfcc5c6 100644 --- a/src/pipecat/services/assemblyai.py +++ b/src/pipecat/services/assemblyai.py @@ -5,7 +5,7 @@ # import asyncio -from typing import AsyncGenerator +from typing import AsyncGenerator, Optional from loguru import logger @@ -38,20 +38,17 @@ class AssemblyAISTTService(STTService): self, *, api_key: str, - sample_rate: int = 16000, + sample_rate: Optional[int] = None, encoding: AudioEncoding = AudioEncoding("pcm_s16le"), language=Language.EN, # Only English is supported for Realtime **kwargs, ): - super().__init__(**kwargs) + super().__init__(sample_rate=sample_rate, **kwargs) aai.settings.api_key = api_key self._transcriber: aai.RealtimeTranscriber | None = None - # Store reference to the main event loop for use in callback functions - self._loop = asyncio.get_event_loop() self._settings = { - "sample_rate": sample_rate, "encoding": encoding, "language": language, } @@ -121,7 +118,7 @@ class AssemblyAISTTService(STTService): # Schedule the coroutine to run in the main event loop # This is necessary because this callback runs in a different thread - asyncio.run_coroutine_threadsafe(self.push_frame(frame), self._loop) + asyncio.run_coroutine_threadsafe(self.push_frame(frame), self.get_event_loop()) def on_error(error: aai.RealtimeError): """Callback for handling errors from AssemblyAI. @@ -131,14 +128,16 @@ class AssemblyAISTTService(STTService): """ logger.error(f"{self}: An error occurred: {error}") # Schedule the coroutine to run in the main event loop - asyncio.run_coroutine_threadsafe(self.push_frame(ErrorFrame(str(error))), self._loop) + asyncio.run_coroutine_threadsafe( + self.push_frame(ErrorFrame(str(error))), self.get_event_loop() + ) def on_close(): """Callback for when the connection to AssemblyAI is closed.""" logger.info(f"{self}: Disconnected from AssemblyAI") self._transcriber = aai.RealtimeTranscriber( - sample_rate=self._settings["sample_rate"], + sample_rate=self.sample_rate, encoding=self._settings["encoding"], on_data=on_data, on_error=on_error, diff --git a/src/pipecat/services/aws.py b/src/pipecat/services/aws.py index 4444f07b7..fdbd73681 100644 --- a/src/pipecat/services/aws.py +++ b/src/pipecat/services/aws.py @@ -124,7 +124,7 @@ class PollyTTSService(TTSService): aws_session_token: Optional[str] = None, region: Optional[str] = None, voice_id: str = "Joanna", - sample_rate: int = 24000, + sample_rate: Optional[int] = None, params: InputParams = InputParams(), **kwargs, ): @@ -138,7 +138,6 @@ class PollyTTSService(TTSService): region_name=region, ) self._settings = { - "sample_rate": sample_rate, "engine": params.engine, "language": self.language_to_service_language(params.language) if params.language @@ -226,9 +225,7 @@ class PollyTTSService(TTSService): yield None return - audio_data = await self._resampler.resample( - audio_data, 16000, self._settings["sample_rate"] - ) + audio_data = await self._resampler.resample(audio_data, 16000, self.sample_rate) await self.start_tts_usage_metrics(text) @@ -239,7 +236,7 @@ class PollyTTSService(TTSService): chunk = audio_data[i : i + chunk_size] if len(chunk) > 0: await self.stop_ttfb_metrics() - frame = TTSAudioRawFrame(chunk, self._settings["sample_rate"], 1) + frame = TTSAudioRawFrame(chunk, self.sample_rate, 1) yield frame yield TTSStoppedFrame() diff --git a/src/pipecat/services/azure.py b/src/pipecat/services/azure.py index c7a0ce547..530c195fd 100644 --- a/src/pipecat/services/azure.py +++ b/src/pipecat/services/azure.py @@ -450,14 +450,13 @@ class AzureBaseTTSService(TTSService): api_key: str, region: str, voice="en-US-SaraNeural", - sample_rate: int = 24000, + sample_rate: Optional[int] = None, params: InputParams = InputParams(), **kwargs, ): super().__init__(sample_rate=sample_rate, **kwargs) self._settings = { - "sample_rate": sample_rate, "emphasis": params.emphasis, "language": self.language_to_service_language(params.language) if params.language @@ -537,7 +536,7 @@ class AzureTTSService(AzureBaseTTSService): speech_recognition_language=self._settings["language"], ) speech_config.set_speech_synthesis_output_format( - sample_rate_to_output_format(self._settings["sample_rate"]) + sample_rate_to_output_format(self.sample_rate) ) speech_config.set_service_property( "synthesizer.synthesis.connection.synthesisConnectionImpl", @@ -591,7 +590,7 @@ class AzureTTSService(AzureBaseTTSService): yield TTSAudioRawFrame( audio=chunk, - sample_rate=self._settings["sample_rate"], + sample_rate=self.sample_rate, num_channels=1, ) @@ -612,7 +611,7 @@ class AzureHttpTTSService(AzureBaseTTSService): speech_recognition_language=self._settings["language"], ) speech_config.set_speech_synthesis_output_format( - sample_rate_to_output_format(self._settings["sample_rate"]) + sample_rate_to_output_format(self.sample_rate) ) self._speech_synthesizer = SpeechSynthesizer(speech_config=speech_config, audio_config=None) @@ -633,7 +632,7 @@ class AzureHttpTTSService(AzureBaseTTSService): # Azure always sends a 44-byte header. Strip it off. yield TTSAudioRawFrame( audio=result.audio_data[44:], - sample_rate=self._settings["sample_rate"], + sample_rate=self.sample_rate, num_channels=1, ) yield TTSStoppedFrame() @@ -650,24 +649,14 @@ class AzureSTTService(STTService): *, api_key: str, region: str, - language=Language.EN_US, - sample_rate=24000, - channels=1, + language: Language = Language.EN_US, + sample_rate: Optional[int] = None, **kwargs, ): - super().__init__(**kwargs) + super().__init__(sample_rate=sample_rate, **kwargs) - speech_config = SpeechConfig(subscription=api_key, region=region) - speech_config.speech_recognition_language = language - - stream_format = AudioStreamFormat(samples_per_second=sample_rate, channels=channels) - self._audio_stream = PushAudioInputStream(stream_format) - - audio_config = AudioConfig(stream=self._audio_stream) - self._speech_recognizer = SpeechRecognizer( - speech_config=speech_config, audio_config=audio_config - ) - self._speech_recognizer.recognized.connect(self._on_handle_recognized) + self._speech_config = SpeechConfig(subscription=api_key, region=region) + self._speech_config.speech_recognition_language = language async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]: await self.start_processing_metrics() @@ -677,6 +666,16 @@ class AzureSTTService(STTService): async def start(self, frame: StartFrame): await super().start(frame) + + stream_format = AudioStreamFormat(samples_per_second=self.sample_rate, channels=1) + self._audio_stream = PushAudioInputStream(stream_format) + + audio_config = AudioConfig(stream=self._audio_stream) + + self._speech_recognizer = SpeechRecognizer( + speech_config=self._speech_config, audio_config=audio_config + ) + self._speech_recognizer.recognized.connect(self._on_handle_recognized) self._speech_recognizer.start_continuous_recognition_async() async def stop(self, frame: EndFrame): diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index 75da116d5..525071271 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -89,7 +89,7 @@ class CartesiaTTSService(WordTTSService, WebsocketService): cartesia_version: str = "2024-06-10", url: str = "wss://api.cartesia.ai/tts/websocket", model: str = "sonic", - sample_rate: int = 24000, + sample_rate: Optional[int] = None, encoding: str = "pcm_s16le", container: str = "raw", params: InputParams = InputParams(), @@ -121,7 +121,7 @@ class CartesiaTTSService(WordTTSService, WebsocketService): "output_format": { "container": container, "encoding": encoding, - "sample_rate": sample_rate, + "sample_rate": 0, }, "language": self.language_to_service_language(params.language) if params.language @@ -174,6 +174,7 @@ class CartesiaTTSService(WordTTSService, WebsocketService): async def start(self, frame: StartFrame): await super().start(frame) + self._settings["output_format"]["sample_rate"] = self.sample_rate await self._connect() async def stop(self, frame: EndFrame): @@ -262,7 +263,7 @@ class CartesiaTTSService(WordTTSService, WebsocketService): self.start_word_timestamps() frame = TTSAudioRawFrame( audio=base64.b64decode(msg["data"]), - sample_rate=self._settings["output_format"]["sample_rate"], + sample_rate=self.sample_rate, num_channels=1, ) await self.push_frame(frame) @@ -328,7 +329,7 @@ class CartesiaHttpTTSService(TTSService): voice_id: str, model: str = "sonic", base_url: str = "https://api.cartesia.ai", - sample_rate: int = 24000, + sample_rate: Optional[int] = None, encoding: str = "pcm_s16le", container: str = "raw", params: InputParams = InputParams(), @@ -341,7 +342,7 @@ class CartesiaHttpTTSService(TTSService): "output_format": { "container": container, "encoding": encoding, - "sample_rate": sample_rate, + "sample_rate": 0, }, "language": self.language_to_service_language(params.language) if params.language @@ -360,6 +361,10 @@ class CartesiaHttpTTSService(TTSService): def language_to_service_language(self, language: Language) -> str | None: return language_to_cartesia_language(language) + async def start(self, frame: StartFrame): + await super().start(frame) + self._settings["output_format"]["sample_rate"] = self.sample_rate + async def stop(self, frame: EndFrame): await super().stop(frame) await self._client.close() @@ -394,9 +399,7 @@ class CartesiaHttpTTSService(TTSService): ) frame = TTSAudioRawFrame( - audio=output["audio"], - sample_rate=self._settings["output_format"]["sample_rate"], - num_channels=1, + audio=output["audio"], sample_rate=self.sample_rate, num_channels=1 ) yield frame except Exception as e: diff --git a/src/pipecat/services/deepgram.py b/src/pipecat/services/deepgram.py index a5d36370f..b9d39fa63 100644 --- a/src/pipecat/services/deepgram.py +++ b/src/pipecat/services/deepgram.py @@ -5,7 +5,7 @@ # import asyncio -from typing import AsyncGenerator +from typing import AsyncGenerator, Optional from loguru import logger @@ -53,14 +53,13 @@ class DeepgramTTSService(TTSService): *, api_key: str, voice: str = "aura-helios-en", - sample_rate: int = 24000, + sample_rate: Optional[int] = None, encoding: str = "linear16", **kwargs, ): super().__init__(sample_rate=sample_rate, **kwargs) self._settings = { - "sample_rate": sample_rate, "encoding": encoding, } self.set_voice(voice) @@ -75,7 +74,7 @@ class DeepgramTTSService(TTSService): options = SpeakOptions( model=self._voice_id, encoding=self._settings["encoding"], - sample_rate=self._settings["sample_rate"], + sample_rate=self.sample_rate, container="none", ) @@ -103,9 +102,7 @@ class DeepgramTTSService(TTSService): chunk = audio_buffer.read(chunk_size) if not chunk: break - frame = TTSAudioRawFrame( - audio=chunk, sample_rate=self._settings["sample_rate"], num_channels=1 - ) + frame = TTSAudioRawFrame(audio=chunk, sample_rate=self.sample_rate, num_channels=1) yield frame yield TTSStoppedFrame() @@ -121,15 +118,16 @@ class DeepgramSTTService(STTService): *, api_key: str, url: str = "", - live_options: LiveOptions = None, + sample_rate: Optional[int] = None, + live_options: Optional[LiveOptions] = None, **kwargs, ): - super().__init__(**kwargs) + super().__init__(sample_rate=sample_rate, **kwargs) + default_options = LiveOptions( encoding="linear16", language=Language.EN, model="nova-2-general", - sample_rate=16000, channels=1, interim_results=True, smart_format=True, @@ -187,6 +185,7 @@ class DeepgramSTTService(STTService): async def start(self, frame: StartFrame): await super().start(frame) + self._settings["sample_rate"] = self.sample_rate await self._connect() async def stop(self, frame: EndFrame): diff --git a/src/pipecat/services/elevenlabs.py b/src/pipecat/services/elevenlabs.py index d37fb499b..74461e51c 100644 --- a/src/pipecat/services/elevenlabs.py +++ b/src/pipecat/services/elevenlabs.py @@ -104,17 +104,17 @@ def language_to_elevenlabs_language(language: Language) -> str | None: return result -def sample_rate_from_output_format(output_format: str) -> int: - match output_format: - case "pcm_16000": - return 16000 - case "pcm_22050": - return 22050 - case "pcm_24000": - return 24000 - case "pcm_44100": - return 44100 - return 16000 +def output_format_from_sample_rate(sample_rate: int) -> str: + match sample_rate: + case 16000: + return "pcm_16000" + case 22050: + return "pcm_22050" + case 24000: + return "pcm_24000" + case 44100: + return "pcm_44100" + return "pcm_16000" def calculate_word_times( @@ -165,7 +165,7 @@ class ElevenLabsTTSService(WordTTSService, WebsocketService): voice_id: str, model: str = "eleven_flash_v2_5", url: str = "wss://api.elevenlabs.io", - output_format: ElevenLabsOutputFormat = "pcm_24000", + sample_rate: Optional[int] = None, params: InputParams = InputParams(), **kwargs, ): @@ -189,7 +189,7 @@ class ElevenLabsTTSService(WordTTSService, WebsocketService): push_text_frames=False, push_stop_frames=True, stop_frame_timeout_s=2.0, - sample_rate=sample_rate_from_output_format(output_format), + sample_rate=sample_rate, **kwargs, ) WebsocketService.__init__(self) @@ -197,11 +197,9 @@ class ElevenLabsTTSService(WordTTSService, WebsocketService): self._api_key = api_key self._url = url self._settings = { - "sample_rate": sample_rate_from_output_format(output_format), "language": self.language_to_service_language(params.language) if params.language else None, - "output_format": output_format, "optimize_streaming_latency": params.optimize_streaming_latency, "stability": params.stability, "similarity_boost": params.similarity_boost, @@ -211,6 +209,7 @@ class ElevenLabsTTSService(WordTTSService, WebsocketService): } self.set_model_name(model) self.set_voice(voice_id) + self._output_format = "" # initialized in start() self._voice_settings = self._set_voice_settings() # Indicates if we have sent TTSStartedFrame. It will reset to False when @@ -254,7 +253,7 @@ class ElevenLabsTTSService(WordTTSService, WebsocketService): await self._disconnect() await self._connect() - async def _update_settings(self, settings: Dict[str, Any]): + async def _update_settings(self, settings: Mapping[str, Any]): prev_voice = self._voice_id await super()._update_settings(settings) if not prev_voice == self._voice_id: @@ -264,6 +263,7 @@ class ElevenLabsTTSService(WordTTSService, WebsocketService): async def start(self, frame: StartFrame): await super().start(frame) + self._output_format = output_format_from_sample_rate(self.sample_rate) await self._connect() async def stop(self, frame: EndFrame): @@ -322,7 +322,7 @@ class ElevenLabsTTSService(WordTTSService, WebsocketService): voice_id = self._voice_id model = self.model_name - output_format = self._settings["output_format"] + output_format = self._output_format url = f"{self._url}/v1/text-to-speech/{voice_id}/stream-input?model_id={model}&output_format={output_format}&auto_mode={self._settings['auto_mode']}" if self._settings["optimize_streaming_latency"]: @@ -375,7 +375,7 @@ class ElevenLabsTTSService(WordTTSService, WebsocketService): self.start_word_timestamps() audio = base64.b64decode(msg["audio"]) - frame = TTSAudioRawFrame(audio, self._settings["sample_rate"], 1) + frame = TTSAudioRawFrame(audio, self.sample_rate, 1) await self.push_frame(frame) if msg.get("alignment"): word_times = calculate_word_times(msg["alignment"], self._cumulative_time) @@ -428,7 +428,7 @@ class ElevenLabsHttpTTSService(TTSService): aiohttp_session: aiohttp ClientSession model: Model ID (default: "eleven_flash_v2_5" for low latency) base_url: API base URL - output_format: Audio output format (PCM) + sample_rate: Output sample rate params: Additional parameters for voice configuration """ @@ -448,24 +448,21 @@ class ElevenLabsHttpTTSService(TTSService): aiohttp_session: aiohttp.ClientSession, model: str = "eleven_flash_v2_5", base_url: str = "https://api.elevenlabs.io", - output_format: ElevenLabsOutputFormat = "pcm_24000", + sample_rate: Optional[int] = None, params: InputParams = InputParams(), **kwargs, ): - super().__init__(sample_rate=sample_rate_from_output_format(output_format), **kwargs) + super().__init__(sample_rate=sample_rate, **kwargs) self._api_key = api_key self._base_url = base_url - self._output_format = output_format self._params = params self._session = aiohttp_session self._settings = { - "sample_rate": sample_rate_from_output_format(output_format), "language": self.language_to_service_language(params.language) if params.language else None, - "output_format": output_format, "optimize_streaming_latency": params.optimize_streaming_latency, "stability": params.stability, "similarity_boost": params.similarity_boost, @@ -474,6 +471,7 @@ class ElevenLabsHttpTTSService(TTSService): } self.set_model_name(model) self.set_voice(voice_id) + self._output_format = "" # initialized in start() self._voice_settings = self._set_voice_settings() def can_generate_metrics(self) -> bool: @@ -508,6 +506,10 @@ class ElevenLabsHttpTTSService(TTSService): return voice_settings or None + async def start(self, frame: StartFrame): + await super().start(frame) + self._output_format = output_format_from_sample_rate(self.sample_rate) + async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: """Generate speech from text using ElevenLabs streaming API. @@ -570,7 +572,7 @@ class ElevenLabsHttpTTSService(TTSService): async for chunk in response.content: if chunk: await self.stop_ttfb_metrics() - yield TTSAudioRawFrame(chunk, self._settings["sample_rate"], 1) + yield TTSAudioRawFrame(chunk, self.sample_rate, 1) yield TTSStoppedFrame() diff --git a/src/pipecat/services/fish.py b/src/pipecat/services/fish.py index 6fadab683..d61514eb2 100644 --- a/src/pipecat/services/fish.py +++ b/src/pipecat/services/fish.py @@ -56,7 +56,7 @@ class FishAudioTTSService(TTSService, WebsocketService): api_key: str, model: str, # This is the reference_id output_format: FishAudioOutputFormat = "pcm", - sample_rate: int = 24000, + sample_rate: Optional[int] = None, params: InputParams = InputParams(), **kwargs, ): @@ -70,7 +70,7 @@ class FishAudioTTSService(TTSService, WebsocketService): self._started = False self._settings = { - "sample_rate": sample_rate, + "sample_rate": 0, "latency": params.latency, "format": output_format, "prosody": { @@ -92,6 +92,7 @@ class FishAudioTTSService(TTSService, WebsocketService): async def start(self, frame: StartFrame): await super().start(frame) + self._settings["sample_rate"] = self.sample_rate await self._connect() async def stop(self, frame: EndFrame): @@ -157,9 +158,7 @@ class FishAudioTTSService(TTSService, WebsocketService): audio_data = msg.get("audio") # Only process larger chunks to remove msgpack overhead if audio_data and len(audio_data) > 1024: - frame = TTSAudioRawFrame( - audio_data, self._settings["sample_rate"], 1 - ) + frame = TTSAudioRawFrame(audio_data, self.sample_rate, 1) await self.push_frame(frame) await self.stop_ttfb_metrics() continue diff --git a/src/pipecat/services/gemini_multimodal_live/events.py b/src/pipecat/services/gemini_multimodal_live/events.py index b2107e7f8..60f688997 100644 --- a/src/pipecat/services/gemini_multimodal_live/events.py +++ b/src/pipecat/services/gemini_multimodal_live/events.py @@ -48,7 +48,7 @@ class AudioInputMessage(BaseModel): realtimeInput: RealtimeInput @classmethod - def from_raw_audio(cls, raw_audio: bytes, sample_rate=16000) -> "AudioInputMessage": + def from_raw_audio(cls, raw_audio: bytes, sample_rate: int) -> "AudioInputMessage": data = base64.b64encode(raw_audio).decode("utf-8") return cls( realtimeInput=RealtimeInput( diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index 9f355d1ba..4f26b6e9f 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/src/pipecat/services/gemini_multimodal_live/gemini.py @@ -203,6 +203,8 @@ class GeminiMultimodalLiveLLMService(LLMService): self._bot_audio_buffer = bytearray() self._bot_text_buffer = "" + self._sample_rate = 24000 + self._settings = { "frequency_penalty": params.frequency_penalty, "max_tokens": params.max_tokens, @@ -521,7 +523,7 @@ class GeminiMultimodalLiveLLMService(LLMService): if self._audio_input_paused: return # Send all audio to Gemini - evt = events.AudioInputMessage.from_raw_audio(frame.audio) + evt = events.AudioInputMessage.from_raw_audio(frame.audio, frame.sample_rate) await self.send_client_event(evt) # Manage a buffer of audio to use for transcription audio = frame.audio @@ -650,7 +652,7 @@ class GeminiMultimodalLiveLLMService(LLMService): inline_data = part.inlineData if not inline_data: return - if inline_data.mimeType != "audio/pcm;rate=24000": + if inline_data.mimeType != f"audio/pcm;rate={self._sample_rate}": logger.warning(f"Unrecognized server_content format {inline_data.mimeType}") return @@ -665,7 +667,7 @@ class GeminiMultimodalLiveLLMService(LLMService): self._bot_audio_buffer.extend(audio) frame = TTSAudioRawFrame( audio=audio, - sample_rate=24000, + sample_rate=self._sample_rate, num_channels=1, ) await self.push_frame(frame) diff --git a/src/pipecat/services/gladia.py b/src/pipecat/services/gladia.py index 3cdeebd3c..3a91c62fa 100644 --- a/src/pipecat/services/gladia.py +++ b/src/pipecat/services/gladia.py @@ -131,7 +131,6 @@ def language_to_gladia_language(language: Language) -> str | None: class GladiaSTTService(STTService): class InputParams(BaseModel): - sample_rate: Optional[int] = 16000 language: Optional[Language] = Language.EN endpointing: Optional[float] = 0.2 maximum_duration_without_endpointing: Optional[int] = 10 @@ -144,17 +143,18 @@ class GladiaSTTService(STTService): api_key: str, url: str = "https://api.gladia.io/v2/live", confidence: float = 0.5, + sample_rate: Optional[int] = None, params: InputParams = InputParams(), **kwargs, ): - super().__init__(**kwargs) + super().__init__(sample_rate=sample_rate, **kwargs) self._api_key = api_key self._url = url self._settings = { "encoding": "wav/pcm", "bit_depth": 16, - "sample_rate": params.sample_rate, + "sample_rate": 0, "channels": 1, "language_config": { "languages": [self.language_to_service_language(params.language)] @@ -178,6 +178,7 @@ class GladiaSTTService(STTService): async def start(self, frame: StartFrame): await super().start(frame) + self._settings["sample_rate"] = self.sample_rate response = await self._setup_gladia() self._websocket = await websockets.connect(response["url"]) self._receive_task = self.create_task(self._receive_task_handler()) diff --git a/src/pipecat/services/google/google.py b/src/pipecat/services/google/google.py index d705bec47..971bf61c0 100644 --- a/src/pipecat/services/google/google.py +++ b/src/pipecat/services/google/google.py @@ -883,14 +883,13 @@ class GoogleTTSService(TTSService): credentials: Optional[str] = None, credentials_path: Optional[str] = None, voice_id: str = "en-US-Neural2-A", - sample_rate: int = 24000, + sample_rate: Optional[int] = None, params: InputParams = InputParams(), **kwargs, ): super().__init__(sample_rate=sample_rate, **kwargs) self._settings = { - "sample_rate": sample_rate, "pitch": params.pitch, "rate": params.rate, "volume": params.volume, @@ -996,7 +995,7 @@ class GoogleTTSService(TTSService): ) audio_config = texttospeech_v1.AudioConfig( audio_encoding=texttospeech_v1.AudioEncoding.LINEAR16, - sample_rate_hertz=self._settings["sample_rate"], + sample_rate_hertz=self.sample_rate, ) request = texttospeech_v1.SynthesizeSpeechRequest( @@ -1019,7 +1018,7 @@ class GoogleTTSService(TTSService): if not chunk: break await self.stop_ttfb_metrics() - frame = TTSAudioRawFrame(chunk, self._settings["sample_rate"], 1) + frame = TTSAudioRawFrame(chunk, self.sample_rate, 1) yield frame await asyncio.sleep(0) # Allow other tasks to run diff --git a/src/pipecat/services/lmnt.py b/src/pipecat/services/lmnt.py index 327901423..e40dc2ce9 100644 --- a/src/pipecat/services/lmnt.py +++ b/src/pipecat/services/lmnt.py @@ -5,7 +5,7 @@ # import json -from typing import AsyncGenerator +from typing import AsyncGenerator, Optional from loguru import logger @@ -66,7 +66,7 @@ class LmntTTSService(TTSService, WebsocketService): *, api_key: str, voice_id: str, - sample_rate: int = 24000, + sample_rate: Optional[int] = None, language: Language = Language.EN, **kwargs, ): @@ -81,7 +81,6 @@ class LmntTTSService(TTSService, WebsocketService): self._api_key = api_key self._voice_id = voice_id self._settings = { - "sample_rate": sample_rate, "language": self.language_to_service_language(language), "format": "raw", # Use raw format for direct PCM data } @@ -132,7 +131,7 @@ class LmntTTSService(TTSService, WebsocketService): "X-API-Key": self._api_key, "voice": self._voice_id, "format": self._settings["format"], - "sample_rate": self._settings["sample_rate"], + "sample_rate": self.sample_rate, "language": self._settings["language"], } @@ -175,7 +174,7 @@ class LmntTTSService(TTSService, WebsocketService): await self.stop_ttfb_metrics() frame = TTSAudioRawFrame( audio=message, - sample_rate=self._settings["sample_rate"], + sample_rate=self.sample_rate, num_channels=1, ) await self.push_frame(frame) diff --git a/src/pipecat/services/openai.py b/src/pipecat/services/openai.py index ab81a1abb..7a090eb0b 100644 --- a/src/pipecat/services/openai.py +++ b/src/pipecat/services/openai.py @@ -415,17 +415,14 @@ class OpenAITTSService(TTSService): def __init__( self, *, - api_key: str | None = None, + api_key: Optional[str] = None, voice: str = "alloy", model: Literal["tts-1", "tts-1-hd"] = "tts-1", - sample_rate: int = 24000, + sample_rate: Optional[int] = None, **kwargs, ): super().__init__(sample_rate=sample_rate, **kwargs) - self._settings = { - "sample_rate": sample_rate, - } self.set_model_name(model) self.set_voice(voice) @@ -465,7 +462,7 @@ class OpenAITTSService(TTSService): async for chunk in r.iter_bytes(8192): if len(chunk) > 0: await self.stop_ttfb_metrics() - frame = TTSAudioRawFrame(chunk, self._settings["sample_rate"], 1) + frame = TTSAudioRawFrame(chunk, self.sample_rate, 1) yield frame yield TTSStoppedFrame() except BadRequestError as e: diff --git a/src/pipecat/services/playht.py b/src/pipecat/services/playht.py index 07e78ae52..87c8f3c09 100644 --- a/src/pipecat/services/playht.py +++ b/src/pipecat/services/playht.py @@ -113,7 +113,7 @@ class PlayHTTTSService(TTSService, WebsocketService): user_id: str, voice_url: str, voice_engine: str = "Play3.0-mini", - sample_rate: int = 24000, + sample_rate: Optional[int] = None, output_format: str = "wav", params: InputParams = InputParams(), **kwargs, @@ -132,7 +132,6 @@ class PlayHTTTSService(TTSService, WebsocketService): self._request_id = None self._settings = { - "sample_rate": sample_rate, "language": self.language_to_service_language(params.language) if params.language else "english", @@ -250,7 +249,7 @@ class PlayHTTTSService(TTSService, WebsocketService): if message.startswith(b"RIFF"): continue await self.stop_ttfb_metrics() - frame = TTSAudioRawFrame(message, self._settings["sample_rate"], 1) + frame = TTSAudioRawFrame(message, self.sample_rate, 1) await self.push_frame(frame) else: logger.debug(f"Received text message: {message}") @@ -301,7 +300,7 @@ class PlayHTTTSService(TTSService, WebsocketService): "voice": self._voice_id, "voice_engine": self._settings["voice_engine"], "output_format": self._settings["output_format"], - "sample_rate": self._settings["sample_rate"], + "sample_rate": self.sample_rate, "language": self._settings["language"], "speed": self._settings["speed"], "seed": self._settings["seed"], @@ -339,7 +338,7 @@ class PlayHTHttpTTSService(TTSService): user_id: str, voice_url: str, voice_engine: str = "Play3.0-mini-http", # Options: Play3.0-mini-http, Play3.0-mini-ws - sample_rate: int = 24000, + sample_rate: Optional[int] = None, params: InputParams = InputParams(), **kwargs, ): @@ -353,7 +352,6 @@ class PlayHTHttpTTSService(TTSService): api_key=self._api_key, ) self._settings = { - "sample_rate": sample_rate, "language": self.language_to_service_language(params.language) if params.language else "english", @@ -377,7 +375,7 @@ class PlayHTHttpTTSService(TTSService): self._options = TTSOptions( voice=self._voice_id, language=playht_language, - sample_rate=self._settings["sample_rate"], + sample_rate=self.sample_rate, format=self._settings["format"], speed=self._settings["speed"], seed=self._settings["seed"], @@ -422,7 +420,7 @@ class PlayHTHttpTTSService(TTSService): else: if len(chunk): await self.stop_ttfb_metrics() - frame = TTSAudioRawFrame(chunk, self._settings["sample_rate"], 1) + frame = TTSAudioRawFrame(chunk, self.sample_rate, 1) yield frame yield TTSStoppedFrame() except Exception as e: diff --git a/src/pipecat/services/rime.py b/src/pipecat/services/rime.py index 444461f6b..51bc7253d 100644 --- a/src/pipecat/services/rime.py +++ b/src/pipecat/services/rime.py @@ -34,7 +34,7 @@ class RimeHttpTTSService(TTSService): api_key: str, voice_id: str = "eva", model: str = "mist", - sample_rate: int = 24000, + sample_rate: Optional[int] = None, params: InputParams = InputParams(), **kwargs, ): @@ -43,7 +43,6 @@ class RimeHttpTTSService(TTSService): self._api_key = api_key self._base_url = "https://users.rime.ai/v1/rime-tts" self._settings = { - "samplingRate": sample_rate, "speedAlpha": params.speed_alpha, "reduceLatency": params.reduce_latency, "pauseBetweenBrackets": params.pause_between_brackets, @@ -71,6 +70,7 @@ class RimeHttpTTSService(TTSService): payload["text"] = text payload["speaker"] = self._voice_id payload["modelId"] = self._model_name + payload["samplingRate"] = self.sample_rate try: await self.start_ttfb_metrics() @@ -96,7 +96,7 @@ class RimeHttpTTSService(TTSService): first_chunk = False if chunk: - frame = TTSAudioRawFrame(chunk, self._settings["samplingRate"], 1) + frame = TTSAudioRawFrame(chunk, self.sample_rate, 1) yield frame yield TTSStoppedFrame() diff --git a/src/pipecat/services/riva.py b/src/pipecat/services/riva.py index cac1946d5..458cb4919 100644 --- a/src/pipecat/services/riva.py +++ b/src/pipecat/services/riva.py @@ -49,7 +49,7 @@ class FastPitchTTSService(TTSService): api_key: str, server: str = "grpc.nvcf.nvidia.com:443", voice_id: str = "English-US.Female-1", - sample_rate: int = 24000, + sample_rate: Optional[int] = None, function_id: str = "0149dedb-2be8-4195-b9a0-e57e0e14f972", params: InputParams = InputParams(), **kwargs, @@ -57,7 +57,6 @@ class FastPitchTTSService(TTSService): super().__init__(sample_rate=sample_rate, **kwargs) self._api_key = api_key self._voice_id = voice_id - self._sample_rate = sample_rate self._language_code = params.language self._quality = params.quality @@ -87,7 +86,7 @@ class FastPitchTTSService(TTSService): text, self._voice_id, self._language_code, - sample_rate_hz=self._sample_rate, + sample_rate_hz=self.sample_rate, audio_prompt_file=None, quality=self._quality, custom_dictionary={}, @@ -114,7 +113,7 @@ class FastPitchTTSService(TTSService): await self.stop_ttfb_metrics() frame = TTSAudioRawFrame( audio=resp.audio, - sample_rate=self._sample_rate, + sample_rate=self.sample_rate, num_channels=1, ) yield frame @@ -136,10 +135,11 @@ class ParakeetSTTService(STTService): api_key: str, server: str = "grpc.nvcf.nvidia.com:443", function_id: str = "1598d209-5e27-4d3c-8079-4751568b1081", + sample_rate: Optional[int] = None, params: InputParams = InputParams(), **kwargs, ): - super().__init__(**kwargs) + super().__init__(sample_rate=sample_rate, **kwargs) self._api_key = api_key self._profanity_filter = False self._automatic_punctuation = False @@ -154,7 +154,6 @@ class ParakeetSTTService(STTService): self._stop_history_eou = -1 self._stop_threshold_eou = -1.0 self._custom_configuration = "" - self._sample_rate: int = 16000 self.set_model_name("parakeet-ctc-1.1b-asr") @@ -166,6 +165,14 @@ class ParakeetSTTService(STTService): self._asr_service = riva.client.ASRService(auth) + self._queue = asyncio.Queue() + + def can_generate_metrics(self) -> bool: + return False + + async def start(self, frame: StartFrame): + await super().start(frame) + config = riva.client.StreamingRecognitionConfig( config=riva.client.RecognitionConfig( encoding=riva.client.AudioEncoding.LINEAR_PCM, @@ -175,14 +182,16 @@ class ParakeetSTTService(STTService): profanity_filter=self._profanity_filter, enable_automatic_punctuation=self._automatic_punctuation, verbatim_transcripts=not self._no_verbatim_transcripts, - sample_rate_hertz=self._sample_rate, + sample_rate_hertz=self.sample_rate, audio_channel_count=1, ), interim_results=True, ) + riva.client.add_word_boosting_to_config( config, self._boosted_lm_words, self._boosted_lm_score ) + riva.client.add_endpoint_parameters_to_config( config, self._start_history, @@ -193,15 +202,9 @@ class ParakeetSTTService(STTService): self._stop_threshold_eou, ) riva.client.add_custom_configuration_to_config(config, self._custom_configuration) + self._config = config - self._queue = asyncio.Queue() - - def can_generate_metrics(self) -> bool: - return False - - async def start(self, frame: StartFrame): - await super().start(frame) self._thread_task = self.create_task(self._thread_task_handler()) self._response_task = self.create_task(self._response_task_handler()) self._response_queue = asyncio.Queue() diff --git a/src/pipecat/services/xtts.py b/src/pipecat/services/xtts.py index a10247acd..7259b91d8 100644 --- a/src/pipecat/services/xtts.py +++ b/src/pipecat/services/xtts.py @@ -4,7 +4,7 @@ # SPDX-License-Identifier: BSD 2-Clause License # -from typing import Any, AsyncGenerator, Dict +from typing import Any, AsyncGenerator, Dict, Optional import aiohttp from loguru import logger @@ -76,7 +76,7 @@ class XTTSService(TTSService): base_url: str, aiohttp_session: aiohttp.ClientSession, language: Language = Language.EN, - sample_rate: int = 24000, + sample_rate: Optional[int] = None, **kwargs, ): super().__init__(sample_rate=sample_rate, **kwargs) @@ -164,18 +164,18 @@ class XTTSService(TTSService): # XTTS uses 24000 so we need to resample to our desired rate. resampled_audio = await self._resampler.resample( - bytes(process_data), 24000, self._sample_rate + bytes(process_data), 24000, self.sample_rate ) # Create the frame with the resampled audio - frame = TTSAudioRawFrame(resampled_audio, self._sample_rate, 1) + frame = TTSAudioRawFrame(resampled_audio, self.sample_rate, 1) yield frame # Process any remaining data in the buffer. if len(buffer) > 0: resampled_audio = await self._resampler.resample( - bytes(buffer), 24000, self._sample_rate + bytes(buffer), 24000, self.sample_rate ) - frame = TTSAudioRawFrame(resampled_audio, self._sample_rate, 1) + frame = TTSAudioRawFrame(resampled_audio, self.sample_rate, 1) yield frame yield TTSStoppedFrame() diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 192101e62..b18776fa8 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -35,6 +35,9 @@ class BaseInputTransport(FrameProcessor): self._params = params + # Input sample rate. It will be initialized on StartFrame. + self._sample_rate = 0 + # We read audio from a single queue one at a time and we then run VAD in # a thread. Therefore, only one thread should be necessary. self._executor = ThreadPoolExecutor(max_workers=1) @@ -43,10 +46,23 @@ class BaseInputTransport(FrameProcessor): # if passthrough is enabled. self._audio_task = None + @property + def sample_rate(self) -> int: + return self._sample_rate + + @property + def vad_analyzer(self) -> VADAnalyzer | None: + return self._params.vad_analyzer + async def start(self, frame: StartFrame): + self._sample_rate = self._params.audio_in_sample_rate or frame.audio_in_sample_rate + + # Configure VAD analyzer. + if self._params.vad_enabled and self._params.vad_analyzer: + self._params.vad_analyzer.set_sample_rate(self._sample_rate) # Start audio filter. if self._params.audio_in_filter: - await self._params.audio_in_filter.start(self._params.audio_in_sample_rate) + await self._params.audio_in_filter.start(self._sample_rate) # Create audio input queue and task if needed. if self._params.audio_in_enabled or self._params.vad_enabled: self._audio_in_queue = asyncio.Queue() @@ -67,9 +83,6 @@ class BaseInputTransport(FrameProcessor): await self.cancel_task(self._audio_task) self._audio_task = None - def vad_analyzer(self) -> VADAnalyzer | None: - return self._params.vad_analyzer - async def push_audio_frame(self, frame: InputAudioRawFrame): if self._params.audio_in_enabled or self._params.vad_enabled: await self._audio_in_queue.put(frame) @@ -104,9 +117,8 @@ class BaseInputTransport(FrameProcessor): await self.push_frame(frame, direction) await self.stop(frame) elif isinstance(frame, VADParamsUpdateFrame): - vad_analyzer = self.vad_analyzer() - if vad_analyzer: - vad_analyzer.set_params(frame.params) + if self.vad_analyzer: + self.vad_analyzer.set_params(frame.params) elif isinstance(frame, FilterUpdateSettingsFrame) and self._params.audio_in_filter: await self._params.audio_in_filter.process_frame(frame) # Other frames @@ -140,11 +152,10 @@ class BaseInputTransport(FrameProcessor): async def _vad_analyze(self, audio_frame: InputAudioRawFrame) -> VADState: state = VADState.QUIET - vad_analyzer = self.vad_analyzer() - if vad_analyzer: + if self.vad_analyzer: logger.trace(f"{self}: analyzing VAD on {audio_frame}") state = await self.get_event_loop().run_in_executor( - self._executor, vad_analyzer.analyze_audio, audio_frame.audio + self._executor, self.vad_analyzer.analyze_audio, audio_frame.audio ) logger.trace(f"{self}: done analyzing VAD on {audio_frame}") return state diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index cf3e52228..7a819b41c 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -57,12 +57,11 @@ class BaseOutputTransport(FrameProcessor): # framerate. self._camera_images = None - # We will write 20ms audio at a time. If we receive long audio frames we - # will chunk them. This will help with interruption handling. - audio_bytes_10ms = ( - int(self._params.audio_out_sample_rate / 100) * self._params.audio_out_channels * 2 - ) - self._audio_chunk_size = audio_bytes_10ms * 2 + # Output sample rate. It will be initialized on StartFrame. + self._sample_rate = 0 + + # Chunk size that will be written. It will be computed on StartFrame + self._audio_chunk_size = 0 self._audio_buffer = bytearray() self._stopped_event = asyncio.Event() @@ -70,10 +69,21 @@ class BaseOutputTransport(FrameProcessor): # Indicates if the bot is currently speaking. self._bot_speaking = False + @property + def sample_rate(self) -> int: + return self._sample_rate + async def start(self, frame: StartFrame): + self._sample_rate = self._params.audio_out_sample_rate or frame.audio_out_sample_rate + + # We will write 20ms audio at a time. If we receive long audio frames we + # will chunk them. This will help with interruption handling. + audio_bytes_10ms = int(self._sample_rate / 100) * self._params.audio_out_channels * 2 + self._audio_chunk_size = audio_bytes_10ms * 2 + # Start audio mixer. if self._params.audio_out_mixer: - await self._params.audio_out_mixer.start(self._params.audio_out_sample_rate) + await self._params.audio_out_mixer.start(self._sample_rate) self._create_camera_task() self._create_sink_tasks() @@ -298,7 +308,7 @@ class BaseOutputTransport(FrameProcessor): # Generate an audio frame with only the mixer's part. frame = OutputAudioRawFrame( audio=await self._params.audio_out_mixer.mix(silence), - sample_rate=self._params.audio_out_sample_rate, + sample_rate=self._sample_rate, num_channels=self._params.audio_out_channels, ) yield frame diff --git a/src/pipecat/transports/base_transport.py b/src/pipecat/transports/base_transport.py index cf6d624b0..f9c1eb65e 100644 --- a/src/pipecat/transports/base_transport.py +++ b/src/pipecat/transports/base_transport.py @@ -31,12 +31,12 @@ class TransportParams(BaseModel): camera_out_color_format: str = "RGB" audio_out_enabled: bool = False audio_out_is_live: bool = False - audio_out_sample_rate: int = 24000 + audio_out_sample_rate: Optional[int] = None audio_out_channels: int = 1 audio_out_bitrate: int = 96000 audio_out_mixer: Optional[BaseAudioMixer] = None audio_in_enabled: bool = False - audio_in_sample_rate: int = 16000 + audio_in_sample_rate: Optional[int] = None audio_in_channels: int = 1 audio_in_filter: Optional[BaseAudioFilter] = None vad_enabled: bool = False diff --git a/src/pipecat/transports/local/audio.py b/src/pipecat/transports/local/audio.py index 52681900d..946ce5bda 100644 --- a/src/pipecat/transports/local/audio.py +++ b/src/pipecat/transports/local/audio.py @@ -28,35 +28,40 @@ except ModuleNotFoundError as e: class LocalAudioInputTransport(BaseInputTransport): def __init__(self, py_audio: pyaudio.PyAudio, params: TransportParams): super().__init__(params) + self._py_audio = py_audio + self._in_stream = None + self._sample_rate = 0 - sample_rate = self._params.audio_in_sample_rate - num_frames = int(sample_rate / 100) * 2 # 20ms of audio + async def start(self, frame: StartFrame): + await super().start(frame) - self._in_stream = py_audio.open( - format=py_audio.get_format_from_width(2), - channels=params.audio_in_channels, - rate=params.audio_in_sample_rate, + self._sample_rate = self._params.audio_in_sample_rate or frame.audio_in_sample_rate + num_frames = int(self._sample_rate / 100) * 2 # 20ms of audio + + self._in_stream = self._py_audio.open( + format=self._py_audio.get_format_from_width(2), + channels=self._params.audio_in_channels, + rate=self._sample_rate, frames_per_buffer=num_frames, stream_callback=self._audio_in_callback, input=True, ) - - async def start(self, frame: StartFrame): - await super().start(frame) self._in_stream.start_stream() async def cleanup(self): await super().cleanup() - self._in_stream.stop_stream() - # This is not very pretty (taken from PyAudio docs). - while self._in_stream.is_active(): - await asyncio.sleep(0.1) - self._in_stream.close() + if self._in_stream: + self._in_stream.stop_stream() + # This is not very pretty (taken from PyAudio docs). + while self._in_stream.is_active(): + await asyncio.sleep(0.1) + self._in_stream.close() + self._in_stream = None def _audio_in_callback(self, in_data, frame_count, time_info, status): frame = InputAudioRawFrame( audio=in_data, - sample_rate=self._params.audio_in_sample_rate, + sample_rate=self._sample_rate, num_channels=self._params.audio_in_channels, ) @@ -68,32 +73,41 @@ class LocalAudioInputTransport(BaseInputTransport): class LocalAudioOutputTransport(BaseOutputTransport): def __init__(self, py_audio: pyaudio.PyAudio, params: TransportParams): super().__init__(params) + self._py_audio = py_audio + self._out_stream = None + self._sample_rate = 0 # We only write audio frames from a single task, so only one thread # should be necessary. self._executor = ThreadPoolExecutor(max_workers=1) - self._out_stream = py_audio.open( - format=py_audio.get_format_from_width(2), - channels=params.audio_out_channels, - rate=params.audio_out_sample_rate, - output=True, - ) - async def start(self, frame: StartFrame): await super().start(frame) + + self._sample_rate = self._params.audio_out_sample_rate or frame.audio_out_sample_rate + + self._out_stream = self._py_audio.open( + format=self._py_audio.get_format_from_width(2), + channels=self._params.audio_out_channels, + rate=self._sample_rate, + output=True, + ) self._out_stream.start_stream() async def cleanup(self): await super().cleanup() - self._out_stream.stop_stream() - # This is not very pretty (taken from PyAudio docs). - while self._out_stream.is_active(): - await asyncio.sleep(0.1) - self._out_stream.close() + if self._out_stream: + self._out_stream.stop_stream() + # This is not very pretty (taken from PyAudio docs). + while self._out_stream.is_active(): + await asyncio.sleep(0.1) + self._out_stream.close() async def write_raw_audio_frames(self, frames: bytes): - await self.get_event_loop().run_in_executor(self._executor, self._out_stream.write, frames) + if self._out_stream: + await self.get_event_loop().run_in_executor( + self._executor, self._out_stream.write, frames + ) class LocalAudioTransport(BaseTransport): diff --git a/src/pipecat/transports/local/tk.py b/src/pipecat/transports/local/tk.py index 44ca45b33..7d6fa6c75 100644 --- a/src/pipecat/transports/local/tk.py +++ b/src/pipecat/transports/local/tk.py @@ -36,35 +36,39 @@ except ModuleNotFoundError as e: class TkInputTransport(BaseInputTransport): def __init__(self, py_audio: pyaudio.PyAudio, params: TransportParams): super().__init__(params) + self._py_audio = py_audio + self._in_stream = None + self._sample_rate = 0 - sample_rate = self._params.audio_in_sample_rate - num_frames = int(sample_rate / 100) * 2 # 20ms of audio + async def start(self, frame: StartFrame): + await super().start(frame) - self._in_stream = py_audio.open( - format=py_audio.get_format_from_width(2), - channels=params.audio_in_channels, - rate=params.audio_in_sample_rate, + self._sample_rate = self._params.audio_in_sample_rate or frame.audio_in_sample_rate + num_frames = int(self._sample_rate / 100) * 2 # 20ms of audio + + self._in_stream = self._py_audio.open( + format=self._py_audio.get_format_from_width(2), + channels=self._params.audio_in_channels, + rate=self._sample_rate, frames_per_buffer=num_frames, stream_callback=self._audio_in_callback, input=True, ) - - async def start(self, frame: StartFrame): - await super().start(frame) self._in_stream.start_stream() async def cleanup(self): await super().cleanup() - self._in_stream.stop_stream() - # This is not very pretty (taken from PyAudio docs). - while self._in_stream.is_active(): - await asyncio.sleep(0.1) - self._in_stream.close() + if self._in_stream: + self._in_stream.stop_stream() + # This is not very pretty (taken from PyAudio docs). + while self._in_stream.is_active(): + await asyncio.sleep(0.1) + self._in_stream.close() def _audio_in_callback(self, in_data, frame_count, time_info, status): frame = InputAudioRawFrame( audio=in_data, - sample_rate=self._params.audio_in_sample_rate, + sample_rate=self._sample_rate, num_channels=self._params.audio_in_channels, ) @@ -76,18 +80,14 @@ class TkInputTransport(BaseInputTransport): class TkOutputTransport(BaseOutputTransport): def __init__(self, tk_root: tk.Tk, py_audio: pyaudio.PyAudio, params: TransportParams): super().__init__(params) + self._py_audio = py_audio + self._out_stream = None + self._sample_rate = 0 # We only write audio frames from a single task, so only one thread # should be necessary. self._executor = ThreadPoolExecutor(max_workers=1) - self._out_stream = py_audio.open( - format=py_audio.get_format_from_width(2), - channels=params.audio_out_channels, - rate=params.audio_out_sample_rate, - output=True, - ) - # Start with a neutral gray background. array = np.ones((1024, 1024, 3)) * 128 data = f"P5 {1024} {1024} 255 ".encode() + array.astype(np.uint8).tobytes() @@ -97,18 +97,31 @@ class TkOutputTransport(BaseOutputTransport): async def start(self, frame: StartFrame): await super().start(frame) + + self._sample_rate = self._params.audio_out_sample_rate or frame.audio_out_sample_rate + + self._out_stream = self._py_audio.open( + format=self._py_audio.get_format_from_width(2), + channels=self._params.audio_out_channels, + rate=self._sample_rate, + output=True, + ) self._out_stream.start_stream() async def cleanup(self): await super().cleanup() - self._out_stream.stop_stream() - # This is not very pretty (taken from PyAudio docs). - while self._out_stream.is_active(): - await asyncio.sleep(0.1) - self._out_stream.close() + if self._out_stream: + self._out_stream.stop_stream() + # This is not very pretty (taken from PyAudio docs). + while self._out_stream.is_active(): + await asyncio.sleep(0.1) + self._out_stream.close() async def write_raw_audio_frames(self, frames: bytes): - await self.get_event_loop().run_in_executor(self._executor, self._out_stream.write, frames) + if self._out_stream: + await self.get_event_loop().run_in_executor( + self._executor, self._out_stream.write, frames + ) async def write_frame_to_camera(self, frame: OutputImageRawFrame): self.get_event_loop().call_soon(self._write_frame_to_tk, frame) diff --git a/src/pipecat/transports/network/fastapi_websocket.py b/src/pipecat/transports/network/fastapi_websocket.py index b53daa3b4..2b3b49478 100644 --- a/src/pipecat/transports/network/fastapi_websocket.py +++ b/src/pipecat/transports/network/fastapi_websocket.py @@ -69,6 +69,7 @@ class FastAPIWebsocketInputTransport(BaseInputTransport): async def start(self, frame: StartFrame): await super().start(frame) + await self._params.serializer.setup(frame) if self._params.session_timeout: self._monitor_websocket_task = self.create_task(self._monitor_websocket()) await self._callbacks.on_client_connected(self._websocket) @@ -118,9 +119,19 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport): self._websocket = websocket self._params = params - self._send_interval = (self._audio_chunk_size / self._params.audio_out_sample_rate) / 2 + # write_raw_audio_frames() is called quickly, as soon as we get audio + # (e.g. from the TTS), and since this is just a network connection we + # would be sending it to quickly. Instead, we want to block to emulate + # an audio device, this is what the send interval is. It will be + # computed on StartFrame. + self._send_interval = 0 self._next_send_time = 0 + async def start(self, frame: StartFrame): + await super().start(frame) + await self._params.serializer.setup(frame) + self._send_interval = (self._audio_chunk_size / self.sample_rate) / 2 + async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) @@ -136,7 +147,7 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport): frame = OutputAudioRawFrame( audio=frames, - sample_rate=self._params.audio_out_sample_rate, + sample_rate=self.sample_rate, num_channels=self._params.audio_out_channels, ) diff --git a/src/pipecat/transports/network/websocket_client.py b/src/pipecat/transports/network/websocket_client.py index 56a7b57ac..a708ec5c6 100644 --- a/src/pipecat/transports/network/websocket_client.py +++ b/src/pipecat/transports/network/websocket_client.py @@ -126,6 +126,7 @@ class WebsocketClientInputTransport(BaseInputTransport): async def start(self, frame: StartFrame): await super().start(frame) + await self._params.serializer.setup(frame) await self._session.setup(frame) await self._session.connect() @@ -154,11 +155,18 @@ class WebsocketClientOutputTransport(BaseOutputTransport): self._session = session self._params = params - self._send_interval = (self._audio_chunk_size / self._params.audio_out_sample_rate) / 2 + # write_raw_audio_frames() is called quickly, as soon as we get audio + # (e.g. from the TTS), and since this is just a network connection we + # would be sending it to quickly. Instead, we want to block to emulate + # an audio device, this is what the send interval is. It will be + # computed on StartFrame. + self._send_interval = 0 self._next_send_time = 0 async def start(self, frame: StartFrame): await super().start(frame) + self._send_interval = (self._audio_chunk_size / self.sample_rate) / 2 + await self._params.serializer.setup(frame) await self._session.setup(frame) await self._session.connect() @@ -176,7 +184,7 @@ class WebsocketClientOutputTransport(BaseOutputTransport): async def write_raw_audio_frames(self, frames: bytes): frame = OutputAudioRawFrame( audio=frames, - sample_rate=self._params.audio_out_sample_rate, + sample_rate=self.sample_rate, num_channels=self._params.audio_out_channels, ) diff --git a/src/pipecat/transports/network/websocket_server.py b/src/pipecat/transports/network/websocket_server.py index 684e030ad..860f20838 100644 --- a/src/pipecat/transports/network/websocket_server.py +++ b/src/pipecat/transports/network/websocket_server.py @@ -24,7 +24,6 @@ from pipecat.frames.frames import ( ) from pipecat.processors.frame_processor import FrameDirection from pipecat.serializers.base_serializer import FrameSerializer -from pipecat.serializers.protobuf import ProtobufFrameSerializer from pipecat.transports.base_input import BaseInputTransport from pipecat.transports.base_output import BaseOutputTransport from pipecat.transports.base_transport import BaseTransport, TransportParams @@ -39,7 +38,7 @@ except ModuleNotFoundError as e: class WebsocketServerParams(TransportParams): add_wav_header: bool = False - serializer: FrameSerializer = ProtobufFrameSerializer() + serializer: FrameSerializer session_timeout: int | None = None @@ -67,20 +66,32 @@ class WebsocketServerInputTransport(BaseInputTransport): self._websocket: websockets.WebSocketServerProtocol | None = None + self._server_task = None + + # This task will monitor the websocket connection periodically. + self._monitor_task = None + self._stop_server_event = asyncio.Event() async def start(self, frame: StartFrame): await super().start(frame) + await self._params.serializer.setup(frame) self._server_task = self.create_task(self._server_task_handler()) async def stop(self, frame: EndFrame): await super().stop(frame) self._stop_server_event.set() - await self.wait_for_task(self._server_task) + if self._monitor_task: + await self.cancel_task(self._monitor_task) + if self._server_task: + await self.wait_for_task(self._server_task) async def cancel(self, frame: CancelFrame): await super().cancel(frame) - await self.cancel_task(self._server_task) + if self._monitor_task: + await self.cancel_task(self._monitor_task) + if self._server_task: + await self.cancel_task(self._server_task) async def _server_task_handler(self): logger.info(f"Starting websocket server on {self._host}:{self._port}") @@ -100,7 +111,9 @@ class WebsocketServerInputTransport(BaseInputTransport): # Create a task to monitor the websocket connection if self._params.session_timeout: - self.create_task(self._monitor_websocket(websocket)) + self._monitor_task = self.create_task( + self._monitor_websocket(websocket, self._params.session_timeout) + ) # Handle incoming messages try: @@ -125,10 +138,13 @@ class WebsocketServerInputTransport(BaseInputTransport): logger.info(f"Client {websocket.remote_address} disconnected") - async def _monitor_websocket(self, websocket: websockets.WebSocketServerProtocol): - """Wait for self._params.session_timeout seconds, if the websocket is still open, trigger timeout event.""" + async def _monitor_websocket( + self, websocket: websockets.WebSocketServerProtocol, session_timeout: int + ): + """Wait for session_timeout seconds, if the websocket is still open, + trigger timeout event.""" try: - await asyncio.sleep(self._params.session_timeout) + await asyncio.sleep(session_timeout) if not websocket.closed: await self._callbacks.on_session_timeout(websocket) except asyncio.CancelledError: @@ -144,7 +160,12 @@ class WebsocketServerOutputTransport(BaseOutputTransport): self._websocket: websockets.WebSocketServerProtocol | None = None - self._send_interval = (self._audio_chunk_size / self._params.audio_out_sample_rate) / 2 + # write_raw_audio_frames() is called quickly, as soon as we get audio + # (e.g. from the TTS), and since this is just a network connection we + # would be sending it to quickly. Instead, we want to block to emulate + # an audio device, this is what the send interval is. It will be + # computed on StartFrame. + self._send_interval = 0 self._next_send_time = 0 async def set_client_connection(self, websocket: websockets.WebSocketServerProtocol | None): @@ -153,6 +174,11 @@ class WebsocketServerOutputTransport(BaseOutputTransport): logger.warning("Only one client allowed, using new connection") self._websocket = websocket + async def start(self, frame: StartFrame): + await super().start(frame) + await self._params.serializer.setup(frame) + self._send_interval = (self._audio_chunk_size / self.sample_rate) / 2 + async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) @@ -168,7 +194,7 @@ class WebsocketServerOutputTransport(BaseOutputTransport): frame = OutputAudioRawFrame( audio=frames, - sample_rate=self._params.audio_out_sample_rate, + sample_rate=self.sample_rate, num_channels=self._params.audio_out_channels, ) @@ -213,14 +239,13 @@ class WebsocketServerOutputTransport(BaseOutputTransport): class WebsocketServerTransport(BaseTransport): def __init__( self, + params: WebsocketServerParams, host: str = "localhost", port: int = 8765, - params: WebsocketServerParams = WebsocketServerParams(), input_name: str | None = None, output_name: str | None = None, - loop: asyncio.AbstractEventLoop | None = None, ): - super().__init__(input_name=input_name, output_name=output_name, loop=loop) + super().__init__(input_name=input_name, output_name=output_name) self._host = host self._port = port self._params = params diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index ebf2b830f..13e52df2c 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -71,11 +71,11 @@ class DailyTransportMessageUrgentFrame(TransportMessageUrgentFrame): class WebRTCVADAnalyzer(VADAnalyzer): - def __init__(self, *, sample_rate=16000, num_channels=1, params: VADParams = VADParams()): - super().__init__(sample_rate=sample_rate, num_channels=num_channels, params=params) + def __init__(self, *, sample_rate: Optional[int] = None, params: VADParams = VADParams()): + super().__init__(sample_rate=sample_rate, params=params) self._webrtc_vad = Daily.create_native_vad( - reset_period_ms=VAD_RESET_PERIOD_MS, sample_rate=sample_rate, channels=num_channels + reset_period_ms=VAD_RESET_PERIOD_MS, sample_rate=self.sample_rate, channels=1 ) logger.debug("Loaded native WebRTC VAD") @@ -222,33 +222,13 @@ class DailyTransportClient(EventHandler): self._callback_queue = asyncio.Queue() self._callback_task = None + # Input and ouput sample rates. They will be initialize on setup(). + self._in_sample_rate = 0 + self._out_sample_rate = 0 + self._camera: VirtualCameraDevice | None = None - if self._params.camera_out_enabled: - self._camera = Daily.create_camera_device( - self._camera_name(), - width=self._params.camera_out_width, - height=self._params.camera_out_height, - color_format=self._params.camera_out_color_format, - ) - self._mic: VirtualMicrophoneDevice | None = None - if self._params.audio_out_enabled: - self._mic = Daily.create_microphone_device( - self._mic_name(), - sample_rate=self._params.audio_out_sample_rate, - channels=self._params.audio_out_channels, - non_blocking=True, - ) - self._speaker: VirtualSpeakerDevice | None = None - if self._params.audio_in_enabled or self._params.vad_enabled: - self._speaker = Daily.create_speaker_device( - self._speaker_name(), - sample_rate=self._params.audio_in_sample_rate, - channels=self._params.audio_in_channels, - non_blocking=True, - ) - Daily.select_speaker_device(self._speaker_name()) def _camera_name(self): return f"camera-{self}" @@ -281,7 +261,7 @@ class DailyTransportClient(EventHandler): if not self._speaker: return None - sample_rate = self._params.audio_in_sample_rate + sample_rate = self._in_sample_rate num_channels = self._params.audio_in_channels num_frames = int(sample_rate / 100) * 2 # 20ms of audio @@ -315,6 +295,34 @@ class DailyTransportClient(EventHandler): self._camera.write_frame(frame.image) async def setup(self, frame: StartFrame): + self._in_sample_rate = self._params.audio_in_sample_rate or frame.audio_in_sample_rate + self._out_sample_rate = self._params.audio_out_sample_rate or frame.audio_out_sample_rate + + if self._params.camera_out_enabled and not self._camera: + self._camera = Daily.create_camera_device( + self._camera_name(), + width=self._params.camera_out_width, + height=self._params.camera_out_height, + color_format=self._params.camera_out_color_format, + ) + + if self._params.audio_out_enabled and not self._mic: + self._mic = Daily.create_microphone_device( + self._mic_name(), + sample_rate=self._out_sample_rate, + channels=self._params.audio_out_channels, + non_blocking=True, + ) + + if (self._params.audio_in_enabled or self._params.vad_enabled) and not self._speaker: + self._speaker = Daily.create_speaker_device( + self._speaker_name(), + sample_rate=self._in_sample_rate, + channels=self._params.audio_in_channels, + non_blocking=True, + ) + Daily.select_speaker_device(self._speaker_name()) + if not self._task_manager: self._task_manager = frame.task_manager self._callback_task = self._task_manager.create_task( @@ -707,6 +715,7 @@ class DailyInputTransport(BaseInputTransport): super().__init__(params, **kwargs) self._client = client + self._params = params self._video_renderers = {} @@ -715,11 +724,10 @@ class DailyInputTransport(BaseInputTransport): self._audio_in_task = None self._vad_analyzer: VADAnalyzer | None = params.vad_analyzer - if params.vad_enabled and not params.vad_analyzer: - self._vad_analyzer = WebRTCVADAnalyzer( - sample_rate=self._params.audio_in_sample_rate, - num_channels=self._params.audio_in_channels, - ) + + @property + def vad_analyzer(self) -> VADAnalyzer | None: + return self._vad_analyzer async def start(self, frame: StartFrame): # Parent start. @@ -728,6 +736,9 @@ class DailyInputTransport(BaseInputTransport): await self._client.setup(frame) # Join the room. await self._client.join() + # Inialize WebRTC VAD if needed. + if self._params.vad_enabled and not self._params.vad_analyzer: + self._vad_analyzer = WebRTCVADAnalyzer(sample_rate=self.sample_rate) # Create audio task. It reads audio frames from Daily and push them # internally for VAD processing. if self._params.audio_in_enabled or self._params.vad_enabled: @@ -757,9 +768,6 @@ class DailyInputTransport(BaseInputTransport): await super().cleanup() await self._client.cleanup() - def vad_analyzer(self) -> VADAnalyzer | None: - return self._vad_analyzer - # # FrameProcessor # diff --git a/src/pipecat/transports/services/livekit.py b/src/pipecat/transports/services/livekit.py index 11926f86c..925f7dae1 100644 --- a/src/pipecat/transports/services/livekit.py +++ b/src/pipecat/transports/services/livekit.py @@ -101,6 +101,7 @@ class LiveKitTransportClient: return self._room async def setup(self, frame: StartFrame): + self._out_sample_rate = self._params.audio_out_sample_rate or frame.audio_out_sample_rate if not self._task_manager: self._task_manager = frame.task_manager self._room = rtc.Room(loop=self._task_manager.get_event_loop()) @@ -138,7 +139,7 @@ class LiveKitTransportClient: # Set up audio source and track self._audio_source = rtc.AudioSource( - self._params.audio_out_sample_rate, self._params.audio_out_channels + self._out_sample_rate, self._params.audio_out_channels ) self._audio_track = rtc.LocalAudioTrack.create_audio_track( "pipecat-audio", self._audio_source @@ -351,6 +352,10 @@ class LiveKitInputTransport(BaseInputTransport): self._vad_analyzer: VADAnalyzer | None = params.vad_analyzer self._resampler = create_default_resampler() + @property + def vad_analyzer(self) -> VADAnalyzer | None: + return self._vad_analyzer + async def start(self, frame: StartFrame): await super().start(frame) await self._client.setup(frame) @@ -372,9 +377,6 @@ class LiveKitInputTransport(BaseInputTransport): if self._audio_in_task and (self._params.audio_in_enabled or self._params.vad_enabled): await self.cancel_task(self._audio_in_task) - def vad_analyzer(self) -> VADAnalyzer | None: - return self._vad_analyzer - async def push_app_message(self, message: Any, sender: str): frame = LiveKitTransportMessageUrgentFrame(message=message, participant_id=sender) await self.push_frame(frame) @@ -401,12 +403,12 @@ class LiveKitInputTransport(BaseInputTransport): audio_frame = audio_frame_event.frame audio_data = await self._resampler.resample( - audio_frame.data.tobytes(), audio_frame.sample_rate, self._params.audio_in_sample_rate + audio_frame.data.tobytes(), audio_frame.sample_rate, self.sample_rate ) return AudioRawFrame( audio=audio_data, - sample_rate=self._params.audio_in_sample_rate, + sample_rate=self.sample_rate, num_channels=audio_frame.num_channels, ) @@ -448,7 +450,7 @@ class LiveKitOutputTransport(BaseOutputTransport): return rtc.AudioFrame( data=pipecat_audio, - sample_rate=self._params.audio_out_sample_rate, + sample_rate=self.sample_rate, num_channels=self._params.audio_out_channels, samples_per_channel=samples_per_channel, )