observers: add a timestamp to on_push_frame()

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.

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:

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}")

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: