lint: fixing linting errors

Author: juelg

examples/teleop/franka.py

@@ -1,26 +1,17 @@
 import logging
-import threading
-from time import sleep
+from typing import Any
 
 import numpy as np
 from rcs._core import common
-from rcs._core.common import RPY, Pose, RobotPlatform
+from rcs._core.common import RobotPlatform
 from rcs._core.sim import SimConfig
 from rcs.camera.hw import HardwareCameraSet
-from rcs.envs.base import (
-    ControlMode,
-    GripperDictType,
-    LimitedTQuatRelDictType,
-    RelativeActionSpace,
-    RelativeTo,
-)
+from rcs.envs.base import ControlMode
 from rcs.envs.creators import SimMultiEnvCreator
-from rcs.envs.storage_wrapper import StorageWrapper
 from rcs.envs.utils import default_digit, default_sim_gripper_cfg, default_sim_robot_cfg
-from rcs.operator.gello import GelloArmConfig, GelloConfig, GelloOperator
+from rcs.operator.gello import GelloConfig, GelloOperator
 from rcs.operator.interface import TeleopLoop
 from rcs.operator.quest import QuestConfig, QuestOperator
-from rcs.utils import SimpleFrameRate
 from rcs_fr3.creators import RCSFR3MultiEnvCreator
 from rcs_fr3.utils import default_fr3_hw_gripper_cfg, default_fr3_hw_robot_cfg
 from rcs_realsense.utils import default_realsense
@@ -66,10 +57,15 @@
 INSTRUCTION = "build a tower with the blocks in front of you"
 
 robot2world = {
-    "right": rcs.common.Pose(translation=[0, 0, 0], rpy_vector=[0.89360858, -0.17453293, 0.46425758]),
-    "left": rcs.common.Pose(translation=[0, 0, 0], rpy_vector=[-0.89360858, -0.17453293, -0.46425758]),
+    "right": rcs.common.Pose(
+        translation=np.array([0, 0, 0]), rpy_vector=np.array([0.89360858, -0.17453293, 0.46425758])
+    ),
+    "left": rcs.common.Pose(
+        translation=np.array([0, 0, 0]), rpy_vector=np.array([-0.89360858, -0.17453293, -0.46425758])
+    ),
 }
 
+config: QuestConfig | GelloConfig
 config = QuestConfig(mq3_addr=MQ3_ADDR, simulation=ROBOT_INSTANCE == RobotPlatform.SIMULATION)
 # config = GelloConfig(
 #     arms={
@@ -83,7 +79,7 @@
 def get_env():
     if ROBOT_INSTANCE == RobotPlatform.HARDWARE:
 
-        cams = []
+        cams: list[Any] = []
         if CAMERA_DICT is not None:
             cams.append(default_realsense(CAMERA_DICT))
         if DIGIT_DICT is not None:

python/rcs/envs/base.py

@@ -12,6 +12,7 @@
 from rcs.envs.space_utils import (
     ActObsInfoWrapper,
     RCSpaceType,
+    Vec1Type,
     Vec6Type,
     Vec7Type,
     Vec18Type,
@@ -99,12 +100,12 @@ class LimitedJointsRelDictType(RCSpaceType):
 
 class GripperDictType(RCSpaceType):
     # 0 for closed, 1 for open (>=0.5 for open)
-    gripper: Annotated[float, gym.spaces.Box(low=0, high=1, dtype=np.float32)]
+    gripper: Annotated[Vec1Type, gym.spaces.Box(low=0, high=1, shape=(1,), dtype=np.float32)]
 
 
 class HandBinDictType(RCSpaceType):
     # 0 for closed, 1 for open (>=0.5 for open)
-    gripper: Annotated[float, gym.spaces.Box(low=0, high=1, dtype=np.float32)]
+    gripper: Annotated[Vec1Type, gym.spaces.Box(low=0, high=1, shape=(1,), dtype=np.float32)]
 
 
 class HandVecDictType(RCSpaceType):

python/rcs/envs/creators.py

@@ -168,9 +168,9 @@ def __call__(  # type: ignore
         for key, mid in name2id.items():
             env: gym.Env = RobotEnv(robots[key], control_mode)
             if gripper_cfg is not None:
-                cfg = copy.copy(gripper_cfg)
-                cfg.add_id(mid)
-                gripper = rcs.sim.SimGripper(simulation, cfg)
+                gripper_cfg_copy = copy.copy(gripper_cfg)
+                gripper_cfg_copy.add_id(mid)
+                gripper = rcs.sim.SimGripper(simulation, gripper_cfg_copy)
                 env = GripperWrapper(env, gripper, binary=True)
 
             if relative_to != RelativeTo.NONE:

python/rcs/envs/space_utils.py

@@ -17,6 +17,7 @@
 
 M = TypeVar("M", bound=int)
 VecType: TypeAlias = np.ndarray[M, np.dtype[np.float64]]
+Vec1Type: TypeAlias = np.ndarray[tuple[Literal[1]], np.dtype[np.float64]]
 Vec7Type: TypeAlias = np.ndarray[tuple[Literal[7]], np.dtype[np.float64]]
 Vec3Type: TypeAlias = np.ndarray[tuple[Literal[3]], np.dtype[np.float64]]
 Vec6Type: TypeAlias = np.ndarray[tuple[Literal[6]], np.dtype[np.float64]]

python/rcs/envs/utils.py

@@ -20,9 +20,9 @@ def default_sim_robot_cfg(scene: str = "fr3_empty_world", idx: str = "0") -> sim
     scene_cfg = rcs.scenes[scene]
     robot_cfg.robot_type = scene_cfg.robot_type
     robot_cfg.tcp_offset = common.Pose(common.FrankaHandTCPOffset())
-    # robot_cfg.add_id(idx)
-    if rcs.scenes[scene].mjb is not None:
-        robot_cfg.mjcf_scene_path = rcs.scenes[scene].mjb
+    robot_cfg.add_id(idx)
+    if (mjb := rcs.scenes[scene].mjb) is not None:
+        robot_cfg.mjcf_scene_path = mjb
     else:
         robot_cfg.mjcf_scene_path = scene_cfg.mjcf_scene
     robot_cfg.kinematic_model_path = scene_cfg.mjcf_robot
@@ -37,7 +37,7 @@ def default_tilburg_hw_hand_cfg(file: str | PathLike | None = None) -> THConfig:
     return hand_cfg
 
 
-def default_sim_gripper_cfg(idx: str = "0") -> sim.SimGripperConfig:
+def default_sim_gripper_cfg(_idx: str = "0") -> sim.SimGripperConfig:
     cfg = sim.SimGripperConfig()
     cfg.collision_geoms = []
     cfg.collision_geoms_fingers = []

python/rcs/operator/gello.py

@@ -1,9 +1,10 @@
+import contextlib
 import copy
 import logging
 import threading
 import time
 from dataclasses import dataclass, field
-from typing import Any, Dict, Iterator, List, Optional, Sequence, Tuple, TypedDict
+from typing import Any, ClassVar, Dict, Iterator, List, Sequence, Tuple
 
 import numpy as np
 
@@ -25,13 +26,7 @@
 except ImportError:
     HAS_PYNPUT = False
 
-from rcs.envs.base import (
-    ArmWithGripper,
-    ControlMode,
-    GripperDictType,
-    JointsDictType,
-    RelativeTo,
-)
+from rcs.envs.base import ControlMode, RelativeTo
 from rcs.operator.interface import BaseOperator, BaseOperatorConfig, TeleopCommands
 from rcs.sim.sim import Sim
 from rcs.utils import SimpleFrameRate
@@ -64,14 +59,15 @@ def __init__(
         pulses_per_revolution: int = 4095,
     ):
         if not HAS_DYNAMIXEL_SDK:
-            raise ImportError("dynamixel_sdk is not installed. Please install it to use GelloOperator.")
+            msg = "dynamixel_sdk is not installed. Please install it to use GelloOperator."
+            raise ImportError(msg)
 
         self._ids = ids
         self._port = port
         self._baudrate = baudrate
         self._pulses_per_revolution = pulses_per_revolution
         self._lock = threading.Lock()
-        self._buffered_joint_positions = None
+        self._buffered_joint_positions: np.ndarray | None = None
 
         self._portHandler = PortHandler(self._port)
         self._packetHandler = PacketHandler(2.0)
@@ -86,29 +82,32 @@ def __init__(
             for dxl_id in self._ids:
                 self._groupSyncReadHandlers[key].addParam(dxl_id)
 
-            if key != "model_number" and key != "present_position":
+            if key not in {"model_number", "present_position"}:
                 self._groupSyncWriteHandlers[key] = GroupSyncWrite(
                     self._portHandler, self._packetHandler, entry["addr"], entry["len"]
                 )
 
         if not self._portHandler.openPort():
-            raise ConnectionError(f"Failed to open port {self._port}")
+            msg = f"Failed to open port {self._port}"
+            raise ConnectionError(msg)
         if not self._portHandler.setBaudRate(self._baudrate):
-            raise ConnectionError(f"Failed to set baudrate {self._baudrate}")
+            msg = f"Failed to set baudrate {self._baudrate}"
+            raise ConnectionError(msg)
 
         self._stop_thread = threading.Event()
-        self._polling_thread = None
+        self._polling_thread: threading.Thread | None = None
         self._is_polling = False
 
     def write_value_by_name(self, name: str, values: Sequence[int | None]):
         if len(values) != len(self._ids):
-            raise ValueError(f"The length of {name} must match the number of servos")
+            msg = f"The length of {name} must match the number of servos"
+            raise ValueError(msg)
 
         handler = self._groupSyncWriteHandlers[name]
         value_len = XL330_CONTROL_TABLE[name]["len"]
 
         with self._lock:
-            for dxl_id, value in zip(self._ids, values):
+            for dxl_id, value in zip(self._ids, values, strict=False):
                 if value is None:
                     continue
                 param = [(value >> (8 * i)) & 0xFF for i in range(value_len)]
@@ -117,7 +116,8 @@ def write_value_by_name(self, name: str, values: Sequence[int | None]):
             comm_result = handler.txPacket()
             if comm_result != COMM_SUCCESS:
                 handler.clearParam()
-                raise RuntimeError(f"Failed to syncwrite {name}: {self._packetHandler.getTxRxResult(comm_result)}")
+                msg = f"Failed to syncwrite {name}: {self._packetHandler.getTxRxResult(comm_result)}"
+                raise RuntimeError(msg)
             handler.clearParam()
 
     def read_value_by_name(self, name: str) -> List[int]:
@@ -128,7 +128,8 @@ def read_value_by_name(self, name: str) -> List[int]:
         with self._lock:
             comm_result = handler.txRxPacket()
             if comm_result != COMM_SUCCESS:
-                raise RuntimeError(f"Failed to sync read {name}: {self._packetHandler.getTxRxResult(comm_result)}")
+                msg = f"Failed to sync read {name}: {self._packetHandler.getTxRxResult(comm_result)}"
+                raise RuntimeError(msg)
 
             values = []
             for dxl_id in self._ids:
@@ -137,7 +138,8 @@ def read_value_by_name(self, name: str) -> List[int]:
                     value = int(np.int32(np.uint32(value)))
                     values.append(value)
                 else:
-                    raise RuntimeError(f"Failed to get {name} for ID {dxl_id}")
+                    msg = f"Failed to get {name} for ID {dxl_id}"
+                    raise RuntimeError(msg)
             return values
 
     def start_joint_polling(self):
@@ -213,7 +215,7 @@ class DynamixelControlConfig:
     goal_current: List[int] = field(default_factory=list)
     operating_mode: List[int] = field(default_factory=list)
 
-    _UPDATE_ORDER = [
+    _UPDATE_ORDER: ClassVar[list[str]] = [
         "operating_mode",
         "goal_current",
         "kp_p",
@@ -329,10 +331,8 @@ def _goal_position_to_pulses(self, goals):
         return [self._driver._rad_to_pulses(rad) for rad in goals_raw]
 
     def close(self):
-        try:
+        with contextlib.suppress(Exception):
             self._driver.write_value_by_name("torque_enable", [0] * self._num_total_joints)
-        except:
-            pass
         self._driver.close()
 
 
@@ -342,7 +342,7 @@ def close(self):
 class GelloOperator(BaseOperator):
     control_mode = (ControlMode.JOINTS, RelativeTo.NONE)
 
-    def __init__(self, config: GelloConfig, sim: Sim | None = None):
+    def __init__(self, config: "GelloConfig", sim: Sim | None = None):
         super().__init__(config, sim)
         self.config: GelloConfig
         self._resource_lock = threading.Lock()
@@ -353,7 +353,7 @@ def __init__(self, config: GelloConfig, sim: Sim | None = None):
 
         self.controller_names = list(self.config.arms.keys())
 
-        self._last_joints = {name: None for name in self.controller_names}
+        self._last_joints: Dict[str, np.ndarray | None] = {name: None for name in self.controller_names}
         self._last_gripper = {name: 1.0 for name in self.controller_names}
         self._hws: Dict[str, GelloHardware] = {}
 
@@ -386,12 +386,13 @@ def reset_operator_state(self):
         pass
 
     def consume_action(self) -> Dict[str, Any]:
-        actions = {}
+        actions: Dict[str, Any] = {}
         with self._resource_lock:
             for name in self.controller_names:
-                if self._last_joints[name] is not None:
+                joints = self._last_joints[name]
+                if joints is not None:
                     actions[name] = {
-                        "joints": self._last_joints[name].copy(),
+                        "joints": joints.copy(),
                         "gripper": np.array([self._last_gripper[name]]),
                     }
         return actions
@@ -434,6 +435,6 @@ def close(self):
 
 @dataclass(kw_only=True)
 class GelloConfig(BaseOperatorConfig):
-    operator_class = GelloOperator
+    operator_class: type[BaseOperator] = field(default=GelloOperator)
     # Dictionary for multi-arm setups: {"left": GelloArmConfig(...), "right": GelloArmConfig(...)}
     arms: Dict[str, GelloArmConfig] = field(default_factory=lambda: {"right": GelloArmConfig()})

python/rcs/operator/interface.py

@@ -27,9 +27,9 @@ class TeleopCommands:
 
 class BaseOperator(ABC, threading.Thread):
     control_mode: tuple[ControlMode, RelativeTo]
-    controller_names: list[str] = field(default=["left", "right"])
+    controller_names: list[str]
 
-    def __init__(self, config: BaseOperatorConfig, sim: Sim | None = None):
+    def __init__(self, config: "BaseOperatorConfig", sim: Sim | None = None):
         threading.Thread.__init__(self)
         self.config = config
         self.sim = sim
@@ -55,7 +55,7 @@ def close(self):
 
 @dataclass(kw_only=True)
 class BaseOperatorConfig:
-    operator_class: BaseOperator
+    operator_class: type[BaseOperator]
     read_frequency: int = 30
     simulation: bool = True
 
@@ -105,13 +105,12 @@ def _translate_keys(self, actions):
         # Fill in missing robots with "hold" actions from last observation
         # This is necessary because absolute environments (like MultiRobotWrapper)
         # require actions for all configured robots in every step.
-        for robot_name in self.env.get_wrapper_attr("envs").keys():
-            if robot_name not in translated:
-                if robot_name in self._last_obs:
-                    translated[robot_name] = {
-                        "joints": self._last_obs[robot_name]["joints"].copy(),
-                        "gripper": self._last_obs[robot_name].get("gripper", 1.0),
-                    }
+        for robot_name in self.env.get_wrapper_attr("envs"):
+            if robot_name not in translated and robot_name in self._last_obs:
+                translated[robot_name] = {
+                    "joints": self._last_obs[robot_name]["joints"].copy(),
+                    "gripper": self._last_obs[robot_name].get("gripper", 1.0),
+                }
         return translated
 
     def environment_step_loop(self):
@@ -141,7 +140,7 @@ def environment_step_loop(self):
                 # consume new commands because of potential origin reset
                 continue
 
-            elif cmds.failure:
+            if cmds.failure:
                 print("Command: Failure! Resetting env...")
                 self._last_obs, _ = self.env.reset()
                 self.operator.reset_operator_state()
@@ -160,7 +159,7 @@ def environment_step_loop(self):
                     print("Waiting for sync... (Press 's' on GELLO/Keyboard to sync)")
 
                 hold_actions = {}
-                for robot_name in self.env.get_wrapper_attr("envs").keys():
+                for robot_name in self.env.get_wrapper_attr("envs"):
                     if robot_name in self._last_obs and "joints" in self._last_obs[robot_name]:
                         hold_actions[robot_name] = {
                             "joints": self._last_obs[robot_name]["joints"].copy(),

python/rcs/operator/pedals.py

@@ -11,7 +11,8 @@ def __init__(self, device_name_substring="Foot Switch"):
         self.device_path = self._find_device(device_name_substring)
 
         if not self.device_path:
-            raise FileNotFoundError(f"Could not find a device matching '{device_name_substring}'")
+            msg = f"Could not find a device matching '{device_name_substring}'"
+            raise FileNotFoundError(msg)
 
         self.device = evdev.InputDevice(self.device_path)
         self.device.grab()  # Prevent events from leaking into the OS/terminal
@@ -45,17 +46,18 @@ def _read_events(self):
                 if event.type == ecodes.EV_KEY:
                     key_event = evdev.categorize(event)
 
-                    with self._lock:
-                        # keystate: 1 is DOWN, 2 is HOLD, 0 is UP
-                        is_pressed = key_event.keystate in [1, 2]
+                    if isinstance(key_event, evdev.KeyEvent):
+                        with self._lock:
+                            # keystate: 1 is DOWN, 2 is HOLD, 0 is UP
+                            is_pressed = key_event.keystate in [1, 2]
 
-                        # Store state using the string name of the key (e.g., 'KEY_A')
-                        # If a key resolves to a list (rare, but happens in evdev), take the first one
-                        key_name = key_event.keycode
-                        if isinstance(key_name, list):
-                            key_name = key_name[0]
+                            # Store state using the string name of the key (e.g., 'KEY_A')
+                            # If a key resolves to a list (rare, but happens in evdev), take the first one
+                            key_name = key_event.keycode
+                            if isinstance(key_name, list):
+                                key_name = key_name[0]
 
-                        self._key_states[key_name] = is_pressed
+                            self._key_states[key_name] = is_pressed
 
         except OSError:
             pass  # Device disconnected or closed