chore(processor): Improve naming conventions

docs/source/hilserl.mdx

@@ -143,27 +143,27 @@ HIL-SERL uses a modular processor pipeline architecture that processes robot obs
 
 The environment processor (`env_processor`) handles incoming observations and environment state:
 
-1. **VanillaObservationProcessor**: Converts raw robot observations into standardized format
-2. **JointVelocityProcessor** (optional): Adds joint velocity information to observations
-3. **MotorCurrentProcessor** (optional): Adds motor current readings to observations
+1. **VanillaObservationProcessorStep**: Converts raw robot observations into standardized format
+2. **JointVelocityProcessorStep** (optional): Adds joint velocity information to observations
+3. **MotorCurrentProcessorStep** (optional): Adds motor current readings to observations
 4. **ForwardKinematicsJointsToEE** (optional): Computes end-effector pose from joint positions
-5. **ImageCropResizeProcessor** (optional): Crops and resizes camera images
-6. **TimeLimitProcessor** (optional): Enforces episode time limits
-7. **GripperPenaltyProcessor** (optional): Applies penalties for inappropriate gripper usage
-8. **RewardClassifierProcessor** (optional): Automated reward detection using vision models
-9. **ToBatchProcessor**: Converts data to batch format for neural network processing
+5. **ImageCropResizeProcessorStep** (optional): Crops and resizes camera images
+6. **TimeLimitProcessorStep** (optional): Enforces episode time limits
+7. **GripperPenaltyProcessorStep** (optional): Applies penalties for inappropriate gripper usage
+8. **RewardClassifierProcessorStep** (optional): Automated reward detection using vision models
+9. **AddBatchDimensionProcessorStep**: Converts data to batch format for neural network processing
 10. **DeviceProcessor**: Moves data to the specified compute device (CPU/GPU)
 
 #### Action Processor Pipeline
 
 The action processor (`action_processor`) handles outgoing actions and human interventions:
 
-1. **AddTeleopActionAsComplimentaryData**: Captures teleoperator actions for logging
-2. **AddTeleopEventsAsInfo**: Records intervention events and episode control signals
+1. **AddTeleopActionAsComplimentaryDataStep**: Captures teleoperator actions for logging
+2. **AddTeleopEventsAsInfoStep**: Records intervention events and episode control signals
 3. **AddRobotObservationAsComplimentaryData**: Stores raw robot state for processing
-4. **InterventionActionProcessor**: Handles human interventions and episode termination
+4. **InterventionActionProcessorStep**: Handles human interventions and episode termination
 5. **Inverse Kinematics Pipeline** (when enabled):
-   - **MapDeltaActionToRobotAction**: Converts delta actions to robot action format
+   - **MapDeltaActionToRobotActionStep**: Converts delta actions to robot action format
    - **EEReferenceAndDelta**: Computes end-effector reference and delta movements
    - **EEBoundsAndSafety**: Enforces workspace safety bounds
    - **InverseKinematicsEEToJoints**: Converts end-effector actions to joint targets

examples/phone_to_so100/evaluate.py

@@ -17,15 +17,15 @@
 from lerobot.cameras.opencv.configuration_opencv import OpenCVCameraConfig
 from lerobot.datasets.lerobot_dataset import LeRobotDataset
 from lerobot.datasets.pipeline_features import aggregate_pipeline_dataset_features
-from lerobot.datasets.utils import merge_features
+from lerobot.datasets.utils import merge_dataset_features
 from lerobot.model.kinematics import RobotKinematics
 from lerobot.policies.act.modeling_act import ACTPolicy
 from lerobot.policies.factory import make_pre_post_processors
 from lerobot.processor.converters import (
+    robot_observation_to_transition,
     to_output_robot_action,
-    to_transition_robot_observation,
 )
-from lerobot.processor.pipeline import RobotProcessor
+from lerobot.processor.pipeline import DataProcessorPipeline
 from lerobot.record import record_loop
 from lerobot.robots.so100_follower.config_so100_follower import SO100FollowerConfig
 from lerobot.robots.so100_follower.robot_kinematic_processor import (
@@ -65,7 +65,7 @@
 )
 
 # Build pipeline to convert ee pose action to joint action
-robot_ee_to_joints = RobotProcessor(
+robot_ee_to_joints = DataProcessorPipeline(
     steps=[
         AddRobotObservationAsComplimentaryData(robot=robot),
         InverseKinematicsEEToJoints(
@@ -79,11 +79,11 @@
 )
 
 # Build pipeline to convert joint observation to ee pose observation
-robot_joints_to_ee_pose = RobotProcessor(
+robot_joints_to_ee_pose = DataProcessorPipeline(
     steps=[
         ForwardKinematicsJointsToEE(kinematics=kinematics_solver, motor_names=list(robot.bus.motors.keys()))
     ],
-    to_transition=to_transition_robot_observation,
+    to_transition=robot_observation_to_transition,
     to_output=lambda tr: tr,
 )
 
@@ -103,7 +103,7 @@
     patterns=["observation.state.ee"],
 )  # Get all ee observation features
 
-dataset_features = merge_features(obs_ee, action_ee_and_gripper)
+dataset_features = merge_dataset_features(obs_ee, action_ee_and_gripper)
 
 print("All dataset features: ", dataset_features)
 

examples/phone_to_so100/record.py

@@ -18,14 +18,14 @@
 from lerobot.cameras.opencv.configuration_opencv import OpenCVCameraConfig
 from lerobot.datasets.lerobot_dataset import LeRobotDataset
 from lerobot.datasets.pipeline_features import aggregate_pipeline_dataset_features
-from lerobot.datasets.utils import merge_features
+from lerobot.datasets.utils import merge_dataset_features
 from lerobot.model.kinematics import RobotKinematics
 from lerobot.processor.converters import (
+    robot_observation_to_transition,
     to_output_robot_action,
-    to_transition_robot_observation,
     to_transition_teleop_action,
 )
-from lerobot.processor.pipeline import RobotProcessor
+from lerobot.processor.pipeline import DataProcessorPipeline
 from lerobot.record import record_loop
 from lerobot.robots.so100_follower.config_so100_follower import SO100FollowerConfig
 from lerobot.robots.so100_follower.robot_kinematic_processor import (
@@ -73,7 +73,7 @@
 )
 
 # Build pipeline to convert phone action to ee pose action
-phone_to_robot_ee_pose = RobotProcessor(
+phone_to_robot_ee_pose = DataProcessorPipeline(
     steps=[
         MapPhoneActionToRobotAction(platform=teleop_config.phone_os),
         AddRobotObservationAsComplimentaryData(robot=robot),
@@ -93,7 +93,7 @@
 )
 
 # Build pipeline to convert ee pose action to joint action
-robot_ee_to_joints = RobotProcessor(
+robot_ee_to_joints = DataProcessorPipeline(
     steps=[
         InverseKinematicsEEToJoints(
             kinematics=kinematics_solver,
@@ -110,11 +110,11 @@
 )
 
 # Build pipeline to convert joint observation to ee pose observation
-robot_joints_to_ee_pose = RobotProcessor(
+robot_joints_to_ee_pose = DataProcessorPipeline(
     steps=[
         ForwardKinematicsJointsToEE(kinematics=kinematics_solver, motor_names=list(robot.bus.motors.keys()))
     ],
-    to_transition=to_transition_robot_observation,
+    to_transition=robot_observation_to_transition,
     to_output=lambda tr: tr,
 )
 
@@ -142,7 +142,7 @@
     patterns=["observation.state.ee"],
 )
 
-dataset_features = merge_features(action_ee, gripper, observation_ee)
+dataset_features = merge_dataset_features(action_ee, gripper, observation_ee)
 
 print("All dataset features: ", dataset_features)
 

examples/phone_to_so100/replay.py

@@ -20,7 +20,7 @@
 from lerobot.datasets.lerobot_dataset import LeRobotDataset
 from lerobot.model.kinematics import RobotKinematics
 from lerobot.processor.converters import to_output_robot_action, to_transition_teleop_action
-from lerobot.processor.pipeline import RobotProcessor
+from lerobot.processor.pipeline import DataProcessorPipeline
 from lerobot.robots.so100_follower.config_so100_follower import SO100FollowerConfig
 from lerobot.robots.so100_follower.robot_kinematic_processor import (
     AddRobotObservationAsComplimentaryData,
@@ -50,7 +50,7 @@
 )
 
 # Build pipeline to convert ee pose action to joint action
-robot_ee_to_joints = RobotProcessor(
+robot_ee_to_joints = DataProcessorPipeline(
     steps=[
         AddRobotObservationAsComplimentaryData(robot=robot),
         InverseKinematicsEEToJoints(

examples/phone_to_so100/teleoperate.py

@@ -16,7 +16,7 @@
 import time
 
 from lerobot.model.kinematics import RobotKinematics
-from lerobot.processor import RobotProcessor
+from lerobot.processor import DataProcessorPipeline
 from lerobot.processor.converters import to_output_robot_action, to_transition_teleop_action
 from lerobot.robots.so100_follower.config_so100_follower import SO100FollowerConfig
 from lerobot.robots.so100_follower.robot_kinematic_processor import (
@@ -49,7 +49,7 @@
 )
 
 # Build pipeline to convert phone action to ee pose action to joint action
-phone_to_robot_joints = RobotProcessor(
+phone_to_robot_joints = DataProcessorPipeline(
     steps=[
         MapPhoneActionToRobotAction(platform=teleop_config.phone_os),
         AddRobotObservationAsComplimentaryData(robot=robot),

src/lerobot/datasets/pipeline_features.py

@@ -17,11 +17,11 @@
 
 from lerobot.constants import ACTION, OBS_IMAGES, OBS_STATE
 from lerobot.datasets.utils import hw_to_dataset_features
-from lerobot.processor.pipeline import RobotProcessor
+from lerobot.processor.pipeline import DataProcessorPipeline
 
 
 def aggregate_pipeline_dataset_features(
-    pipeline: RobotProcessor,
+    pipeline: DataProcessorPipeline,
     initial_features: dict[str, Any],
     *,
     use_videos: bool = True,

src/lerobot/datasets/utils.py

@@ -470,7 +470,7 @@ def dataset_to_policy_features(features: dict[str, dict]) -> dict[str, PolicyFea
     return policy_features
 
 
-def merge_features(*dicts: dict) -> dict:
+def merge_dataset_features(*dicts: dict) -> dict:
     """
     Merge LeRobot grouped feature dicts.
 

src/lerobot/policies/act/processor_act.py

@@ -18,34 +18,34 @@
 from lerobot.constants import POSTPROCESSOR_DEFAULT_NAME, PREPROCESSOR_DEFAULT_NAME
 from lerobot.policies.act.configuration_act import ACTConfig
 from lerobot.processor import (
-    DeviceProcessor,
-    NormalizerProcessor,
-    RenameProcessor,
-    RobotProcessor,
-    ToBatchProcessor,
-    UnnormalizerProcessor,
+    AddBatchDimensionProcessorStep,
+    DeviceProcessorStep,
+    NormalizerProcessorStep,
+    PolicyProcessorPipeline,
+    RenameProcessorStep,
+    UnnormalizerProcessorStep,
 )
 
 
 def make_act_pre_post_processors(
     config: ACTConfig, dataset_stats: dict[str, dict[str, torch.Tensor]] | None = None
-) -> tuple[RobotProcessor, RobotProcessor]:
+) -> tuple[PolicyProcessorPipeline, PolicyProcessorPipeline]:
     input_steps = [
-        RenameProcessor(rename_map={}),
-        NormalizerProcessor(
+        RenameProcessorStep(rename_map={}),
+        NormalizerProcessorStep(
             features={**config.input_features, **config.output_features},
             norm_map=config.normalization_mapping,
             stats=dataset_stats,
         ),
-        ToBatchProcessor(),
-        DeviceProcessor(device=config.device),
+        AddBatchDimensionProcessorStep(),
+        DeviceProcessorStep(device=config.device),
     ]
     output_steps = [
-        DeviceProcessor(device="cpu"),
-        UnnormalizerProcessor(
+        DeviceProcessorStep(device="cpu"),
+        UnnormalizerProcessorStep(
             features=config.output_features, norm_map=config.normalization_mapping, stats=dataset_stats
         ),
     ]
-    return RobotProcessor(steps=input_steps, name=PREPROCESSOR_DEFAULT_NAME), RobotProcessor(
-        steps=output_steps, name=POSTPROCESSOR_DEFAULT_NAME
-    )
+    return PolicyProcessorPipeline(
+        steps=input_steps, name=PREPROCESSOR_DEFAULT_NAME
+    ), PolicyProcessorPipeline(steps=output_steps, name=POSTPROCESSOR_DEFAULT_NAME)

src/lerobot/policies/diffusion/processor_diffusion.py

@@ -19,34 +19,34 @@
 from lerobot.constants import POSTPROCESSOR_DEFAULT_NAME, PREPROCESSOR_DEFAULT_NAME
 from lerobot.policies.diffusion.configuration_diffusion import DiffusionConfig
 from lerobot.processor import (
-    DeviceProcessor,
-    NormalizerProcessor,
-    RenameProcessor,
-    RobotProcessor,
-    ToBatchProcessor,
-    UnnormalizerProcessor,
+    AddBatchDimensionProcessorStep,
+    DeviceProcessorStep,
+    NormalizerProcessorStep,
+    PolicyProcessorPipeline,
+    RenameProcessorStep,
+    UnnormalizerProcessorStep,
 )
 
 
 def make_diffusion_pre_post_processors(
     config: DiffusionConfig, dataset_stats: dict[str, dict[str, torch.Tensor]] | None = None
-) -> tuple[RobotProcessor, RobotProcessor]:
+) -> tuple[PolicyProcessorPipeline, PolicyProcessorPipeline]:
     input_steps = [
-        RenameProcessor(rename_map={}),
-        NormalizerProcessor(
+        RenameProcessorStep(rename_map={}),
+        NormalizerProcessorStep(
             features={**config.input_features, **config.output_features},
             norm_map=config.normalization_mapping,
             stats=dataset_stats,
         ),
-        ToBatchProcessor(),
-        DeviceProcessor(device=config.device),
+        AddBatchDimensionProcessorStep(),
+        DeviceProcessorStep(device=config.device),
     ]
     output_steps = [
-        DeviceProcessor(device="cpu"),
-        UnnormalizerProcessor(
+        DeviceProcessorStep(device="cpu"),
+        UnnormalizerProcessorStep(
             features=config.output_features, norm_map=config.normalization_mapping, stats=dataset_stats
         ),
     ]
-    return RobotProcessor(steps=input_steps, name=PREPROCESSOR_DEFAULT_NAME), RobotProcessor(
-        steps=output_steps, name=POSTPROCESSOR_DEFAULT_NAME
-    )
+    return PolicyProcessorPipeline(
+        steps=input_steps, name=PREPROCESSOR_DEFAULT_NAME
+    ), PolicyProcessorPipeline(steps=output_steps, name=POSTPROCESSOR_DEFAULT_NAME)

src/lerobot/policies/factory.py

@@ -38,7 +38,7 @@
 from lerobot.policies.smolvla.configuration_smolvla import SmolVLAConfig
 from lerobot.policies.tdmpc.configuration_tdmpc import TDMPCConfig
 from lerobot.policies.vqbet.configuration_vqbet import VQBeTConfig
-from lerobot.processor.pipeline import RobotProcessor
+from lerobot.processor.pipeline import PolicyProcessorPipeline
 
 
 def get_policy_class(name: str) -> type[PreTrainedPolicy]:
@@ -120,7 +120,7 @@ def make_pre_post_processors(
     policy_cfg: PreTrainedConfig,
     pretrained_path: str | None = None,
     **kwargs: Unpack[ProcessorConfigKwargs],
-) -> tuple[RobotProcessor, RobotProcessor]:
+) -> tuple[PolicyProcessorPipeline, PolicyProcessorPipeline]:
     """Make a processor instance for a given policy type.
 
     This function creates the appropriate processor configuration based on the policy type.
@@ -140,12 +140,12 @@ def make_pre_post_processors(
     """
     if pretrained_path:
         return (
-            RobotProcessor.from_pretrained(
+            PolicyProcessorPipeline.from_pretrained(
                 pretrained_model_name_or_path=pretrained_path,
                 config_filename=kwargs.get("preprocessor_config_filename", "robot_preprocessor.json"),
                 overrides=kwargs.get("preprocessor_overrides", {}),
             ),
-            RobotProcessor.from_pretrained(
+            PolicyProcessorPipeline.from_pretrained(
                 pretrained_model_name_or_path=pretrained_path,
                 config_filename=kwargs.get("postprocessor_config_filename", "robot_postprocessor.json"),
                 overrides=kwargs.get("postprocessor_overrides", {}),