diff --git a/src/pipecat/observers/base_observer.py b/src/pipecat/observers/base_observer.py index 2a17b5668..46f746946 100644 --- a/src/pipecat/observers/base_observer.py +++ b/src/pipecat/observers/base_observer.py @@ -20,7 +20,12 @@ class BaseObserver(ABC): @abstractmethod async def on_push_frame( - self, src: FrameProcessor, dst: FrameProcessor, frame: Frame, direction: FrameDirection + self, + src: FrameProcessor, + dst: FrameProcessor, + frame: Frame, + direction: FrameDirection, + timestamp: int, ): """Abstract method to handle the event when a frame is pushed from one processor to another. @@ -30,6 +35,7 @@ class BaseObserver(ABC): dst (FrameProcessor): The destination frame processor that will receive the frame. frame (Frame): The frame being transferred between processors. direction (FrameDirection): The direction of the frame transfer. + timestamp (int): The timestamp when the frame was pushed (based on the pipeline clock). This method should be implemented by subclasses to define specific behavior when a frame is pushed. diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index db9a00049..fc404f75b 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -109,10 +109,15 @@ class Observer(BaseObserver): self._observers = observers async def on_push_frame( - self, src: FrameProcessor, dst: FrameProcessor, frame: Frame, direction: FrameDirection + self, + src: FrameProcessor, + dst: FrameProcessor, + frame: Frame, + direction: FrameDirection, + timestamp: int, ): for observer in self._observers: - await observer.on_push_frame(src, dst, frame, direction) + await observer.on_push_frame(src, dst, frame, direction, timestamp) class PipelineTask: diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index ed6f8a4cb..302ec06d1 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -258,15 +258,20 @@ class FrameProcessor: async def __internal_push_frame(self, frame: Frame, direction: FrameDirection): try: + timestamp = self._clock.get_time() if direction == FrameDirection.DOWNSTREAM and self._next: logger.trace(f"Pushing {frame} from {self} to {self._next}") if self._observer: - await self._observer.on_push_frame(self, self._next, frame, direction) + await self._observer.on_push_frame( + self, self._next, frame, direction, timestamp + ) await self._next.queue_frame(frame, direction) elif direction == FrameDirection.UPSTREAM and self._prev: logger.trace(f"Pushing {frame} upstream from {self} to {self._prev}") if self._observer: - await self._observer.on_push_frame(self, self._prev, frame, direction) + await self._observer.on_push_frame( + self, self._prev, frame, direction, timestamp + ) await self._prev.queue_frame(frame, direction) except Exception as e: logger.exception(f"Uncaught exception in {self}: {e}") diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index fa72d5499..a689b93ad 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -579,7 +579,12 @@ class RTVIObserver(BaseObserver): self._frames_seen = set() async def on_push_frame( - self, src: FrameProcessor, dst: FrameProcessor, frame: Frame, direction: FrameDirection + self, + src: FrameProcessor, + dst: FrameProcessor, + frame: Frame, + direction: FrameDirection, + timestamp: int, ): # If we have already seen this frame, let's skip it. if frame.id in self._frames_seen: