admin/lerobot
1
0
Fork 0
mirror of https://github.com/huggingface/lerobot.git synced 2026-05-18 10:10:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

have only rtc thread read obs and expose it

Browse Source
This commit is contained in:
Pepijn
2026-01-09 15:48:49 +01:00
parent 1f93a74d8c
commit 4166eeb7da
1 changed files with 18 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/rac/rac_data_collection_openarms_rtc.py
+18 -8
View File
@@ -258,6 +258,7 @@ def get_actions_thread(
robot: RobotWrapper, robot: RobotWrapper,
robot_observation_processor, robot_observation_processor,
action_queue_holder: dict, action_queue_holder: dict,
obs_holder: dict,
shutdown_event: Event, shutdown_event: Event,
cfg: RaCRTCConfig, cfg: RaCRTCConfig,
policy_active: Event, policy_active: Event,
@@ -299,6 +300,8 @@ def get_actions_thread(
inference_delay = math.ceil(inference_latency / time_per_chunk) if inference_latency else 0 inference_delay = math.ceil(inference_latency / time_per_chunk) if inference_latency else 0
obs = robot.get_observation() obs = robot.get_observation()
obs_filtered = {k: v for k, v in obs.items() if k in robot.observation_features}
obs_holder["obs"] = obs_filtered # Share with main loop for recording
obs_processed = robot_observation_processor(obs) obs_processed = robot_observation_processor(obs)
obs_with_policy_features = build_dataset_frame( obs_with_policy_features = build_dataset_frame(
@@ -431,10 +434,11 @@ def main(cfg: RaCRTCConfig):
logger.info(f"Policy loaded: {policy.name}") logger.info(f"Policy loaded: {policy.name}")
action_queue_holder = {"queue": ActionQueue(cfg.rtc)} action_queue_holder = {"queue": ActionQueue(cfg.rtc)}
obs_holder = {"obs": None}
get_actions_t = Thread( get_actions_t = Thread(
target=get_actions_thread, target=get_actions_thread,
args=(policy, robot, obs_proc, action_queue_holder, shutdown_event, cfg, policy_active, fps), args=(policy, robot, obs_proc, action_queue_holder, obs_holder, shutdown_event, cfg, policy_active, fps),
daemon=True, daemon=True,
name="GetActions", name="GetActions",
) )
@@ -499,11 +503,12 @@ def main(cfg: RaCRTCConfig):
events["start_correction"] = False events["start_correction"] = False
print("[RaC] Correction mode - you have control") print("[RaC] Correction mode - you have control")
obs = robot.get_observation()
obs_filtered = {k: v for k, v in obs.items() if k in robot.observation_features}
obs_frame = build_dataset_frame(dataset_features, obs_filtered, prefix="observation")
if events["correction_active"]: if events["correction_active"]:
# Human controlling - read obs and record
obs = robot.get_observation()
obs_filtered = {k: v for k, v in obs.items() if k in robot.observation_features}
obs_frame = build_dataset_frame(dataset_features, obs_filtered, prefix="observation")
robot_action = teleop.get_action() robot_action = teleop.get_action()
for key in robot_action: for key in robot_action:
if "gripper" in key: if "gripper" in key:
@@ -515,9 +520,11 @@ def main(cfg: RaCRTCConfig):
frame_buffer.append(frame) frame_buffer.append(frame)
elif events["policy_paused"]: elif events["policy_paused"]:
# Paused - don't touch robot, RTC thread is also paused
pass pass
else: else:
# Policy running - RTC thread handles observations, we execute actions and record
if interp_idx >= len(interpolated_actions): if interp_idx >= len(interpolated_actions):
new_action = action_queue.get() new_action = action_queue.get()
if new_action is not None: if new_action is not None:
@@ -546,9 +553,12 @@ def main(cfg: RaCRTCConfig):
robot.send_action(action_processed) robot.send_action(action_processed)
robot_send_count += 1 robot_send_count += 1
action_frame = build_dataset_frame(dataset_features, action_dict, prefix="action") # Record frame using observation from RTC thread
frame = {**obs_frame, **action_frame, "task": cfg.dataset.single_task} if obs_holder["obs"] is not None:
frame_buffer.append(frame) obs_frame = build_dataset_frame(dataset_features, obs_holder["obs"], prefix="observation")
action_frame = build_dataset_frame(dataset_features, action_dict, prefix="action")
frame = {**obs_frame, **action_frame, "task": cfg.dataset.single_task}
frame_buffer.append(frame)
now = time.perf_counter() now = time.perf_counter()
if now - last_hz_print >= 5.0: if now - last_hz_print >= 5.0: