Improve behaviour of CircularOutput2

This was previously a bit prone to throwing away occasional frames as
"expired" if timestamps were uneven. Also the duration held in the
buffer could shrink if an output had to wait a while for the first I
frame. The new behaviour should be more reliable.

Signed-off-by: David Plowman <[email protected]>

Author: davidplowman

picamera2/outputs/circularoutput2.py

@@ -24,7 +24,7 @@ def __init__(self, pts=None, buffer_duration_ms=5000):
         # A note on locking. The lock is principally to protect outputframe, which is called by
         # the background encoder thread. Applications are going to call things like open_output,
         # close_output, start and stop. These only grab that lock for a short period of time to
-        # manipulate _output_available, which controls whether outputframe will do anything.
+        # manipulate _output, which controls whether outputframe will do anything.
         # THe application API does not have it's own lock, because there doesn't seem to be a
         # need to drive it from different threads (though we could add one if necessary).
         self._lock = Lock()
@@ -33,7 +33,6 @@ def __init__(self, pts=None, buffer_duration_ms=5000):
         self._buffer_duration_ms = buffer_duration_ms
         self._circular = collections.deque()
         self._output = None
-        self._output_available = False
         self._streams = []
 
     @property
@@ -52,15 +51,14 @@ def open_output(self, output):
         if self._output:
             raise RuntimeError("Underlying output must be closed first")
 
-        self._output = output
-        self._output.start()
+        output.start()
         # Some outputs (PyavOutput) may need to know about the encoder's streams.
         for encoder_stream, codec, kwargs in self._streams:
             output._add_stream(encoder_stream, codec, **kwargs)
 
         # Now it's OK for the background thread to output frames.
         with self._lock:
-            self._output_available = True
+            self._output = output
             self._first_frame = True
 
     def close_output(self):
@@ -69,43 +67,39 @@ def close_output(self):
             raise RuntimeError("No underlying output has been opened")
 
         # After this, we guarantee that the background thread will never use the output.
+        output = self._output
         with self._lock:
-            self._output_available = False
+            self._output = None
 
-        self._output.stop()
-        self._output = None
+        output.stop()
+
+    def _flush(self, timestamp_now, output):
+        # Flush out anything that is time-expired compared to timestamp_now.
+        # If timestamp_now is None, flush everything.
+        while self._circular and (front := self._circular[0]):
+            _, keyframe, timestamp, _, audio = front
 
-    def _get_frame(self):
-        # Fetch the next frame to be saved to the underlying output.
-        if not self._circular:
-            return
-        if not self._first_frame:
-            return self._circular.popleft()
-        # Must skip ahead to the first I frame if we haven't seen one yet.
-        while self._circular:
-            entry = self._circular.popleft()
-            _, key_frame, _, _, audio = entry
-            # If there is audio, all audio frames are likely to be keyframes, so we must ignore them when
-            # deciding when the streams can resume - only the video counts.
-            if key_frame and not audio:
+            if timestamp_now and timestamp_now - timestamp < self.buffer_duration_ms * 1000:
+                break
+
+            # We need to drop this entry, writing it out if we can.
+            self._circular.popleft()
+
+            if keyframe and not audio:
                 self._first_frame = False
-                return entry
+
+            if not self._first_frame and output:
+                output.outputframe(*front)
 
     def outputframe(self, frame, keyframe=True, timestamp=None, packet=None, audio=False):
         """Write frame to circular buffer"""
         with self._lock:
             if self._buffer_duration_ms == 0 or not self.recording:
                 return
-            self._circular.append((frame, keyframe, timestamp, packet, audio))
-            # Discard any expired buffer entries.
-            while timestamp - self._circular[0][2] > self._buffer_duration_ms * 1000:
-                self._circular.popleft()
 
-            if self._output_available:
-                # Actually write this to the underlying output.
-                entry = self._get_frame()
-                if entry:
-                    self._output.outputframe(*entry)
+            # Add this new frame to the buffer and flush anything that is now expired.
+            self._circular.append((frame, keyframe, timestamp, packet, audio))
+            self._flush(timestamp, self._output)
 
     def start(self):
         """Start recording in the circular buffer."""
@@ -116,20 +110,19 @@ def start(self):
 
     def stop(self):
         """Close file handle and stop recording"""
+        output = self._output
         with self._lock:
             if not self.recording:
                 raise RuntimeError("Circular output was not started")
             self._recording = False
-            self._output_available = False
-
-        # Flush out anything remaining in the buffer if the underlying output is still going
-        # when we stop.
-        if self._output:
-            while (entry := self._get_frame()):
-                self._output.outputframe(*entry)
-            self._output.stop()
             self._output = None
 
+        # At this point the background thread can't be using the circular buffer or the output,
+        # so we can flush everything out.
+        if output:
+            self._flush(None, output)
+            output.stop()
+
     def _add_stream(self, encoder_stream, codec_name, **kwargs):
         # Notice the PyavOutput of a stream that will be sending it packets to write out. It will need
         # to forward these whenever a new underlying output is opened.