From 7c61c36825816ddf752467e65deef735636516f0 Mon Sep 17 00:00:00 2001 From: filipi87 Date: Wed, 20 May 2026 19:03:01 -0300 Subject: [PATCH] Recording the audios that we are receiving. --- .../video-avatar-tavus-video-service.py | 40 ++++++++++++++++++- scripts/daily/test_tavus_transport.py | 31 ++++++++++++++ 2 files changed, 70 insertions(+), 1 deletion(-) diff --git a/examples/video-avatar/video-avatar-tavus-video-service.py b/examples/video-avatar/video-avatar-tavus-video-service.py index ce3075074..bdef3a856 100644 --- a/examples/video-avatar/video-avatar-tavus-video-service.py +++ b/examples/video-avatar/video-avatar-tavus-video-service.py @@ -5,8 +5,12 @@ # +import datetime +import io import os +import wave +import aiofiles import aiohttp from dotenv import load_dotenv from loguru import logger @@ -21,6 +25,7 @@ from pipecat.processors.aggregators.llm_response_universal import ( LLMContextAggregatorPair, LLMUserAggregatorParams, ) +from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import create_transport from pipecat.services.cartesia.tts import CartesiaTTSService @@ -32,6 +37,21 @@ from pipecat.transports.daily.transport import DailyParams load_dotenv(override=True) + +async def save_audio_file(audio: bytes, filename: str, sample_rate: int, num_channels: int): + """Save audio data to a WAV file.""" + if len(audio) > 0: + with io.BytesIO() as buffer: + with wave.open(buffer, "wb") as wf: + wf.setsampwidth(2) + wf.setnchannels(num_channels) + wf.setframerate(sample_rate) + wf.writeframes(audio) + async with aiofiles.open(filename, "wb") as file: + await file.write(buffer.getvalue()) + logger.info(f"Audio saved to {filename}") + + # We use lambdas to defer transport parameter creation until the transport # type is selected at runtime. transport_params = { @@ -59,7 +79,7 @@ transport_params = { async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): logger.info(f"Starting bot") async with aiohttp.ClientSession() as session: - stt = DeepgramSTTService(api_key=os.environ["DEEPGRAM_API_KEY"]) + stt = DeepgramSTTService(api_key=os.environ["DEEPGRAM_API_KEY"], audio_passthrough=True) tts = CartesiaTTSService( api_key=os.environ["CARTESIA_API_KEY"], @@ -87,6 +107,8 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()), ) + audiobuffer = AudioBufferProcessor() + pipeline = Pipeline( [ transport.input(), # Transport user input @@ -96,6 +118,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): tts, # TTS tavus, # Tavus output layer transport.output(), # Transport bot output + audiobuffer, # Audio recording assistant_aggregator, # Assistant spoken responses ] ) @@ -114,6 +137,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): @transport.event_handler("on_client_connected") async def on_client_connected(transport, client): logger.info(f"Client connected") + await audiobuffer.start_recording() # Kick off the conversation. context.add_message( { @@ -128,6 +152,20 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): logger.info(f"Client disconnected") await task.cancel() + @audiobuffer.event_handler("on_audio_data") + async def on_audio_data(buffer, audio, sample_rate, num_channels): + timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") + filename = f"recordings/merged_{timestamp}.wav" + os.makedirs("recordings", exist_ok=True) + await save_audio_file(audio, filename, sample_rate, num_channels) + + @audiobuffer.event_handler("on_track_audio_data") + async def on_track_audio_data(buffer, user_audio, bot_audio, sample_rate, num_channels): + timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") + os.makedirs("recordings", exist_ok=True) + await save_audio_file(user_audio, f"recordings/user_{timestamp}.wav", sample_rate, 1) + await save_audio_file(bot_audio, f"recordings/bot_{timestamp}.wav", sample_rate, 1) + runner = PipelineRunner(handle_sigint=runner_args.handle_sigint) await runner.run(task) diff --git a/scripts/daily/test_tavus_transport.py b/scripts/daily/test_tavus_transport.py index fd5a55de0..6f542ef63 100644 --- a/scripts/daily/test_tavus_transport.py +++ b/scripts/daily/test_tavus_transport.py @@ -1,7 +1,10 @@ import array import asyncio +import datetime +import io import os import signal +import wave from daily import ( AudioData, @@ -54,6 +57,8 @@ class DailyProxyApp(EventHandler): # Raw PCM buffer — filled at DECLARED_SAMPLE_RATE speed, drained at TRUE_SAMPLE_RATE speed. self._buffer = bytearray() self._audio_task: asyncio.Task | None = None + self._wav_file: wave.Wave_write | None = None + self._wav_io: io.FileIO | None = None self._client: CallClient = CallClient(event_handler=self) self._client.update_subscription_profiles( @@ -70,8 +75,31 @@ class DailyProxyApp(EventHandler): print(f"Unable to join meeting: {error}") self._loop.call_soon_threadsafe(self._loop.stop) + def _open_wav(self): + os.makedirs("recordings", exist_ok=True) + timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") + path = f"recordings/received_{timestamp}.wav" + self._wav_io = open(path, "wb") + self._wav_file = wave.open(self._wav_io, "wb") + self._wav_file.setnchannels(1) + self._wav_file.setsampwidth(2) + # Declare TRUE_SAMPLE_RATE so timestamps match bot_*.wav for comparison. + # Bytes arrive at DECLARED_SAMPLE_RATE speed (2x real-time) but each byte + # is 24kHz content, so the WAV plays back at normal speed. + self._wav_file.setframerate(TRUE_SAMPLE_RATE) + logger.info(f"Recording received audio to {path}") + + def _close_wav(self): + if self._wav_file: + self._wav_file.close() + self._wav_file = None + if self._wav_io: + self._wav_io.close() + self._wav_io = None + def run(self, meeting_url: str): asyncio.set_event_loop(self._loop) + self._open_wav() self._create_audio_task() def handle_exit(): @@ -104,6 +132,7 @@ class DailyProxyApp(EventHandler): if self._audio_task: self._loop.run_until_complete(self._cancel_audio_task()) + self._close_wav() self._client.leave() self._client.release() @@ -179,6 +208,8 @@ class DailyProxyApp(EventHandler): self._buffer.extend(new_bytes) def _audio_data_received(self, participant_id: str, audio_data: AudioData, audio_source: str): + if self._wav_file: + self._wav_file.writeframes(audio_data.audio_frames) asyncio.run_coroutine_threadsafe(self._buffer_audio(audio_data), self._loop) async def _handle_interrupt(self):