admin/lerobot
1
0
Fork 0
mirror of https://github.com/huggingface/lerobot.git synced 2026-07-14 13:31:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(so100): add retry parameter to sync_read and sync_write methods

Browse Source
This commit is contained in:
Khalil Meftah
2026-07-12 17:31:09 +02:00
parent 81b6ea1669
commit 348efac2bd
2 changed files with 5 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lerobot/robots/so_follower/so_follower.py
+3 -3
View File
@@ -182,7 +182,7 @@ class SOFollower(Robot):
def get_observation(self) -> RobotObservation:
# Read arm position
start = time.perf_counter()
obs_dict = self.bus.sync_read("Present_Position")
obs_dict = self.bus.sync_read("Present_Position", num_retry=3)
obs_dict = {f"{motor}.pos": val for motor, val in obs_dict.items()}
dt_ms = (time.perf_counter() - start) * 1e3
logger.debug(f"{self} read state: {dt_ms:.1f}ms")
@@ -223,12 +223,12 @@ class SOFollower(Robot):
# Cap goal position when too far away from present position.
# /!\ Slower fps expected due to reading from the follower.
if self.config.max_relative_target is not None:
present_pos = self.bus.sync_read("Present_Position")
present_pos = self.bus.sync_read("Present_Position", num_retry=3)
goal_present_pos = {key: (g_pos, present_pos[key]) for key, g_pos in goal_pos.items()}
goal_pos = ensure_safe_goal_position(goal_present_pos, self.config.max_relative_target)
# Send goal position to the arm
self.bus.sync_write("Goal_Position", goal_pos)
self.bus.sync_write("Goal_Position", goal_pos, num_retry=3)
return {f"{motor}.pos": val for motor, val in goal_pos.items()}
@check_if_not_connected
src/lerobot/teleoperators/so_leader/so_leader.py
+2 -2
View File
@@ -145,7 +145,7 @@ class SOLeader(Teleoperator):
@check_if_not_connected
def get_action(self) -> dict[str, float]:
start = time.perf_counter()
action = self.bus.sync_read("Present_Position")
action = self.bus.sync_read("Present_Position", num_retry=3)
action = {f"{motor}.pos": val for motor, val in action.items()}
dt_ms = (time.perf_counter() - start) * 1e3
logger.debug(f"{self} read action: {dt_ms:.1f}ms")
@@ -155,7 +155,7 @@ class SOLeader(Teleoperator):
def send_feedback(self, feedback: dict[str, float]) -> None:
goals = {k.removesuffix(".pos"): v for k, v in feedback.items() if k.endswith(".pos")}
if goals:
self.bus.sync_write("Goal_Position", goals)
self.bus.sync_write("Goal_Position", goals, num_retry=3)
@check_if_not_connected
def disconnect(self) -> None: