Pre-commit alignment

Signed-off-by: vmayoral <[email protected]>

examples/u850/1_teleop_ufactory_850.py

@@ -1,12 +1,12 @@
 import time
-import tqdm
-from lerobot.common.robot_devices.motors.ufactory import xArmWrapper
-from lerobot.common.robot_devices.robots.manipulator import ManipulatorRobot
+
 from lerobot.common.robot_devices.cameras.opencv import OpenCVCamera
+from lerobot.common.robot_devices.motors.ufactory import XArmWrapper
+from lerobot.common.robot_devices.robots.manipulator import ManipulatorRobot
 
 # Defines how to communicate with the motors of the leader and follower arms
 leader_arms = {
-    "main": xArmWrapper(
+    "main": XArmWrapper(
         port="192.168.1.236",
         motors={
             # name: (index, model)
@@ -21,7 +21,7 @@
     ),
 }
 follower_arms = {
-    "main": xArmWrapper(
+    "main": XArmWrapper(
         port="192.168.1.218",
         motors={
             # name: (index, model)
@@ -43,7 +43,7 @@
     cameras={
         "top": OpenCVCamera(4, fps=30, width=640, height=480),
         "wrist": OpenCVCamera(10, fps=30, width=640, height=480),
-    },    
+    },
 )
 
 # Connect motors buses and cameras if any (Required)
@@ -90,7 +90,7 @@
 
 
 except KeyboardInterrupt:
-    print('Operation interrupted by user.')
+    print("Operation interrupted by user.")
 
 # seconds = 30
 # frequency = 200

examples/u850/2_teleop_api.py

@@ -8,9 +8,10 @@
 
 import os
 import sys
-import time
 import threading
-sys.path.append(os.path.join(os.path.dirname(__file__), '../../..'))
+import time
+
+sys.path.append(os.path.join(os.path.dirname(__file__), "../../.."))
 from xarm.wrapper import XArmAPI
 
 # Global variables for caching speed and acceleration limits
@@ -20,8 +21,8 @@
 
 def initialize_limits(arm):
     global MAX_SPEED_LIMIT, MAX_ACC_LIMIT
-    MAX_SPEED_LIMIT = max(arm.joint_speed_limit)/3
-    MAX_ACC_LIMIT = max(arm.joint_acc_limit)/3
+    MAX_SPEED_LIMIT = max(arm.joint_speed_limit) / 3
+    MAX_ACC_LIMIT = max(arm.joint_acc_limit) / 3
 
 
 def mimic_arm(arm_source, arm_target, stop_event):
@@ -68,12 +69,11 @@ def monitor_digital_output(arm, stop_event):
 
 
 def monitor_digital_input(arm, stop_event):
-    SINGLE_CLICK_TIME = 0.2
-    DOUBLE_CLICK_TIME = 0.5
-    LONG_CLICK_TIME = 1.0
+    single_click_time = 0.2
+    double_click_time = 0.5
+    long_click_time = 1.0
 
     last_press_time = 0
-    last_release_time = 0
     last_click_time = 0
     long_click_detected = False
     click_count = 0
@@ -83,11 +83,11 @@ def monitor_digital_input(arm, stop_event):
         code, value = arm.get_tgpio_digital(ionum=2)
         if code == 0:
             current_time = time.time()
-            
+
             if value == 1:  # Button pressed
                 if last_press_time == 0:
                     last_press_time = current_time
-                elif not long_click_detected and current_time - last_press_time >= LONG_CLICK_TIME:
+                elif not long_click_detected and current_time - last_press_time >= long_click_time:
                     print("Long click detected -> Switching manual mode")
                     long_click_detected = True
                     long_click_state = not long_click_state
@@ -108,12 +108,12 @@ def monitor_digital_input(arm, stop_event):
                     press_duration = current_time - last_press_time
 
                     if not long_click_detected:
-                        if press_duration < SINGLE_CLICK_TIME:
+                        if press_duration < single_click_time:
                             click_count += 1
                             if click_count == 1:
                                 last_click_time = current_time
                             elif click_count == 2:
-                                if current_time - last_click_time < DOUBLE_CLICK_TIME:
+                                if current_time - last_click_time < double_click_time:
                                     print("Double click detected -> Open gripper")
                                     arm.set_gripper_position(pos=600, wait=False)  # Open gripper
                                     click_count = 0
@@ -127,73 +127,73 @@ def monitor_digital_input(arm, stop_event):
                             arm.set_gripper_position(pos=50, wait=False)  # Close gripper
                             click_count = 0
 
-                    last_release_time = current_time
                     last_press_time = 0
                     long_click_detected = False
 
             # Reset click count if too much time has passed since last click
-            if click_count == 1 and current_time - last_click_time >= DOUBLE_CLICK_TIME:
+            if click_count == 1 and current_time - last_click_time >= double_click_time:
                 print("Single click detected -> Close gripper")
                 arm.set_gripper_position(pos=50, wait=False)  # Close gripper
                 click_count = 0
 
         time.sleep(0.01)  # Check every 10ms for more precise detection
 
+
 # IP addresses of the arms
-ipL = "192.168.1.236"
-ipR = "192.168.1.218"
+ip_left = "192.168.1.236"
+ip_right = "192.168.1.218"
 
 # Initialize both arms
-armL = XArmAPI(ipL)
-armR = XArmAPI(ipR)
+arm_left = XArmAPI(ip_left)
+arm_right = XArmAPI(ip_right)
 
 # Enable both arms, and grippers
 ## L
-armL.motion_enable(enable=True)
-armL.clean_error()
-armL.set_mode(0)
-armL.set_state(state=0)
+arm_left.motion_enable(enable=True)
+arm_left.clean_error()
+arm_left.set_mode(0)
+arm_left.set_state(state=0)
 #
-armL.set_gripper_mode(0)
-armL.set_gripper_enable(True)
-armL.set_gripper_speed(5000)  # default speed, as there's no way to fetch gripper speed from API
+arm_left.set_gripper_mode(0)
+arm_left.set_gripper_enable(True)
+arm_left.set_gripper_speed(5000)  # default speed, as there's no way to fetch gripper speed from API
 
 
 ## R
-armR.motion_enable(enable=True)
-armR.clean_error()
-armR.set_mode(1)
-armR.set_state(state=0)
+arm_right.motion_enable(enable=True)
+arm_right.clean_error()
+arm_right.set_mode(1)
+arm_right.set_state(state=0)
+#
+arm_right.set_gripper_mode(0)
+arm_right.set_gripper_enable(True)
+arm_right.set_gripper_speed(5000)  # default speed, as there's no way to fetch gripper speed from API
+# According to User Manual, should range in 1000-5000
 #
-armR.set_gripper_mode(0)
-armR.set_gripper_enable(True)
-armR.set_gripper_speed(5000)  # default speed, as there's no way to fetch gripper speed from API
-                              # According to User Manual, should range in 1000-5000
-                              #
-                              # see https://www.ufactory.cc/wp-content/uploads/2023/05/xArm-User-Manual-V2.0.0.pdf
+# see https://www.ufactory.cc/wp-content/uploads/2023/05/xArm-User-Manual-V2.0.0.pdf
 
 # Initialize the global speed and acceleration limits
-initialize_limits(armR)
+initialize_limits(arm_right)
 
 # Set left arm to manual mode
-armL.set_mode(2)
-armL.set_state(0)
+arm_left.set_mode(2)
+arm_left.set_state(0)
 # Light up the digital output 2 (button), to signal manual mode
-armL.set_tgpio_digital(ionum=2, value=1)
+arm_left.set_tgpio_digital(ionum=2, value=1)
 
 # Create a stop event for synchronization
 stop_event = threading.Event()
 
 # Create and start the mimic thread
-mimic_thread = threading.Thread(target=mimic_arm, args=(armL, armR, stop_event))
+mimic_thread = threading.Thread(target=mimic_arm, args=(arm_left, arm_right, stop_event))
 mimic_thread.start()
 
 # # Create and start the digital output monitoring thread
-# monitor_output_thread = threading.Thread(target=monitor_digital_output, args=(armL, stop_event))
+# monitor_output_thread = threading.Thread(target=monitor_digital_output, args=(arm_left, stop_event))
 # monitor_output_thread.start()
 
 # Create and start the digital input monitoring thread
-monitor_input_thread = threading.Thread(target=monitor_digital_input, args=(armL, stop_event))
+monitor_input_thread = threading.Thread(target=monitor_digital_input, args=(arm_left, stop_event))
 monitor_input_thread.start()
 
 # Run
@@ -213,11 +213,11 @@ def monitor_digital_input(arm, stop_event):
 print("Mimic operation, digital output monitoring, and digital input monitoring completed.")
 
 # Turn off manual mode after recording
-armL.set_mode(0)
-armL.set_state(0)
+arm_left.set_mode(0)
+arm_left.set_state(0)
 # Light down the digital output 2 (button), to signal manual mode
-armL.set_tgpio_digital(ionum=2, value=0)
+arm_left.set_tgpio_digital(ionum=2, value=0)
 
 # Disconnect both arms
-armL.disconnect()
-armR.disconnect()
+arm_left.disconnect()
+arm_right.disconnect()

lerobot/common/robot_devices/control_utils.py

@@ -98,7 +98,6 @@ def predict_action(observation, policy, device, use_amp):
     ):
         # Convert to pytorch format: channel first and float32 in [0,1] with batch dimension
         for name in observation:
-
             # Check if tensor is double precision and convert if needed
             if observation[name].dtype == torch.float64:  # Explicitly check for double
                 observation[name] = observation[name].double().float()  # Convert double to float32
@@ -309,7 +308,6 @@ def reset_environment(robot, events, reset_time_s, episode_index):
     # Wait if necessary
     with tqdm.tqdm(total=reset_time_s, desc=f"Waiting (next episode: {episode_index + 1})") as pbar:
         while timestamp < reset_time_s:
-
             ## previous implementation
             ##  increments in 1 second and not controllable
             #