Feat/camera async read require new

src/lerobot/cameras/opencv/camera_opencv.py

@@ -454,12 +454,14 @@ def _read_loop(self) -> None:
 
     def _start_read_thread(self) -> None:
         """Starts or restarts the background read thread if it's not running."""
-        if self.thread is not None and self.thread.is_alive():
-            self.thread.join(timeout=0.1)
         if self.stop_event is not None:
             self.stop_event.set()
+        if self.thread is not None and self.thread.is_alive():
+            self.thread.join(timeout=0.1)
 
         self.stop_event = Event()
+        self.new_frame_event = Event()
+
         self.thread = Thread(target=self._read_loop, args=(), name=f"{self}_read_loop")
         self.thread.daemon = True
         self.thread.start()
@@ -475,7 +477,7 @@ def _stop_read_thread(self) -> None:
         self.thread = None
         self.stop_event = None
 
-    def async_read(self, timeout_ms: float = 200) -> NDArray[Any]:
+    def async_read(self, timeout_ms: float = 200, require_new: bool = True) -> NDArray[Any]:
         """
         Reads the latest available frame asynchronously.
 
@@ -486,6 +488,9 @@ def async_read(self, timeout_ms: float = 200) -> NDArray[Any]:
         Args:
             timeout_ms (float): Maximum time in milliseconds to wait for a frame
                 to become available. Defaults to 200ms (0.2 seconds).
+            require_new (bool): If True, only return when a new frame has been produced
+                (guarantees freshness); otherwise, return the most recent frame immediately,
+                even if it is the same frame as last time (no freshness guarantee).
 
         Returns:
             np.ndarray: The latest captured frame as a NumPy array in the format
@@ -502,6 +507,11 @@ def async_read(self, timeout_ms: float = 200) -> NDArray[Any]:
         if self.thread is None or not self.thread.is_alive():
             self._start_read_thread()
 
+        with self.frame_lock:
+            frame = self.latest_frame
+        if not require_new and frame is not None:
+            return frame
+
         if not self.new_frame_event.wait(timeout=timeout_ms / 1000.0):
             thread_alive = self.thread is not None and self.thread.is_alive()
             raise TimeoutError(

src/lerobot/cameras/realsense/camera_realsense.py

@@ -477,12 +477,14 @@ def _read_loop(self) -> None:
 
     def _start_read_thread(self) -> None:
         """Starts or restarts the background read thread if it's not running."""
-        if self.thread is not None and self.thread.is_alive():
-            self.thread.join(timeout=0.1)
         if self.stop_event is not None:
             self.stop_event.set()
+        if self.thread is not None and self.thread.is_alive():
+            self.thread.join(timeout=0.1)
 
         self.stop_event = Event()
+        self.new_frame_event = Event()
+
         self.thread = Thread(target=self._read_loop, args=(), name=f"{self}_read_loop")
         self.thread.daemon = True
         self.thread.start()
@@ -499,7 +501,7 @@ def _stop_read_thread(self) -> None:
         self.stop_event = None
 
     # NOTE(Steven): Missing implementation for depth for now
-    def async_read(self, timeout_ms: float = 200) -> NDArray[Any]:
+    def async_read(self, timeout_ms: float = 200, require_new: bool = True) -> NDArray[Any]:
         """
         Reads the latest available frame data (color) asynchronously.
 
@@ -510,6 +512,9 @@ def async_read(self, timeout_ms: float = 200) -> NDArray[Any]:
         Args:
             timeout_ms (float): Maximum time in milliseconds to wait for a frame
                 to become available. Defaults to 200ms (0.2 seconds).
+            require_new (bool): If True, only return when a new frame has been produced
+                (guarantees freshness); otherwise, return the most recent frame immediately,
+                even if it is the same frame as last time (no freshness guarantee).
 
         Returns:
             np.ndarray:
@@ -526,6 +531,11 @@ def async_read(self, timeout_ms: float = 200) -> NDArray[Any]:
         if self.thread is None or not self.thread.is_alive():
             self._start_read_thread()
 
+        with self.frame_lock:
+            frame = self.latest_frame
+        if not require_new and frame is not None:
+            return frame
+
         if not self.new_frame_event.wait(timeout=timeout_ms / 1000.0):
             thread_alive = self.thread is not None and self.thread.is_alive()
             raise TimeoutError(