Decode video frames at the publish rate, not the stream rate

inorbit_edge/tests/test_video.py

@@ -5,6 +5,7 @@
 import time
 
 import cv2
+import numpy
 
 from inorbit_edge.robot import RobotSession
 from inorbit_edge.video import CameraStreamer, OpenCVCamera
@@ -44,19 +45,25 @@ def get_frame_jpg(self):
 
 
 class FakeCapture:
-    """Minimal cv2.VideoCapture double that counts grabs."""
+    """Minimal cv2.VideoCapture double that counts grabs and retrieves."""
 
-    def __init__(self, grab_ok=False):
+    def __init__(self, grab_ok=False, frame=None, delay=0.0):
         self.grab_ok = grab_ok
+        self.frame = frame
+        self.delay = delay
         self.grabs = 0
+        self.retrieves = 0
         self.released = False
 
     def grab(self):
         self.grabs += 1
+        if self.delay:
+            time.sleep(self.delay)  # a real grab is paced by the stream
         return self.grab_ok
 
     def retrieve(self):
-        return False, None
+        self.retrieves += 1
+        return self.frame is not None, self.frame
 
     def release(self):
         self.released = True
@@ -87,6 +94,28 @@ def fake_open_capture():
     assert captures[0].released
 
 
+def test_frames_are_decoded_at_the_publish_rate_not_the_stream_rate(mocker):
+    """Every frame is drained with grab(); only published ones are decoded."""
+    frame = numpy.zeros((16, 16, 3), dtype=numpy.uint8)
+    capture = FakeCapture(grab_ok=True, frame=frame, delay=0.002)
+    camera = OpenCVCamera("rtsp://camera.invalid/stream", rate=1, scaling=0.5)
+    mocker.patch.object(camera, "_open_capture", return_value=capture)
+
+    camera.open()
+    try:
+        time.sleep(0.6)
+        jpg, _w, _h, _ts = camera.get_frame_jpg()
+        retrieves_after_publishing = capture.retrieves
+    finally:
+        camera.close()
+
+    assert capture.grabs > 20
+    assert retrieves_after_publishing <= 3, "decoding every grabbed frame"
+    assert jpg is not None
+    # Publishing reads the buffered frame; it never touches the capture.
+    assert capture.retrieves == retrieves_after_publishing
+
+
 def test_get_frame_jpg_returns_no_frame_while_the_capture_is_reopening():
     camera = OpenCVCamera("rtsp://camera.invalid/stream", rate=1)
 

inorbit_edge/video.py

@@ -89,6 +89,15 @@ def __init__(
         # Set by close() so a capture thread waiting out a reopen backoff wakes
         # up immediately instead of holding up teardown.
         self._closing = threading.Event()
+        # Latest decoded frame as (frame, monotonic timestamp). Published by the
+        # capture thread, read by get_frame_jpg(); a reference assignment is
+        # atomic under the GIL, so no lock is involved.
+        self._frame = None
+        # grab() decodes the frame; retrieve() adds the YUV->BGR conversion and a
+        # copy on top (4.5ms vs 7.1ms per frame at 1080p). Only one frame per
+        # publish is ever used, so convert at twice the publish rate and let the
+        # rest of the stream drain through grab() alone.
+        self._retrieve_interval = 0.5 / max(rate, 1)
 
     def _open_capture(self):
         """Return a new ``cv2.VideoCapture`` for this camera's source."""
@@ -100,6 +109,7 @@ def _open_capture(self):
     def open(self):
         """Opens the capturing device / stream"""
         self._closing.clear()
+        self._frame = None
         with self.capture_mutex:
             if self.capture is None:
                 self.capture = self._open_capture()
@@ -125,27 +135,27 @@ def close(self):
     def get_frame_jpg(self):
         """Returns the latest frame captured by the camera as JPG"""
         ts = time.time() * 1000
-        with self.capture_mutex:
-            if self.capture is None:
-                # No capture between a failed grab and its reopen
-                return None, 0, 0, ts
-            # decode the latest grabbed frame
-            ret, frame = self.capture.retrieve()
-            if not ret:
-                return None, 0, 0, ts
-            width = self.capture.get(cv2.CAP_PROP_FRAME_WIDTH)
-            height = self.capture.get(cv2.CAP_PROP_FRAME_HEIGHT)
-            jpg, w, h = convert_frame(frame, width, height, self.scaling, self.quality)
-            return jpg, w, h, ts
+        frame = self._frame
+        if frame is None:
+            return None, 0, 0, ts
+        height, width = frame[0].shape[:2]
+        jpg, w, h = convert_frame(frame[0], width, height, self.scaling, self.quality)
+        return jpg, w, h, ts
 
     def _run(self):
-        """Thread to grab always the most recent frame"""
+        """Thread to grab the most recent frame, decoding at the publish rate
+
+        Only the capture thread touches ``self.capture``, so the grab loop does
+        not hold ``capture_mutex``: holding it across every grab starved
+        ``get_frame_jpg()``, which could then wait seconds for a frame that was
+        already decoded.
+        """
         failures = 0
+        next_retrieve = 0.0
         while self.running:
             try:
-                with self.capture_mutex:
-                    # Try to grab always the latest frame
-                    grabbed = self.capture is not None and self.capture.grab()
+                # Try to grab always the latest frame
+                grabbed = self.capture is not None and self.capture.grab()
             except Exception as e:
                 self.logger.error(f"Failed to grab video frame {e}")
                 grabbed = False
@@ -155,6 +165,17 @@ def _run(self):
                         f"Video stream recovered after {failures} failed grabs"
                     )
                     failures = 0
+                now = time.monotonic()
+                if now < next_retrieve:
+                    continue
+                try:
+                    retrieved, frame = self.capture.retrieve()
+                except Exception as e:
+                    self.logger.error(f"Failed to decode video frame {e}")
+                    retrieved = False
+                if retrieved:
+                    self._frame = (frame, now)
+                    next_retrieve = now + self._retrieve_interval
                 continue
             # A grab against a dead stream fails immediately, so without a
             # reopen this loop spins at 100% of a core for as long as the stream