tangger/lerobot
1
0
Fork 1
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Revert "Remove @require_x"

Browse Source 
This reverts commit 8a7b5c45c7.
This commit is contained in:
Remi Cadene
2024-09-26 14:35:26 +02:00
parent 8a7b5c45c7
commit 395720a5de
5 changed files with 150 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
tests/test_cameras.py
View File

@@ -25,11 +25,7 @@ import numpy as np
import pytest
from lerobot.common.robot_devices.utils import RobotDeviceAlreadyConnectedError, RobotDeviceNotConnectedError
from tests.utils import (
TEST_CAMERA_TYPES,
make_camera,
mock_camera_or_skip_test_when_not_available,
)
from tests.utils import TEST_CAMERA_TYPES, make_camera, require_camera
# Maximum absolute difference between two consecutive images recored by a camera.
# This value differs with respect to the camera.
@@ -41,7 +37,8 @@ def compute_max_pixel_difference(first_image, second_image):
@pytest.mark.parametrize("camera_type, mock", TEST_CAMERA_TYPES)
def test_camera(monkeypatch, camera_type, mock):
@require_camera
def test_camera(request, camera_type, mock):
"""Test assumes that `camera.read()` returns the same image when called multiple times in a row.
So the environment should not change (you shouldnt be in front of the camera) and the camera should not be moving.
@@ -51,8 +48,6 @@ def test_camera(monkeypatch, camera_type, mock):
# TODO(rcadene): measure fps in nightly?
# TODO(rcadene): test logs
mock_camera_or_skip_test_when_not_available(monkeypatch, camera_type, mock)
# Test instantiating
camera = make_camera(camera_type)
@@ -146,10 +141,9 @@ def test_camera(monkeypatch, camera_type, mock):
@pytest.mark.parametrize("camera_type, mock", TEST_CAMERA_TYPES)
def test_save_images_from_cameras(tmpdir, monkeypatch, camera_type, mock):
@require_camera
def test_save_images_from_cameras(tmpdir, request, camera_type, mock):
# TODO(rcadene): refactor
mock_camera_or_skip_test_when_not_available(monkeypatch, camera_type, mock)
if camera_type == "opencv":
from lerobot.common.robot_devices.cameras.opencv import save_images_from_cameras
elif camera_type == "intelrealsense":

27
tests/test_control_robot.py
View File

@@ -31,18 +31,12 @@ from lerobot.common.policies.factory import make_policy
from lerobot.common.utils.utils import init_hydra_config
from lerobot.scripts.control_robot import calibrate, get_available_arms, record, replay, teleoperate
from tests.test_robots import make_robot
from tests.utils import (
DEFAULT_CONFIG_PATH,
DEVICE,
TEST_ROBOT_TYPES,
mock_robot_or_skip_test_when_not_available,
)
from tests.utils import DEFAULT_CONFIG_PATH, DEVICE, TEST_ROBOT_TYPES, require_robot
@pytest.mark.parametrize("robot_type, mock", TEST_ROBOT_TYPES)
def test_teleoperate(monkeypatch, robot_type, mock):
mock_robot_or_skip_test_when_not_available(monkeypatch, robot_type, mock)
@require_robot
def test_teleoperate(request, robot_type, mock):
robot = make_robot(robot_type)
teleoperate(robot, teleop_time_s=1)
teleoperate(robot, fps=30, teleop_time_s=1)
@@ -51,18 +45,16 @@ def test_teleoperate(monkeypatch, robot_type, mock):
@pytest.mark.parametrize("robot_type, mock", TEST_ROBOT_TYPES)
def test_calibrate(monkeypatch, robot_type, mock):
mock_robot_or_skip_test_when_not_available(monkeypatch, robot_type, mock)
@require_robot
def test_calibrate(request, robot_type, mock):
robot = make_robot(robot_type)
calibrate(robot, arms=get_available_arms(robot))
del robot
@pytest.mark.parametrize("robot_type, mock", TEST_ROBOT_TYPES)
def test_record_without_cameras(tmpdir, monkeypatch, robot_type, mock):
mock_robot_or_skip_test_when_not_available(monkeypatch, robot_type, mock)
@require_robot
def test_record_without_cameras(tmpdir, request, robot_type, mock):
root = Path(tmpdir)
repo_id = "lerobot/debug"
@@ -82,9 +74,8 @@ def test_record_without_cameras(tmpdir, monkeypatch, robot_type, mock):
@pytest.mark.parametrize("robot_type, mock", TEST_ROBOT_TYPES)
def test_record_and_replay_and_policy(tmpdir, monkeypatch, robot_type, mock):
mock_robot_or_skip_test_when_not_available(monkeypatch, robot_type, mock)
@require_robot
def test_record_and_replay_and_policy(tmpdir, request, robot_type, mock):
env_name = "koch_real"
policy_name = "act_koch_real"

24
tests/test_motors.py
View File

@@ -31,22 +31,17 @@ import numpy as np
import pytest
from lerobot.common.robot_devices.utils import RobotDeviceAlreadyConnectedError, RobotDeviceNotConnectedError
from tests.utils import (
TEST_MOTOR_TYPES,
make_motors_bus,
mock_input,
mock_motor_or_skip_test_when_not_available,
)
from tests.utils import TEST_MOTOR_TYPES, make_motors_bus, mock_input, require_motor
@pytest.mark.parametrize("motor_type, mock", TEST_MOTOR_TYPES)
def test_find_port(monkeypatch, motor_type, mock):
mock_motor_or_skip_test_when_not_available(monkeypatch, motor_type, mock)
@require_motor
def test_find_port(request, motor_type, mock):
from lerobot.common.robot_devices.motors.dynamixel import find_port
if mock:
# To run find_port without user input
monkeypatch = request.getfixturevalue("monkeypatch")
monkeypatch.setattr("builtins.input", mock_input)
with pytest.raises(OSError):
@@ -56,11 +51,11 @@ def test_find_port(monkeypatch, motor_type, mock):
@pytest.mark.parametrize("motor_type, mock", TEST_MOTOR_TYPES)
def test_configure_motors_all_ids_1(monkeypatch, motor_type, mock):
mock_motor_or_skip_test_when_not_available(monkeypatch, motor_type, mock)
@require_motor
def test_configure_motors_all_ids_1(request, motor_type, mock):
if mock:
# To run find_port without user input
monkeypatch = request.getfixturevalue("monkeypatch")
monkeypatch.setattr("builtins.input", mock_input)
input("Are you sure you want to re-configure the motors? Press enter to continue...")
@@ -80,9 +75,8 @@ def test_configure_motors_all_ids_1(monkeypatch, motor_type, mock):
@pytest.mark.parametrize("motor_type, mock", TEST_MOTOR_TYPES)
def test_motors_bus(monkeypatch, motor_type, mock):
mock_motor_or_skip_test_when_not_available(monkeypatch, motor_type, mock)
@require_motor
def test_motors_bus(request, motor_type, mock):
motors_bus = make_motors_bus(motor_type)
# Test reading and writting before connecting raises an error

7
tests/test_robots.py
View File

@@ -29,16 +29,15 @@ import pytest
import torch
from lerobot.common.robot_devices.utils import RobotDeviceAlreadyConnectedError, RobotDeviceNotConnectedError
from tests.utils import TEST_ROBOT_TYPES, make_robot, mock_robot_or_skip_test_when_not_available
from tests.utils import TEST_ROBOT_TYPES, make_robot, require_robot
@pytest.mark.parametrize("robot_type, mock", TEST_ROBOT_TYPES)
def test_robot(tmpdir, monkeypatch, robot_type, mock):
@require_robot
def test_robot(tmpdir, request, robot_type, mock):
# TODO(rcadene): measure fps in nightly?
# TODO(rcadene): test logs
# TODO(rcadene): add compatibility with other robots
mock_robot_or_skip_test_when_not_available(monkeypatch, robot_type, mock)
from lerobot.common.robot_devices.robots.manipulator import ManipulatorRobot
if robot_type == "aloha" and mock:

166
tests/utils.py
View File

@@ -185,52 +185,134 @@ def require_package(package_name):
return decorator
def mock_robot_or_skip_test_when_not_available(monkeypatch, robot_type, mock):
if mock:
def require_robot(func):
"""
Decorator that skips the test if a robot is not available
The decorated function must have two arguments `request` and `robot_type`.
Example of usage:
```python
@pytest.mark.parametrize(
"robot_type", ["koch", "aloha"]
)
@require_robot
def test_require_robot(request, robot_type):
pass
```
"""
@wraps(func)
def wrapper(*args, **kwargs):
# Access the pytest request context to get the mockeypatch fixture
request = kwargs.get("request")
robot_type = kwargs.get("robot_type")
mock = kwargs.get("mock")
if robot_type is None:
raise ValueError("The 'robot_type' must be an argument of the test function.")
if request is None:
raise ValueError("The 'request' fixture must be an argument of the test function.")
if mock is None:
raise ValueError("The 'mock' variable must be an argument of the test function.")
if robot_type not in available_robots:
raise ValueError(
f"The camera type '{robot_type}' is not valid. Expected one of these '{available_robots}"
)
# Run test with a monkeypatched version of the robot devices.
# TODO(rcadene): redesign mocking to not have this hardcoded logic
if robot_type in ["koch", "koch_bimanual"]:
camera_type = "opencv"
elif robot_type == "aloha":
camera_type = "intelrealsense"
else:
camera_type = "all"
if mock:
# TODO(rcadene): redesign mocking to not have this hardcoded logic
if robot_type in ["koch", "koch_bimanual"]:
camera_type = "opencv"
elif robot_type == "aloha":
camera_type = "intelrealsense"
else:
camera_type = "all"
mock_cameras(request, camera_type)
mock_motors(request)
mock_cameras(monkeypatch, camera_type)
mock_motors(monkeypatch)
# To run calibration without user input
monkeypatch = request.getfixturevalue("monkeypatch")
monkeypatch.setattr("builtins.input", mock_input)
# To run calibration without user input
monkeypatch.setattr("builtins.input", mock_input)
elif not is_robot_available(robot_type):
# Run test with a real robot. Skip test if robot connection fails.
pytest.skip(f"A {robot_type} robot is not available.")
else:
if not is_robot_available(robot_type):
pytest.skip(f"A {robot_type} robot is not available.")
return func(*args, **kwargs)
return wrapper
def mock_camera_or_skip_test_when_not_available(monkeypatch, camera_type, mock):
if camera_type not in available_cameras:
raise ValueError(
f"The camera type '{camera_type}' is not valid. Expected one of these '{available_cameras}"
)
if mock:
# Run test with a monkeypatched version of the cameras
mock_cameras(monkeypatch, camera_type)
elif not is_camera_available(camera_type):
# Run test with a real camera. Skip test if camera connection fails
pytest.skip(f"A {camera_type} camera is not available.")
def require_camera(func):
@wraps(func)
def wrapper(*args, **kwargs):
# Access the pytest request context to get the mockeypatch fixture
request = kwargs.get("request")
camera_type = kwargs.get("camera_type")
mock = kwargs.get("mock")
if request is None:
raise ValueError("The 'request' fixture must be an argument of the test function.")
if camera_type is None:
raise ValueError("The 'camera_type' must be an argument of the test function.")
if mock is None:
raise ValueError("The 'mock' variable must be an argument of the test function.")
if camera_type not in available_cameras:
raise ValueError(
f"The camera type '{camera_type}' is not valid. Expected one of these '{available_cameras}"
)
# Run test with a monkeypatched version of the robot devices.
if mock:
mock_cameras(request, camera_type)
# Run test with a real robot. Skip test if robot connection fails.
else:
if not is_camera_available(camera_type):
pytest.skip(f"A {camera_type} camera is not available.")
return func(*args, **kwargs)
return wrapper
def mock_motor_or_skip_test_when_not_available(monkeypatch, motor_type, mock):
if motor_type not in available_motors:
raise ValueError(
f"The motor type '{motor_type}' is not valid. Expected one of these '{available_motors}"
)
if mock:
# Run test with a monkeypatched version of the motors
mock_motors(monkeypatch)
elif not is_motor_available(motor_type):
# Run test with a real motor. Skip test if motor connection fails
pytest.skip(f"A {motor_type} motor is not available.")
def require_motor(func):
@wraps(func)
def wrapper(*args, **kwargs):
# Access the pytest request context to get the mockeypatch fixture
request = kwargs.get("request")
motor_type = kwargs.get("motor_type")
mock = kwargs.get("mock")
if request is None:
raise ValueError("The 'request' fixture must be an argument of the test function.")
if motor_type is None:
raise ValueError("The 'motor_type' must be an argument of the test function.")
if mock is None:
raise ValueError("The 'mock' variable must be an argument of the test function.")
if motor_type not in available_motors:
raise ValueError(
f"The motor type '{motor_type}' is not valid. Expected one of these '{available_motors}"
)
# Run test with a monkeypatched version of the robot devices.
if mock:
mock_motors(request)
# Run test with a real robot. Skip test if robot connection fails.
else:
if not is_motor_available(motor_type):
pytest.skip(f"A {motor_type} motor is not available.")
return func(*args, **kwargs)
return wrapper
def mock_input(text=None):
@@ -238,8 +320,9 @@ def mock_input(text=None):
print(text)
def mock_cameras(monkeypatch, camera_type="all"):
def mock_cameras(request, camera_type="all"):
# TODO(rcadene): Redesign the mocking tests
monkeypatch = request.getfixturevalue("monkeypatch")
if camera_type in ["opencv", "all"]:
try:
@@ -276,8 +359,9 @@ def mock_cameras(monkeypatch, camera_type="all"):
)
def mock_motors(monkeypatch):
def mock_motors(request):
# TODO(rcadene): Redesign the mocking tests
monkeypatch = request.getfixturevalue("monkeypatch")
try:
import dynamixel_sdk
@@ -349,8 +433,6 @@ def is_robot_available(robot_type):
config_path = ROBOT_CONFIG_PATH_TEMPLATE.format(robot=robot_type)
robot_cfg = init_hydra_config(config_path)
robot = make_robot(robot_cfg)
print("DEBUG", robot.leader_arms)
print("DEBUG", robot.cameras)
robot.connect()
del robot
return True