style: formatting and linting

Author: juelg

examples/so101_env_cartesian_control.py

@@ -1,5 +1,4 @@
 import logging
-from time import sleep
 
 from rcs.envs.base import ControlMode, RelativeTo
 from rcs.envs.creators import SimEnvCreator

examples/so101_env_joint_control.py

@@ -1,5 +1,4 @@
 import logging
-from time import sleep
 
 import numpy as np
 from rcs.envs.base import ControlMode, RelativeTo

extensions/rcs_realsense/src/rcs_realsense/calibration.py

@@ -154,4 +154,4 @@ def get_marker_pose(calib_tag_id, detector, intrinsics, frame):
         tag_size=0.1,
     )
 
-    return marker_det, pose
+    return marker_det, pose

extensions/rcs_realsense/src/rcs_realsense/utils.py

@@ -17,6 +17,7 @@ def default_realsense(name2id: dict[str, str] | None) -> RealSenseCameraSet | No
     calibration_strategy = {name: typing.cast(CalibrationStrategy, FR3BaseArucoCalibration(name)) for name in name2id}
     return RealSenseCameraSet(cameras=cameras, calibration_strategy=calibration_strategy)
 
+
 def default_realsense_dummy_calibration(name2id: dict[str, str] | None) -> RealSenseCameraSet | None:
     if name2id is None:
         return None

extensions/rcs_so101/src/rcs_so101/env_cartesian_control.py

@@ -18,6 +18,7 @@
 
 def main():
 
+    robot_cfg: rcs._core.common.RobotConfig
     if ROBOT_INSTANCE == RobotPlatform.HARDWARE:
         robot_cfg = SO101Config()
         robot_cfg.id = "jobi_follower"

extensions/rcs_so101/src/rcs_so101/env_joint_control.py

@@ -19,6 +19,7 @@
 
 def main():
 
+    robot_cfg: rcs._core.common.RobotConfig
     if ROBOT_INSTANCE == RobotPlatform.HARDWARE:
         robot_cfg = SO101Config()
         robot_cfg.id = "jobi_follower"

extensions/rcs_so101/src/rcs_so101/hw.py

@@ -1,11 +1,9 @@
 import threading
-import time
 import typing
 from pathlib import Path
 
 import numpy as np
-from attr import dataclass
-from lerobot.robots import make_robot_from_config  # noqa: F401
+from lerobot.robots import make_robot_from_config
 from lerobot.robots.so101_follower.config_so101_follower import SO101FollowerConfig
 from lerobot.robots.so101_follower.so101_follower import SO101Follower
 from rcs.utils import SimpleFrameRate
@@ -19,14 +17,14 @@ class SO101Config(common.RobotConfig):
     calibration_dir: str = "."
 
 
-class SO101:
+class SO101(common.Robot):
     def __init__(self, robot_cfg: SO101Config, ik: common.IK):
         super().__init__()
         self.ik = ik
         cfg = SO101FollowerConfig(id=robot_cfg.id, calibration_dir=Path(robot_cfg.calibration_dir), port=robot_cfg.port)
         self.hf_robot = make_robot_from_config(cfg)
         self.hf_robot.connect()
-        self._thread = None
+        self._thread: threading.Thread | None = None
         self._running = False
         self._goal = None
         self._goal_lock = threading.Lock()
@@ -87,8 +85,8 @@ def move_home(self) -> None:
         #     np.ndarray[tuple[typing.Literal[5]], np.dtype[np.float64]],
         #     common.robots_meta_config(common.RobotType.SO101).q_home,
         # )
-        home = np.array([-0.01914898, -1.90521916,  1.56476701,  1.04783839, -1.40323926])
-        self.set_joint_position(home)
+        home = np.array([-0.01914898, -1.90521916, 1.56476701, 1.04783839, -1.40323926])
+        self.set_joint_position(home)  # type: ignore
 
     def reset(self) -> None:
         pass
@@ -119,7 +117,6 @@ def _set_joint_position(self, q: np.ndarray[tuple[typing.Literal[5]], np.dtype[n
 
     def set_joint_position(self, q: np.ndarray[tuple[typing.Literal[5]], np.dtype[np.float64]]) -> None:  # type: ignore
         self._set_joint_position(q)
-        
 
     def _controller(self):
         print("Controller thread started")
@@ -187,6 +184,8 @@ def __init__(self, hf_robot: SO101Follower, robot: SO101):
     def get_normalized_width(self) -> float:
         # obs = self.hf_robot.get_observation()
         obs = self.robot.obs
+        if obs is None:
+            return 0.0
         return obs["gripper.pos"] / 100.0
 
     # def get_parameters(self) -> GripperConfig: ...

extensions/rcs_usb_cam/src/rcs_usb_cam/camera.py

@@ -1,38 +1,44 @@
 import copy
+import datetime
 import logging
 import threading
 import typing
-from dataclasses import dataclass
 
+import cv2
 import numpy as np
 from rcs.camera.hw import CalibrationStrategy, DummyCalibrationStrategy, HardwareCamera
-from rcs.camera.interface import BaseCameraSet, DataFrame, CameraFrame, Frame
-import cv2
+from rcs.camera.interface import CameraFrame, DataFrame, Frame
+
 from rcs import common
-import datetime
-'''
+
+"""
 A generic extension class for handling USB-connected cameras. 
 Uses OpenCV to interface with the camera hardware, specifically using cv2.VideoCapture(id).
 The ID can be both a single integer passed as a string, i.e. str(0), str(1), or the full /dev/ path, like /dev/video0.
 
-'''
+"""
+
+
 class USBCameraConfig(common.BaseCameraConfig):
     color_intrinsics: np.ndarray[tuple[typing.Literal[3], typing.Literal[4]], np.dtype[np.float64]] | None = None
     distortion_coeffs: np.ndarray[tuple[typing.Literal[5]], np.dtype[np.float64]] | None = None
 
+
 class USBCameraSet(HardwareCamera):
-    def __init__(self, 
-                 cameras: dict[str, USBCameraConfig], 
-                 calibration_strategy: dict[str, CalibrationStrategy] | None = None,):
+    def __init__(
+        self,
+        cameras: dict[str, USBCameraConfig],
+        calibration_strategy: dict[str, CalibrationStrategy] | None = None,
+    ):
         self.cameras = cameras
         self.CALIBRATION_FRAME_SIZE = 30
         if calibration_strategy is None:
             calibration_strategy = {camera_name: DummyCalibrationStrategy() for camera_name in cameras}
-        for name, cam in self.cameras.items():
+        for cam in self.cameras.values():
             if cam.color_intrinsics is None:
-                cam.color_intrinsics = np.zeros((3, 4), dtype=np.float64)
+                cam.color_intrinsics = np.zeros((3, 4), dtype=np.float64)  # type: ignore
             if cam.distortion_coeffs is None:
-                cam.distortion_coeffs = np.zeros((5,), dtype=np.float64)
+                cam.distortion_coeffs = np.zeros((5,), dtype=np.float64)  # type: ignore
             if cam.resolution_height is None:
                 cam.resolution_height = 480
             if cam.resolution_width is None:
@@ -44,7 +50,9 @@ def __init__(self,
         self._captures: dict[str, cv2.VideoCapture] = {}
         self._logger = logging.getLogger(__name__)
         self._logger.info("USBCamera initialized with cameras: %s", self._camera_names)
-        self._logger.info("If the camera streams are not correct, try v4l2-ctl --list-devices to see the available cameras.")
+        self._logger.info(
+            "If the camera streams are not correct, try v4l2-ctl --list-devices to see the available cameras."
+        )
         self._frame_buffer_lock: dict[str, threading.Lock] = {}
         self._frame_buffer: dict[str, list] = {}
 
@@ -58,7 +66,8 @@ def open(self):
             cap.set(cv2.CAP_PROP_FPS, camera.frame_rate)
 
             if not cap.isOpened():
-                raise RuntimeError(f"Could not open camera {name} with id {camera.identifier}")
+                msg = f"Could not open camera {name} with id {camera.identifier}"
+                raise RuntimeError(msg)
             self._captures[name] = cap
 
     @property
@@ -70,20 +79,27 @@ def poll_frame(self, camera_name: str) -> Frame:
         timestamp = datetime.datetime.now().timestamp()
         ret, color_frame = cap.read()
         if not ret:
-            raise RuntimeError(f"Failed to read frame from camera {camera_name}")
+            msg = f"Failed to read frame from camera {camera_name}"
+            raise RuntimeError(msg)
         with self._frame_buffer_lock[camera_name]:
             if len(self._frame_buffer[camera_name]) >= self.CALIBRATION_FRAME_SIZE:
                 self._frame_buffer[camera_name].pop(0)
             self._frame_buffer[camera_name].append(copy.deepcopy(color_frame))
-        color = DataFrame(data=color_frame, 
-                          timestamp=timestamp, 
-                          intrinsics=self.cameras[camera_name].color_intrinsics, 
-                          extrinsics=self.calibration_strategy[camera_name].get_extrinsics())
-        depth_frame = np.zeros((self.cameras[camera_name].resolution_height, self.cameras[camera_name].resolution_width), dtype=np.uint16)
-        depth = DataFrame(data=depth_frame, 
-                          timestamp=timestamp,
-                          intrinsics=self.cameras[camera_name].color_intrinsics,
-                          extrinsics=self.calibration_strategy[camera_name].get_extrinsics())
+        color = DataFrame(
+            data=color_frame,
+            timestamp=timestamp,
+            intrinsics=self.cameras[camera_name].color_intrinsics,
+            extrinsics=self.calibration_strategy[camera_name].get_extrinsics(),
+        )
+        depth_frame = np.zeros(
+            (self.cameras[camera_name].resolution_height, self.cameras[camera_name].resolution_width), dtype=np.uint16
+        )
+        depth = DataFrame(
+            data=depth_frame,
+            timestamp=timestamp,
+            intrinsics=self.cameras[camera_name].color_intrinsics,
+            extrinsics=self.calibration_strategy[camera_name].get_extrinsics(),
+        )
         cf = CameraFrame(color=color, depth=depth)
         return Frame(camera=cf, avg_timestamp=timestamp)
 

extensions/rcs_usb_cam/src/rcs_usb_cam/webcam.py

@@ -1,7 +1,10 @@
 import cv2
-'''
+
+"""
 Simple script for opening a webcam using OpenCV and displaying the video feed.
-'''
+"""
+
+
 def open_webcam():
     # Open the default webcam (0).
     # If you have multiple cameras, you can try 1, 2, etc.
@@ -26,12 +29,13 @@ def open_webcam():
         cv2.imshow("Webcam", frame)
 
         # Press 'q' to quit
-        if cv2.waitKey(1) & 0xFF == ord('q'):
+        if cv2.waitKey(1) & 0xFF == ord("q"):
             break
 
     # Release the camera and close windows
     cap.release()
     cv2.destroyAllWindows()
 
+
 if __name__ == "__main__":
     open_webcam()