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

processors: allow passing a callback when queued frame is processed

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2024-10-20 22:35:46 -07:00
parent 22ca4c5a02
commit 2eccb33e73
1 changed files with 13 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -8,6 +8,7 @@ import asyncio
import inspect import inspect
from enum import Enum from enum import Enum
from typing import Awaitable, Callable, Optional
from pipecat.clocks.base_clock import BaseClock from pipecat.clocks.base_clock import BaseClock
from pipecat.frames.frames import ( from pipecat.frames.frames import (
@@ -155,14 +156,19 @@ class FrameProcessor:
return self._clock return self._clock
async def queue_frame( async def queue_frame(
self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM self,
frame: Frame,
direction: FrameDirection = FrameDirection.DOWNSTREAM,
callback: Optional[
Callable[["FrameProcessor", Frame, FrameDirection], Awaitable[None]]
] = None,
): ):
if isinstance(frame, SystemFrame): if isinstance(frame, SystemFrame):
# We don't want to queue system frames. # We don't want to queue system frames.
await self.process_frame(frame, direction) await self.process_frame(frame, direction)
else: else:
# We queue everything else. # We queue everything else.
await self.__input_queue.put((frame, direction)) await self.__input_queue.put((frame, direction, callback))
async def resume_processing_frames(self): async def resume_processing_frames(self):
self.__input_event.set() self.__input_event.set()
@@ -259,11 +265,15 @@ class FrameProcessor:
self.__input_event.clear() self.__input_event.clear()
should_block_frames = False should_block_frames = False
(frame, direction) = await self.__input_queue.get() (frame, direction, callback) = await self.__input_queue.get()
# Process the frame. # Process the frame.
await self.process_frame(frame, direction) await self.process_frame(frame, direction)
# If this frame has an associated callback, call it now.
if callback:
await callback(self, frame, direction)
# Check if we should block incoming frames from now on (until we # Check if we should block incoming frames from now on (until we
# resume processing). # resume processing).
if isinstance(frame, self._block_on_frames): if isinstance(frame, self._block_on_frames):