mirror of
https://github.com/huggingface/lerobot.git
synced 2026-07-25 10:46:01 +00:00
Caroline Pascal
ab2b5b04dd
* (depth image processing): excluding depth frames from the RGB to BGR image processing * test(update): updating tests to include RGB/BGR conversion checks
255 lines
8.8 KiB
Python
255 lines
8.8 KiB
Python
#!/usr/bin/env python
|
|
|
|
# Copyright 2024 The HuggingFace Inc. team. All rights reserved.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# 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.
|
|
|
|
# Example of running a specific test:
|
|
# ```bash
|
|
# pytest tests/cameras/test_opencv.py::test_connect
|
|
# ```
|
|
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from lerobot.cameras.configs import ColorMode, Cv2Rotation
|
|
from lerobot.utils.errors import DeviceAlreadyConnectedError, DeviceNotConnectedError
|
|
|
|
pytest.importorskip("pyrealsense2")
|
|
|
|
from lerobot.cameras.realsense import RealSenseCamera, RealSenseCameraConfig
|
|
|
|
TEST_ARTIFACTS_DIR = Path(__file__).parent.parent / "artifacts" / "cameras"
|
|
BAG_FILE_PATH = TEST_ARTIFACTS_DIR / "test_rs.bag"
|
|
|
|
# NOTE(Steven): For some reason these tests take ~20sec in macOS but only ~2sec in Linux.
|
|
|
|
|
|
def mock_rs_config_enable_device_from_file(rs_config_instance, _sn):
|
|
return rs_config_instance.enable_device_from_file(str(BAG_FILE_PATH), repeat_playback=True)
|
|
|
|
|
|
def mock_rs_config_enable_device_bad_file(rs_config_instance, _sn):
|
|
return rs_config_instance.enable_device_from_file("non_existent_file.bag", repeat_playback=True)
|
|
|
|
|
|
@pytest.fixture(name="patch_realsense", autouse=True)
|
|
def fixture_patch_realsense():
|
|
"""Automatically mock pyrealsense2.config.enable_device for all tests."""
|
|
with patch(
|
|
"pyrealsense2.config.enable_device", side_effect=mock_rs_config_enable_device_from_file
|
|
) as mock:
|
|
yield mock
|
|
|
|
|
|
def test_abc_implementation():
|
|
"""Instantiation should raise an error if the class doesn't implement abstract methods/properties."""
|
|
config = RealSenseCameraConfig(serial_number_or_name="042")
|
|
_ = RealSenseCamera(config)
|
|
|
|
|
|
def test_connect():
|
|
config = RealSenseCameraConfig(serial_number_or_name="042", warmup_s=0)
|
|
|
|
with RealSenseCamera(config) as camera:
|
|
assert camera.is_connected
|
|
|
|
|
|
def test_connect_already_connected():
|
|
config = RealSenseCameraConfig(serial_number_or_name="042", warmup_s=0)
|
|
with RealSenseCamera(config) as camera, pytest.raises(DeviceAlreadyConnectedError):
|
|
camera.connect(warmup=False)
|
|
|
|
|
|
def test_connect_invalid_camera_path(patch_realsense):
|
|
patch_realsense.side_effect = mock_rs_config_enable_device_bad_file
|
|
config = RealSenseCameraConfig(serial_number_or_name="042")
|
|
camera = RealSenseCamera(config)
|
|
|
|
with pytest.raises(ConnectionError):
|
|
camera.connect(warmup=False)
|
|
|
|
|
|
def test_invalid_width_connect():
|
|
config = RealSenseCameraConfig(serial_number_or_name="042", width=99999, height=480, fps=30)
|
|
camera = RealSenseCamera(config)
|
|
|
|
with pytest.raises(ConnectionError):
|
|
camera.connect(warmup=False)
|
|
|
|
|
|
def test_read():
|
|
config = RealSenseCameraConfig(serial_number_or_name="042", width=640, height=480, fps=30, warmup_s=0)
|
|
with RealSenseCamera(config) as camera:
|
|
img = camera.read()
|
|
assert isinstance(img, np.ndarray)
|
|
|
|
|
|
# TODO(Steven): Fix this test for the latest version of pyrealsense2.
|
|
@pytest.mark.skip("Skipping test: pyrealsense2 version > 2.55.1.6486")
|
|
def test_read_depth():
|
|
config = RealSenseCameraConfig(serial_number_or_name="042", width=640, height=480, fps=30, use_depth=True)
|
|
camera = RealSenseCamera(config)
|
|
camera.connect(warmup=False)
|
|
|
|
img = camera.read_depth(timeout_ms=2000) # NOTE(Steven): Reading depth takes longer in CI environments.
|
|
assert isinstance(img, np.ndarray)
|
|
|
|
|
|
# These exercise _postprocess_image directly rather than read(): the bag playback returns
|
|
# non-deterministic frames we can't compare against, and the depth read() path is skipped
|
|
# (see test_read_depth) with the current pyrealsense2 version.
|
|
def test_color_mode_conversion(img_array_factory):
|
|
"""RGB (native for RealSense) is passed through; BGR reverses the channel axis."""
|
|
color = img_array_factory(height=3, width=4)
|
|
|
|
outputs = {}
|
|
for color_mode in (ColorMode.RGB, ColorMode.BGR):
|
|
camera = RealSenseCamera(RealSenseCameraConfig(serial_number_or_name="042", color_mode=color_mode))
|
|
camera.capture_height, camera.capture_width = color.shape[:2]
|
|
outputs[color_mode] = camera._postprocess_image(color)
|
|
|
|
np.testing.assert_array_equal(outputs[ColorMode.RGB], color)
|
|
np.testing.assert_array_equal(outputs[ColorMode.BGR], color[..., ::-1])
|
|
|
|
|
|
def test_depth_frame_not_color_converted(img_array_factory):
|
|
"""Depth frames must bypass color conversion, even when a BGR color_mode is set."""
|
|
camera = RealSenseCamera(RealSenseCameraConfig(serial_number_or_name="042", color_mode=ColorMode.BGR))
|
|
depth = img_array_factory(height=3, width=4, channels=1, dtype=np.uint16)[..., 0]
|
|
camera.capture_height, camera.capture_width = depth.shape
|
|
|
|
np.testing.assert_array_equal(camera._postprocess_image(depth, depth_frame=True), depth)
|
|
|
|
|
|
def test_read_before_connect():
|
|
config = RealSenseCameraConfig(serial_number_or_name="042")
|
|
camera = RealSenseCamera(config)
|
|
|
|
with pytest.raises(DeviceNotConnectedError):
|
|
_ = camera.read()
|
|
|
|
|
|
def test_disconnect():
|
|
config = RealSenseCameraConfig(serial_number_or_name="042")
|
|
camera = RealSenseCamera(config)
|
|
camera.connect(warmup=False)
|
|
|
|
camera.disconnect()
|
|
|
|
assert not camera.is_connected
|
|
|
|
|
|
def test_disconnect_before_connect():
|
|
config = RealSenseCameraConfig(serial_number_or_name="042")
|
|
camera = RealSenseCamera(config)
|
|
|
|
with pytest.raises(DeviceNotConnectedError):
|
|
camera.disconnect()
|
|
|
|
|
|
def test_async_read():
|
|
config = RealSenseCameraConfig(serial_number_or_name="042", width=640, height=480, fps=30, warmup_s=0)
|
|
|
|
with RealSenseCamera(config) as camera:
|
|
img = camera.async_read()
|
|
|
|
assert camera.thread is not None
|
|
assert camera.thread.is_alive()
|
|
assert isinstance(img, np.ndarray)
|
|
|
|
|
|
def test_async_read_timeout():
|
|
config = RealSenseCameraConfig(serial_number_or_name="042", width=640, height=480, fps=30, warmup_s=0)
|
|
with RealSenseCamera(config) as camera, pytest.raises(TimeoutError):
|
|
camera.async_read(timeout_ms=0) # consumes any available frame by then
|
|
camera.async_read(timeout_ms=0) # request immediately another one
|
|
|
|
|
|
def test_async_read_before_connect():
|
|
config = RealSenseCameraConfig(serial_number_or_name="042")
|
|
camera = RealSenseCamera(config)
|
|
|
|
with pytest.raises(DeviceNotConnectedError):
|
|
_ = camera.async_read()
|
|
|
|
|
|
def test_read_latest():
|
|
config = RealSenseCameraConfig(serial_number_or_name="042", width=640, height=480, fps=30, warmup_s=0)
|
|
with RealSenseCamera(config) as camera:
|
|
img = camera.read()
|
|
latest = camera.read_latest()
|
|
|
|
assert isinstance(latest, np.ndarray)
|
|
assert latest.shape == img.shape
|
|
|
|
|
|
def test_read_latest_high_frequency():
|
|
config = RealSenseCameraConfig(serial_number_or_name="042", width=640, height=480, fps=30, warmup_s=0)
|
|
with RealSenseCamera(config) as camera:
|
|
# prime with one read to ensure frames are available
|
|
ref = camera.read()
|
|
|
|
for _ in range(20):
|
|
latest = camera.read_latest()
|
|
assert isinstance(latest, np.ndarray)
|
|
assert latest.shape == ref.shape
|
|
|
|
|
|
def test_read_latest_before_connect():
|
|
config = RealSenseCameraConfig(serial_number_or_name="042")
|
|
camera = RealSenseCamera(config)
|
|
|
|
with pytest.raises(DeviceNotConnectedError):
|
|
_ = camera.read_latest()
|
|
|
|
|
|
def test_read_latest_too_old():
|
|
config = RealSenseCameraConfig(serial_number_or_name="042")
|
|
|
|
with RealSenseCamera(config) as camera:
|
|
# prime to ensure frames are available
|
|
_ = camera.read()
|
|
|
|
with pytest.raises(TimeoutError):
|
|
_ = camera.read_latest(max_age_ms=0) # immediately too old
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"rotation",
|
|
[
|
|
Cv2Rotation.NO_ROTATION,
|
|
Cv2Rotation.ROTATE_90,
|
|
Cv2Rotation.ROTATE_180,
|
|
Cv2Rotation.ROTATE_270,
|
|
],
|
|
ids=["no_rot", "rot90", "rot180", "rot270"],
|
|
)
|
|
def test_rotation(rotation):
|
|
config = RealSenseCameraConfig(serial_number_or_name="042", rotation=rotation, warmup_s=0)
|
|
with RealSenseCamera(config) as camera:
|
|
img = camera.read()
|
|
assert isinstance(img, np.ndarray)
|
|
|
|
if rotation in (Cv2Rotation.ROTATE_90, Cv2Rotation.ROTATE_270):
|
|
assert camera.width == 480
|
|
assert camera.height == 640
|
|
assert img.shape[:2] == (640, 480)
|
|
else:
|
|
assert camera.width == 640
|
|
assert camera.height == 480
|
|
assert img.shape[:2] == (480, 640)
|