wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #1017 from pipecat-ai/aleix/additional-trace-logging

Browse Source 
additional trace logging
This commit is contained in:
Aleix Conchillo Flaqué
2025-01-16 12:38:47 -08:00
committed by GitHub
parent c211580fec 359b55a85e
commit 73ed0c1ad7
4 changed files with 17 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
src/pipecat/processors/frame_processor.py
View File

@@ -182,9 +182,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
@@ -295,8 +297,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()
@@ -311,10 +315,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):
@@ -334,10 +338,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):

2
src/pipecat/services/cartesia.py
View File

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

2
src/pipecat/services/tavus.py
View File

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

12
src/pipecat/transports/base_input.py
View File

@@ -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.