Recording the audios that we are receiving.

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)

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