feat(processors): use pipelines across the codebase

[AdilZouitine]

🔄 Summary of Changes

This PR introduces a fundamental architectural shift in LeRobot by implementing a modular processor pipeline system that decouples data transformations from model logic. The change enables flexible robot control, better cross-platform compatibility, and cleaner separation of concerns between neural network models and data preprocessing/postprocessing.

🧠 What are Processors?

Processors are universal translators that bridge the gap between raw robot sensor data and machine learning models. In robotics, there's a fundamental mismatch:

• Robots produce: Raw sensor data (camera images, joint positions, motor currents)
• Models expect: Normalized, batched, tokenized data on specific devices
• Robots need: Real-world commands in their native coordinate systems
• Models output: Normalized predictions that need conversion back to robot scales

The processor system solves this through:

• ProcessorStep: Individual transformation units (normalize, tokenize, batch, etc.)
• DataProcessorPipeline: Chains multiple steps together
• EnvTransition: Universal data container flowing through pipelines
• Two Pipeline Types:
  • RobotProcessorPipeline: Handles unbatched robot hardware data
  • PolicyProcessorPipeline: Handles batched model training/inference data

💥 Breaking Changes

Normalization Migration (PR #1452)

• ❌ Before: Normalization embedded in model weights (normalize_inputs.*)
• ✅ After: External PolicyProcessorPipeline components handle normalization
• 🔄 Migration Required: Existing models need conversion using provided migration script

Usage Pattern Change

# Before: Built-in normalization
policy = Policy.from_pretrained("model")
action = policy(batch)

# After: External processors  
policy = make_policy(config, ds_meta=dataset.meta)
preprocessor, postprocessor = make_pre_post_processors(config, dataset.meta.stats)
processed_batch = preprocessor(raw_batch)
action = policy.select_action(processed_batch)
final_action = postprocessor(action)

🚀 Major Features Added

1. End-Effector Control System

• Enable recording and training with end-effector poses instead of joint positions
• Support arbitrary action/observation spaces for flexible robot control
• Kinematic processors for forward/inverse kinematics transformations

2. Phone Teleoperator 📱

• iOS/Android phone-based robot teleoperation
• Map phone pose to robot end-effector targets
• Safety bounds and gripper control integration
• Real-time pose streaming and robot command generation

3. Modular Pipeline Architecture

• 30+ Registered Processors: normalization, tokenization, device placement, batching, etc.
• Type-Safe Pipelines: Generic types with compile-time checking
• Feature Contracts: Steps declare how they transform data shapes/types
• Debug Tools: Hooks, step-through debugging, feature validation

4. Cross-Platform Data Translation

• Converter Functions: Bridge robot hardware ↔ policy model formats
• Universal Data Flow: Sensors → Processors → Models → Processors → Actuators
• Robot Agnostic: Same pipeline works across different robot platforms

🔧 Technical Implementation

Core Components

# ProcessorStep: Individual transformation
class MyProcessor(ProcessorStep):
    def __call__(self, transition: EnvTransition) -> EnvTransition:
        # Transform data
        return modified_transition
    
    def transform_features(self, features):
        # Declare shape/type changes
        return updated_features

# Pipeline: Chain multiple steps
pipeline = PolicyProcessorPipeline[dict, dict](
    steps=[normalize_step, batch_step, device_step],
    to_transition=batch_to_transition,
    to_output=transition_to_batch
)

Data Flow Architecture

Raw Robot Data → RobotProcessorPipeline → PolicyProcessorPipeline → Model
                                                                      ↓
Robot Commands ← RobotProcessorPipeline ← PolicyProcessorPipeline ← Model Output

📚 Documentation Added

1. Introduction to Processors: Core concepts and usage patterns
2. Debug Your Pipeline: Hooks, step-through debugging, validation
3. Processors for Robots and Teleoperators: Real-world integration
4. Implement Your Own Processor: Custom processor development
5. Backward Compatibility: Migration guide for existing models

🔄 Migration Guide

For Existing Models:

python src/lerobot/processor/migrate_policy_normalization.py \
    --pretrained-path your/model/path \
    --push-to-hub \
    --branch migrated

Updated Model Usage:

from lerobot.policies.factory import make_policy, make_pre_post_processors

policy = make_policy(config, ds_meta=dataset.meta)
preprocessor, postprocessor = make_pre_post_processors(
    policy_cfg=config,
    dataset_stats=dataset.meta.stats
)

🎯 Benefits

• 🔧 Modularity: Composable, reusable transformation steps
• 🚀 Performance: GPU-accelerated operations with optimal ordering
• 🛡️ Type Safety: Compile-time checking prevents runtime errors
• 🔄 Flexibility: Easy robot/teleop swapping without code changes
• 📊 Observability: Built-in debugging and monitoring tools
• 🌐 Cross-Platform: Works across different robot platforms and datasets

[HuggingFaceDocBuilderDev]

The docs for this PR live here. All of your documentation changes will be reflected on that endpoint. The docs are available until 30 days after the last update.

[imstevenpmwork]

Hello team,

Great work on this PR—it's looking much better. I'm approving it as I believe it's as clean as a change of this magnitude can be. While this feature touches the entire codebase, the community benefits are clear.

I've left some comments for future reference, but here are the key follow-up items we should prioritize:

• Migrate more code from policies to the new pipelines.
• Decompose ProcessorSteps into more specialized classes to better separate concerns.
• Migrate Lekiwi & Async inference to use the new pipeline features in a future PR.
• Refactor and test Reinforcement Learning components to fully utilize the new pipelines.
• Improve the evaluate script by saving feature keys in the policy config to avoid manual specification.
• Enhance dataset recording by adding more of the EnvTransition (e.g., next.reward).
• Refactor batch_to_transition to reduce its hard-coded assumptions.
• Consider adding an optional deepcopy argument to specialized processors.
• Standardize on using constants from TransitionKey.

The base models will be updated accordingly in HuggingFace Hub - Lerobot organization.
Thanks again for the massive effort on this!