Allow the timestamp for a frame to be supplied to the MJPEGStream.

src/labthings_fastapi/outputs/mjpeg_stream.py

@@ -43,7 +43,10 @@
     frame: bytes
     """The frame as a `bytes` object, which is a JPEG image for an MJPEG stream."""
     timestamp: datetime
-    """The time the frame was added to the ringbuffer."""
+    """The time the frame was captured.
+    If the capture time is unknown then the timestamp is
+    the time that the frame was added to the ringbuffer."""
+
     index: int
     """The index of the frame within the stream."""
 
@@ -191,15 +194,15 @@
         :raise ValueError: if the frame is not available.
         """
         if i < 0:
             raise ValueError("i must be >= 0")
         if i < self.last_frame_i - len(self._ringbuffer) + 2:
             raise ValueError("the ith frame has been overwritten")
         if i > self.last_frame_i:
             # TODO: await the ith frame
             raise ValueError("the ith frame has not yet been acquired")
         entry = self._ringbuffer[i % len(self._ringbuffer)]
         if entry.index != i:
             raise ValueError("the ith frame has been overwritten")
         return entry
 
     async def next_frame(self) -> int:
@@ -227,9 +230,9 @@
 
         :return: The next JPEG frame, as a `bytes` object.
         """
         i = await self.next_frame()
         entry = await self.ringbuffer_entry(i)
         return entry.frame
 
     async def next_frame_size(self) -> int:
         """Wait for the next frame and return its size.
@@ -238,9 +241,9 @@
 
         :return: The size of the next JPEG frame, in bytes.
         """
         i = await self.next_frame()
         entry = await self.ringbuffer_entry(i)
         return len(entry.frame)
 
     async def frame_async_generator(self) -> AsyncGenerator[bytes, None]:
         """Yield frames as bytes objects.
@@ -260,12 +263,12 @@
                 yield entry.frame
             except StopAsyncIteration:
                 break
             except Exception as e:  # noqa: BLE001
                 # It's important that errors in the stream don't crash the server.
                 # This may be something we can remove in the future, now streams stop
                 # more elegantly. However, it will require careful testing.f
                 logging.exception(f"Error in stream: {e}, stream stopped")
                 return
 
     async def mjpeg_stream_response(self) -> MJPEGStreamResponse:
         """Return a StreamingResponse that streams an MJPEG stream.
@@ -280,7 +283,7 @@
         """
         return MJPEGStreamResponse(self.frame_async_generator())
 
-    def add_frame(self, frame: bytes) -> None:
+    def add_frame(self, frame: bytes, timestamp: Optional[datetime] = None) -> None:
         """Add a JPEG to the MJPEG stream.
 
         This function adds a frame to the stream. It may be called from
@@ -289,6 +292,9 @@
         are handled.
 
         :param frame: The frame to add
+        :param timestamp: The time the frame was captured. If not supplied,
+            the timestamp falls back to the time the frame  was added to the
+            ringbuffer.
 
         :raise ValueError: if the supplied frame does not start with the JPEG
             start bytes and end with the end bytes.
@@ -299,10 +305,10 @@
             and frame[-2] == 0xFF
             and frame[-1] == 0xD9
         ):
             raise ValueError("Invalid JPEG")
         with self._lock:
             entry = self._ringbuffer[(self.last_frame_i + 1) % len(self._ringbuffer)]
-            entry.timestamp = datetime.now()
+            entry.timestamp = timestamp if timestamp is not None else datetime.now()
             entry.frame = frame
             entry.index = self.last_frame_i + 1
             self._thing_server_interface.start_async_task_soon(
@@ -327,7 +333,7 @@
         :raises RuntimeError: if the stream is still streaming.
         """
         if self._streaming is True:
             raise RuntimeError(
                 "This function should only be called when the stream is stopped."
             )
         async with self.condition: