Allow arbitrary camera frame dimensions

lerobot/common/robot_devices/cameras/opencv.py

@@ -4,6 +4,7 @@
 
 import argparse
 import concurrent.futures
+import logging
 import math
 import platform
 import shutil
@@ -276,19 +277,17 @@ def connect(self):
             raise OSError(
                 f"Can't set {self.fps=} for OpenCVCamera({self.camera_index}). Actual value is {actual_fps}."
             )
+        resolution_info_fstring = (
+            "The provided {name} ({value}) could not be set natively on your camera. The settings "
+            f"are now (width={int(actual_width)}, height={int(actual_height)}). This class will resize the "
+            "frames to your desired resolution before returning them."
+        )
         if self.width is not None and self.width != actual_width:
-            raise OSError(
-                f"Can't set {self.width=} for OpenCVCamera({self.camera_index}). Actual value is {actual_width}."
-            )
+            logging.info(resolution_info_fstring.format(name="width", value=self.width))
         if self.height is not None and self.height != actual_height:
-            raise OSError(
-                f"Can't set {self.height=} for OpenCVCamera({self.camera_index}). Actual value is {actual_height}."
-            )
+            logging.info(resolution_info_fstring.format(name="height", value=self.height))
 
         self.fps = actual_fps
-        self.width = actual_width
-        self.height = actual_height
-
         self.is_connected = True
 
     def read(self, temporary_color_mode: str | None = None) -> np.ndarray:
@@ -324,9 +323,7 @@ def read(self, temporary_color_mode: str | None = None) -> np.ndarray:
 
         h, w, _ = color_image.shape
         if h != self.height or w != self.width:
-            raise OSError(
-                f"Can't capture color image with expected height and width ({self.height} x {self.width}). ({h} x {w}) returned instead."
-            )
+            color_image = cv2.resize(color_image, (self.width, self.height))
 
         # log the number of seconds it took to read the image
         self.logs["delta_timestamp_s"] = time.perf_counter() - start_time

tests/conftest.py

@@ -13,8 +13,10 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
+import platform
 import traceback
 
+import cv2
 import pytest
 
 from lerobot.common.utils.utils import init_hydra_config
@@ -41,3 +43,14 @@ def is_robot_available(robot_type):
         traceback.print_exc()
         print(f"\nA {robot_type} robot is not available.")
         return False
+
+
+@pytest.fixture
+def is_camera_available(request: pytest.FixtureRequest) -> bool:
+    camera_index = request.param
+    if platform.system() == "Linux":
+        tmp_camera = cv2.VideoCapture(f"/dev/video{camera_index}")
+    else:
+        tmp_camera = cv2.VideoCapture(camera_index)
+
+    return tmp_camera.isOpened()

tests/test_cameras.py

@@ -7,11 +7,18 @@
 ```
 """
 
+from contextlib import nullcontext
+from unittest.mock import patch
+
 import numpy as np
 import pytest
 
 from lerobot import available_robots
-from lerobot.common.robot_devices.cameras.opencv import OpenCVCamera, save_images_from_cameras
+from lerobot.common.robot_devices.cameras.opencv import (
+    OpenCVCamera,
+    OpenCVCameraConfig,
+    save_images_from_cameras,
+)
 from lerobot.common.robot_devices.utils import RobotDeviceAlreadyConnectedError, RobotDeviceNotConnectedError
 from tests.utils import require_robot
 
@@ -131,6 +138,40 @@ def test_camera(request, robot_type):
     del camera
 
 
+@pytest.mark.parametrize("request_resolution", [(10, 10)])
+# This parameter is only used when the camera is not available. Checks for various configurations of
+# (mis)match between the requested resolution and the target resolution.
+@pytest.mark.parametrize("read_resolution", [(10, 10), (10, 20), (20, 10), (20, 20)])
+# Note: indirect=True passes the CAMERA_INDEX to the `is_camera_available` fixture, then this returns the
+# `is_camera_available` parameter for the test.
+@pytest.mark.parametrize("is_camera_available", [CAMERA_INDEX], indirect=True)
+def test_camera_resize(
+    request_resolution: tuple[int, int], read_resolution: tuple[int, int], is_camera_available: bool
+):
+    """
+    Check that, even if the requested resolution is not supported natively, the `OpenCVCamera.read` method
+    returns an image with the requested resolution.
+
+    When `is_camera_available=True` this test also checks that the `OpenCVCamera.connect` works with natively
+    unsupported resolutions, otherwise if `is_camera_available=False` we have to use a mock patch which
+    unfortunately can't test this.
+    """
+    camera = OpenCVCamera(0, OpenCVCameraConfig(height=request_resolution[0], width=request_resolution[1]))
+    with nullcontext() if is_camera_available else patch("cv2.VideoCapture") as mock_video_capture:
+        if not is_camera_available:
+            # Set an output resolution that is different from the requested one.
+            mock_video_capture.return_value.read.return_value = (
+                True,
+                np.zeros(read_resolution, dtype=np.uint8),
+            )
+            mock_video_capture.return_value.isOpened.return_value = True
+        camera.connect()
+    assert camera.height == request_resolution[0]
+    assert camera.width == request_resolution[1]
+    img = camera.read()
+    assert img.shape[:2] == request_resolution
+
+
 @pytest.mark.parametrize("robot_type", available_robots)
 @require_robot
 def test_save_images_from_cameras(tmpdir, request, robot_type):