Views
No views yet
EnvTransition.EnvTransition tuple and returns a potentially modified tuple of the same structure.EnvTransition is a 7-tuple containing:save_pretrained() and from_pretrained()pip install lerobot1from lerobot.processor.pipeline import RobotProcessor
2from your_steps import ObservationNormalizer, VelocityCalculator
3
4# Create a processor with multiple steps
5processor = RobotProcessor(
6 steps=[
7 ObservationNormalizer(mean=0, std=1),
8 VelocityCalculator(window_size=5),
9 ],
10 name="my_robot_processor",
11 seed=42
12)
13
14# Process a transition
15obs, info = env.reset()
16transition = (obs, None, 0.0, False, False, info, {})
17processed_transition = processor(transition)
18
19# Extract processed observation
20processed_obs = processed_transition[0]1# Save locally
2processor.save_pretrained("./my_processor")
3
4# Push to Hugging Face Hub
5processor.push_to_hub("username/my-robot-processor")
6
7# Load from Hub
8loaded_processor = RobotProcessor.from_pretrained("username/my-robot-processor")1# Inspect intermediate results
2for idx, intermediate_transition in enumerate(processor.step_through(transition)):
3 print(f"After step {idx}: {intermediate_transition[0]}") # Print observation1# Add monitoring hook
2def log_observation(step_idx, transition):
3 print(f"Step {step_idx}: obs shape = {transition[0].shape}")
4 return None # Don't modify transition
5
6processor.register_before_step_hook(log_observation)ProcessorStep protocol:1from lerobot.processor.pipeline import ProcessorStepRegistry, EnvTransition
2
3@ProcessorStepRegistry.register("my_custom_step")
4class MyCustomStep:
5 def __init__(self, param1=1.0):
6 self.param1 = param1
7 self.buffer = []
8
9 def __call__(self, transition: EnvTransition) -> EnvTransition:
10 obs, action, reward, done, truncated, info, comp_data = transition
11 # Process observation
12 processed_obs = obs * self.param1
13 return (processed_obs, action, reward, done, truncated, info, comp_data)
14
15 def get_config(self) -> dict:
16 return {"param1": self.param1}
17
18 def state_dict(self) -> dict:
19 # Return only torch.Tensor state
20 return {}
21
22 def load_state_dict(self, state: dict) -> None:
23 # Load tensor state
24 pass
25
26 def reset(self) -> None:
27 # Clear buffers at episode boundaries
28 self.buffer.clear()1# Move all tensor states to GPU
2processor = processor.to("cuda")
3
4# Move to specific device
5processor = processor.to(torch.device("cuda:1"))1# Profile step execution times
2profile_results = processor.profile_steps(transition, num_runs=100)
3for step_name, time_ms in profile_results.items():
4 print(f"{step_name}: {time_ms:.3f} ms")1# Get a single step
2first_step = processor[0]
3
4# Create a sub-processor with steps 1-3
5sub_processor = processor[1:4]1@misc{cadene2024lerobot,
2 author = {Cadene, Remi and Alibert, Simon and Soare, Alexander and Gallouedec, Quentin and Zouitine, Adil and Palma, Steven and Kooijmans, Pepijn and Aractingi, Michel and Shukor, Mustafa and Aubakirova, Dana and Russi, Martino and Capuano, Francesco and Pascale, Caroline and Choghari, Jade and Moss, Jess and Wolf, Thomas},
3 title = {LeRobot: State-of-the-art Machine Learning for Real-World Robotics in Pytorch},
4 howpublished = "\url{https://github.com/huggingface/lerobot}",
5 year = {2024}
6}