feat: add 'frame_delay' to stats

This commit adds the frame delay between the processing time and the
expected presentation time to the `/stats` endpoints.

Author: rickstaa

server/app.py

@@ -22,8 +22,9 @@
 import time
 
 logger = logging.getLogger(__name__)
-logging.getLogger('aiortc.rtcrtpsender').setLevel(logging.WARNING)
-logging.getLogger('aiortc.rtcrtpreceiver').setLevel(logging.WARNING)
+logging.getLogger("aiortc").setLevel(logging.DEBUG)
+logging.getLogger("aiortc.rtcrtpsender").setLevel(logging.WARNING)
+logging.getLogger("aiortc.rtcrtpreceiver").setLevel(logging.WARNING)
 
 
 MAX_BITRATE = 2000000
@@ -51,36 +52,60 @@ def __init__(self, track: MediaStreamTrack, pipeline: Pipeline):
         super().__init__()
         self.track = track
         self.pipeline = pipeline
-        self._frame_count = 0
-        self._start_time = time.monotonic()
         self._lock = threading.Lock()
         self._fps = 0.0
+        self._frame_delay = 0.0
         self._running = True
+        self._fps_interval_frame_count = 0
+        self._last_fps_calculation_time = time.monotonic()
+        self._stream_start_time = None
+        self._last_frame_presentation_time = None
+        self._last_frame_processed_time = None
         self._start_fps_thread()
+        self._start_frame_delay_thread()
 
     def _start_fps_thread(self):
         """Start a separate thread to calculate FPS periodically."""
-        self.fps_thread = threading.Thread(target=self._calculate_fps_loop, daemon=True)
-        self.fps_thread.start()
+        self._fps_thread = threading.Thread(
+            target=self._calculate_fps_loop, daemon=True
+        )
+        self._fps_thread.start()
 
     def _calculate_fps_loop(self):
         """Loop to calculate FPS periodically."""
         while self._running:
             time.sleep(1)  # Calculate FPS every second.
             with self._lock:
                 current_time = time.monotonic()
-                time_diff = current_time - self._start_time
+                time_diff = current_time - self._last_fps_calculation_time
                 if time_diff > 0:
-                    self._fps = self._frame_count / time_diff
+                    self._fps = self._fps_interval_frame_count / time_diff
 
                     # Reset start_time and frame_count for the next interval.
-                    self._start_time = current_time
-                    self._frame_count = 0
+                    self._last_fps_calculation_time = current_time
+                    self._fps_interval_frame_count = 0
+
+    def _start_frame_delay_thread(self):
+        """Start a separate thread to calculate frame delay periodically."""
+        self._frame_delay_thread = threading.Thread(
+            target=self._calculate_frame_delay_loop, daemon=True
+        )
+        self._frame_delay_thread.start()
+
+    def _calculate_frame_delay_loop(self):
+        """Loop to calculate frame delay periodically."""
+        while self._running:
+            time.sleep(1)  # Calculate frame delay every second.
+            with self._lock:
+                if self._last_frame_presentation_time is not None:
+                    current_time = time.monotonic()
+                    self._frame_delay = (current_time - self._stream_start_time ) - float(self._last_frame_presentation_time)
 
     def stop(self):
         """Stop the FPS calculation thread."""
         self._running = False
-        self.fps_thread.join()
+        self._fps_thread.join()
+        self._frame_delay_thread.join()
 
     @property
     def fps(self) -> float:
@@ -92,19 +117,34 @@ def fps(self) -> float:
         with self._lock:
             return self._fps
 
+    @property
+    def frame_delay(self) -> float:
+        """Get the current frame delay.
+
+        Returns:
+            The current frame delay.
+        """
+        with self._lock:
+            return self._frame_delay
+
     async def recv(self) -> av.VideoFrame:
         """Receive and process a video frame. Called by the WebRTC library when a frame
         is received.
 
         Returns:
             The processed video frame.
         """
+        if self._stream_start_time is None:
+            self._stream_start_time = time.monotonic()
+
         input_frame = await self.track.recv()
         processed_frame = await self.pipeline(input_frame)
 
-        # Increment frame count for FPS calculation.
+        # Store frame info for stats.
         with self._lock:
-            self._frame_count += 1
+            self._fps_interval_frame_count += 1
+            self._last_frame_presentation_time = input_frame.time
+            self._last_frame_processed_time = time.monotonic()
 
         return processed_frame
 
@@ -187,30 +227,29 @@ async def offer(request):
     @pc.on("datachannel")
     def on_datachannel(channel):
         if channel.label == "control":
+
             @channel.on("message")
             async def on_message(message):
                 try:
                     params = json.loads(message)
-                    
+
                     if params.get("type") == "get_nodes":
                         nodes_info = await pipeline.get_nodes_info()
-                        response = {
-                            "type": "nodes_info",
-                            "nodes": nodes_info
-                        }
+                        response = {"type": "nodes_info", "nodes": nodes_info}
                         channel.send(json.dumps(response))
                     elif params.get("type") == "update_prompt":
                         if "prompt" not in params:
-                            logger.warning("[Control] Missing prompt in update_prompt message")
+                            logger.warning(
+                                "[Control] Missing prompt in update_prompt message"
+                            )
                             return
                         pipeline.set_prompt(params["prompt"])
-                        response = {
-                            "type": "prompt_updated",
-                            "success": True
-                        }
+                        response = {"type": "prompt_updated", "success": True}
                         channel.send(json.dumps(response))
                     else:
-                        logger.warning("[Server] Invalid message format - missing required fields")
+                        logger.warning(
+                            "[Server] Invalid message format - missing required fields"
+                        )
                 except json.JSONDecodeError:
                     logger.error("[Server] Invalid JSON received")
                 except Exception as e:
@@ -309,8 +348,8 @@ async def on_shutdown(app: web.Application):
 
     logging.basicConfig(
         level=args.log_level.upper(),
-        format='%(asctime)s [%(levelname)s] %(message)s',
-        datefmt='%H:%M:%S'
+        format="%(asctime)s [%(levelname)s] %(message)s",
+        datefmt="%H:%M:%S",
     )
 
     app = web.Application()

server/utils.py

@@ -85,6 +85,7 @@ def get_video_track_stats(self, video_track: MediaStreamTrack) -> Dict[str, Any]
         """
         return {
             "fps": video_track.fps,
+            "frame_delay": video_track.frame_delay,
         }
 
     async def get_stats(self, _) -> web.Response: