From 359b55a85e2371e744549070e3eab8ebff1d9146 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= <aleix@daily.co>
Date: Thu, 16 Jan 2025 11:19:42 -0800
Subject: [PATCH] additional trace logging

---
 src/pipecat/processors/frame_processor.py | 12 ++++++++----
 src/pipecat/services/cartesia.py          |  2 +-
 src/pipecat/services/tavus.py             |  2 +-
 src/pipecat/transports/base_input.py      | 12 +++++++-----
 4 files changed, 17 insertions(+), 11 deletions(-)

diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py
index 653b313b8..5277f2bc8 100644
--- a/src/pipecat/processors/frame_processor.py
+++ b/src/pipecat/processors/frame_processor.py
@@ -181,9 +181,11 @@ class FrameProcessor:
             await self.__input_queue.put((frame, direction, callback))
 
     async def pause_processing_frames(self):
+        logger.trace(f"{self}: pausing frame processing")
         self.__should_block_frames = True
 
     async def resume_processing_frames(self):
+        logger.trace("f{self}: resuming frame processing")
         self.__input_event.set()
         self.__should_block_frames = False
 
@@ -284,8 +286,10 @@ class FrameProcessor:
         while running:
             try:
                 if self.__should_block_frames:
+                    logger.trace(f"{self}: frame processing paused")
                     await self.__input_event.wait()
                     self.__input_event.clear()
+                    logger.trace(f"{self}: frame processing resumed")
 
                 (frame, direction, callback) = await self.__input_queue.get()
 
@@ -300,10 +304,10 @@ class FrameProcessor:
 
                 self.__input_queue.task_done()
             except asyncio.CancelledError:
-                logger.trace(f"Cancelled input task in {self}")
+                logger.trace(f"{self}: cancelled input task")
                 break
             except Exception as e:
-                logger.exception(f"Uncaught exception in {self}: {e}")
+                logger.exception(f"{self}: Uncaught exception {e}")
                 await self.push_error(ErrorFrame(str(e)))
 
     def __create_push_task(self):
@@ -323,10 +327,10 @@ class FrameProcessor:
                 running = not isinstance(frame, EndFrame)
                 self.__push_queue.task_done()
             except asyncio.CancelledError:
-                logger.trace(f"Cancelled push task in {self}")
+                logger.trace(f"{self}: cancelled push task")
                 break
             except Exception as e:
-                logger.exception(f"Uncaught exception in {self}: {e}")
+                logger.exception(f"{self}: Uncaught exception {e}")
                 await self.push_error(ErrorFrame(str(e)))
 
     async def _call_event_handler(self, event_name: str, *args, **kwargs):
diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py
index 67a10b1cf..a91c06176 100644
--- a/src/pipecat/services/cartesia.py
+++ b/src/pipecat/services/cartesia.py
@@ -238,7 +238,7 @@ class CartesiaTTSService(WordTTSService, WebsocketService):
     async def flush_audio(self):
         if not self._context_id or not self._websocket:
             return
-        logger.trace("Flushing audio")
+        logger.trace(f"{self}: flushing audio")
         msg = self._build_msg(text="", continue_transcript=False)
         await self._websocket.send(msg)
 
diff --git a/src/pipecat/services/tavus.py b/src/pipecat/services/tavus.py
index 32752205c..7b43f6176 100644
--- a/src/pipecat/services/tavus.py
+++ b/src/pipecat/services/tavus.py
@@ -88,7 +88,7 @@ class TavusVideoService(AIService):
         if not done:
             audio = resample_audio(audio, original_sample_rate, 16000)
         audio_base64 = base64.b64encode(audio).decode("utf-8")
-        logger.trace(f"TavusVideoService sending {len(audio)} bytes")
+        logger.trace(f"{self}: sending {len(audio)} bytes")
         await self._send_audio_message(audio_base64, done=done)
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py
index b0d6a7953..994bd1351 100644
--- a/src/pipecat/transports/base_input.py
+++ b/src/pipecat/transports/base_input.py
@@ -138,17 +138,19 @@ class BaseInputTransport(FrameProcessor):
     # Audio input
     #
 
-    async def _vad_analyze(self, audio_frames: bytes) -> VADState:
+    async def _vad_analyze(self, audio_frame: InputAudioRawFrame) -> VADState:
         state = VADState.QUIET
         vad_analyzer = self.vad_analyzer()
         if 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_frames
+                self._executor, vad_analyzer.analyze_audio, audio_frame.audio
             )
+            logger.trace(f"{self}: done analyzing VAD on {audio_frame}")
         return state
 
-    async def _handle_vad(self, audio_frames: bytes, vad_state: VADState):
-        new_vad_state = await self._vad_analyze(audio_frames)
+    async def _handle_vad(self, audio_frame: InputAudioRawFrame, vad_state: VADState):
+        new_vad_state = await self._vad_analyze(audio_frame)
         if (
             new_vad_state != vad_state
             and new_vad_state != VADState.STARTING
@@ -181,7 +183,7 @@ class BaseInputTransport(FrameProcessor):
                 # Check VAD and push event if necessary. We just care about
                 # changes from QUIET to SPEAKING and vice versa.
                 if self._params.vad_enabled:
-                    vad_state = await self._handle_vad(frame.audio, vad_state)
+                    vad_state = await self._handle_vad(frame, vad_state)
                     audio_passthrough = self._params.vad_audio_passthrough
 
                 # Push audio downstream if passthrough.