lerobot/pr_3613/en/api/processor.md

Processors

Processors are the data transformation layer between a robot, a dataset and a policy. A pipeline is a chain of ProcessorSteps; each step declares how it transforms both the data and the feature contract.

See Introduction to Robot Processors for the concepts, Implement your own processor to write a step, and Debug your processor pipeline when a pipeline misbehaves.

ProcessorStep[[lerobot.processor.ProcessorStep]]

lerobot.processor.ProcessorStep[[lerobot.processor.ProcessorStep]]

lerobot.processor.ProcessorStep()

Source

Abstract base class for a single step in a data processing pipeline.

Each step must implement the __call__ method to perform its transformation on a data transition and the transform_features method to describe how it alters the shape or type of data features.

Subclasses can optionally be stateful by implementing state_dict and load_state_dict.

get_config[[lerobot.processor.ProcessorStep.get_config]]

get_config()

Source

Returns:

A JSON-serializable dictionary of configuration parameters.

Returns the configuration of the step for serialization.

load_state_dict[[lerobot.processor.ProcessorStep.load_state_dict]]

load_state_dict(state: dict[str, torch.Tensor])

Source

Parameters:

state : A dictionary of state tensors.

Loads the step's state from a state dictionary.

reset[[lerobot.processor.ProcessorStep.reset]]

reset()

Source

Resets the internal state of the processor step, if any.

save_artifacts[[lerobot.processor.ProcessorStep.save_artifacts]]

save_artifacts(save_directory: Path)

Source

Save non-tensor assets and map constructor arguments to relative paths.

state_dict[[lerobot.processor.ProcessorStep.state_dict]]

state_dict()

Source

Returns:

A dictionary mapping state names to tensors.

Returns the state of the step (e.g., learned parameters, running means).

transform_features[[lerobot.processor.ProcessorStep.transform_features]]

transform_features(features: dict[PipelineFeatureType, dict[str, PolicyFeature]])

Source

Parameters:

features : A dictionary describing the input features for observations, actions, etc.

Returns:

A dictionary describing the output features after this step's transformation.

Defines how this step modifies the description of pipeline features.

This method is used to track changes in data shapes, dtypes, or modalities as data flows through the pipeline, without needing to process actual data.

DataProcessorPipeline[[lerobot.processor.DataProcessorPipeline]]

lerobot.processor.DataProcessorPipeline[[lerobot.processor.DataProcessorPipeline]]

lerobot.processor.DataProcessorPipeline(steps: Sequence[ProcessorStep] = <factory>, name: str = 'DataProcessorPipeline', to_transition: Callable[[TInput], EnvTransition] = <factory>, to_output: Callable[[EnvTransition], TOutput] = <factory>, before_step_hooks: list[Callable[[int, EnvTransition], None]] = <factory>, after_step_hooks: list[Callable[[int, EnvTransition], None]] = <factory>)

Source

Parameters:

steps : A sequence of ProcessorStep objects that make up the pipeline.

name : A descriptive name for the pipeline.

to_transition : A function to convert raw input data into the standardized EnvTransition format.

to_output : A function to convert the final EnvTransition into the desired output format.

before_step_hooks : A list of functions to be called before each step is executed.

after_step_hooks : A list of functions to be called after each step is executed.

A sequential pipeline for processing data, integrated with the Hugging Face Hub.

This class chains together multiple ProcessorStep instances to form a complete data processing workflow. It's generic, allowing for custom input and output types, which are handled by the to_transition and to_output converters.

from_config[[lerobot.processor.DataProcessorPipeline.from_config]]

from_config(config: dict[str, Any], state_dict: dict[str, dict[str, torch.Tensor]] | None = None, overrides: dict[str, Any] | None = None, to_transition: Callable[[TInput], EnvTransition] | None = None, to_output: Callable[[EnvTransition], TOutput] | None = None)

Source

Parameters:

config : A config dictionary with the same structure as the saved processor JSON.

state_dict : Optional in-memory pipeline state grouped by suffixless state key.

overrides : Optional constructor overrides keyed by registry name or class name.

to_transition : Optional converter from input data to EnvTransition.

to_output : Optional converter from EnvTransition to output data.

Returns:

A processor pipeline built from the config and optional state.

Build a pipeline from an in-memory config and optional state tensors.

from_pretrained[[lerobot.processor.DataProcessorPipeline.from_pretrained]]

from_pretrained(pretrained_model_name_or_path: str | Path, config_filename: str, force_download: bool = False, resume_download: bool | None = None, proxies: dict[str, str] | None = None, token: str | bool | None = None, cache_dir: str | Path | None = None, local_files_only: bool = False, revision: str | None = None, overrides: dict[str, Any] | None = None, to_transition: Callable[[TInput], EnvTransition] | None = None, to_output: Callable[[EnvTransition], TOutput] | None = None, **kwargs)

Source

Parameters:

pretrained_model_name_or_path : The identifier of the repository on the Hugging Face Hub, a path to a local directory, or a path to a single config file.

config_filename : The name of the pipeline's JSON configuration file. Always required to prevent ambiguity when multiple configs exist (e.g., preprocessor vs postprocessor).

force_download : Whether to force (re)downloading the files.

resume_download : Whether to resume a previously interrupted download.

proxies : A dictionary of proxy servers to use.

token : The token to use as HTTP bearer authorization for private Hub repositories.

cache_dir : The path to a specific cache folder to store downloaded files.

local_files_only : If True, avoid downloading files from the Hub.

revision : The specific model version to use (e.g., a branch name, tag name, or commit id).

overrides : A dictionary to override the configuration of specific steps. Keys should match the step's class name or registry name.

to_transition : A custom function to convert input data to EnvTransition.

to_output : A custom function to convert the final EnvTransition to the output format.

• **kwargs : Additional arguments (not used).

Returns:

An instance of DataProcessorPipeline loaded with the specified configuration and state.

Raises: FileNotFoundError or ValueError or ImportError or KeyError or ProcessorMigrationError

• FileNotFoundError -- If the config file cannot be found.
• ValueError -- If configuration is ambiguous or instantiation fails.
• ImportError -- If a step's class cannot be imported.
• KeyError -- If an override key doesn't match any step in the pipeline.
• ProcessorMigrationError -- If the model requires migration to processor format.

Loads a pipeline from a local directory, single file, or Hugging Face Hub repository.

This method implements a simplified loading pipeline with intelligent migration detection:

Simplified Loading Strategy:

1. Config Loading (_load_config):

   • Directory: Load specified config_filename from directory
   • Single file: Load file directly (config_filename ignored)
   • Hub repository: Download specified config_filename from Hub

2. Config Validation (_validate_loaded_config):

   • Format validation: Ensure config is valid processor format
   • Migration detection: Guide users to migrate old LeRobot models
   • Clear errors: Provide actionable error messages

3. Step Construction (_build_steps_with_overrides):

   • Class resolution: Registry lookup or dynamic imports
   • Override merging: User parameters override saved config
   • State loading: Load .safetensors files for stateful steps

4. Override Validation (_validate_overrides_used):

   • Ensure all user overrides were applied (catch typos)
   • Provide helpful error messages with available keys

Migration Detection:

• Smart detection: Analyzes JSON files to detect old LeRobot models
• Precise targeting: Avoids false positives on other HuggingFace models
• Clear guidance: Provides exact migration command to run
• Error mode: Always raises ProcessorMigrationError for clear user action

Loading Examples:

# Directory loading
pipeline = DataProcessorPipeline.from_pretrained("/models/my_model", config_filename="processor.json")

# Single file loading
pipeline = DataProcessorPipeline.from_pretrained(
    "/models/my_model/processor.json", config_filename="processor.json"
)

# Hub loading
pipeline = DataProcessorPipeline.from_pretrained("user/repo", config_filename="processor.json")

# Multiple configs (preprocessor/postprocessor)
preprocessor = DataProcessorPipeline.from_pretrained(
    "model", config_filename="policy_preprocessor.json"
)
postprocessor = DataProcessorPipeline.from_pretrained(
    "model", config_filename="policy_postprocessor.json"
)

Override System:

• Key matching: Use registry names or class names as override keys
• Config merging: User overrides take precedence over saved config
• Validation: Ensure all override keys match actual steps (catch typos)
• Example: overrides={"NormalizeStep": {"device": "cuda"}}

get_config[[lerobot.processor.DataProcessorPipeline.get_config]]

get_config()

Source

Returns:

A dictionary with the same content that save_pretrained() writes as JSON.

Return the JSON-serializable pipeline configuration.

load_state_dict[[lerobot.processor.DataProcessorPipeline.load_state_dict]]

load_state_dict(state_dict: dict[str, dict[str, torch.Tensor]])

Source

Parameters:

state_dict : A dictionary mapping suffixless state keys to step state dictionaries.

Raises: KeyError

• KeyError -- If loading finds missing expected state or unexpected extra state.

Load pipeline state tensors into the existing steps.

process_action[[lerobot.processor.DataProcessorPipeline.process_action]]

process_action(action: PolicyAction | RobotAction | EnvAction)

Source

Parameters:

action : The action data.

Returns:

The processed action.

Processes only the action part of a transition through the pipeline.

process_complementary_data[[lerobot.processor.DataProcessorPipeline.process_complementary_data]]

process_complementary_data(complementary_data: dict[str, Any])

Source

Parameters:

complementary_data : The complementary data dictionary.

Returns:

The processed complementary data dictionary.

Processes only the complementary data part of a transition through the pipeline.

process_done[[lerobot.processor.DataProcessorPipeline.process_done]]

process_done(done: bool | torch.Tensor)

Source

Parameters:

done : The done flag.

Returns:

The processed done flag.

Processes only the done flag of a transition through the pipeline.

process_info[[lerobot.processor.DataProcessorPipeline.process_info]]

process_info(info: dict[str, Any])

Source

Parameters:

info : The info dictionary.

Returns:

The processed info dictionary.

Processes only the info dictionary of a transition through the pipeline.

process_observation[[lerobot.processor.DataProcessorPipeline.process_observation]]

process_observation(observation: RobotObservation)

Source

Parameters:

observation : The observation dictionary.

Returns:

The processed observation dictionary.

Processes only the observation part of a transition through the pipeline.

process_reward[[lerobot.processor.DataProcessorPipeline.process_reward]]

process_reward(reward: float | torch.Tensor)

Source

Parameters:

reward : The reward value.

Returns:

The processed reward.

Processes only the reward part of a transition through the pipeline.

process_truncated[[lerobot.processor.DataProcessorPipeline.process_truncated]]

process_truncated(truncated: bool | torch.Tensor)

Source

Parameters:

truncated : The truncated flag.

Returns:

The processed truncated flag.

Processes only the truncated flag of a transition through the pipeline.

register_after_step_hook[[lerobot.processor.DataProcessorPipeline.register_after_step_hook]]

register_after_step_hook(fn: Callable[[int, EnvTransition], None])

Source

Parameters:

fn : A callable that accepts the step index and the current transition.

Registers a function to be called after each step.

register_before_step_hook[[lerobot.processor.DataProcessorPipeline.register_before_step_hook]]

register_before_step_hook(fn: Callable[[int, EnvTransition], None])

Source

Parameters:

fn : A callable that accepts the step index and the current transition.

Registers a function to be called before each step.

reset[[lerobot.processor.DataProcessorPipeline.reset]]

reset()

Source

Resets the state of all stateful steps in the pipeline.

save_pretrained[[lerobot.processor.DataProcessorPipeline.save_pretrained]]

save_pretrained(save_directory: str | Path | None = None, repo_id: str | None = None, push_to_hub: bool = False, card_kwargs: dict[str, Any] | None = None, config_filename: str | None = None, **push_to_hub_kwargs)

Source

Parameters:

save_directory : The directory where the pipeline will be saved. If None, saves to HF_LEROBOT_HOME/processors/{sanitized_pipeline_name}.

repo_id : ID of your repository on the Hub. Used only if push_to_hub=true.

push_to_hub : Whether or not to push your object to the Hugging Face Hub after saving it.

card_kwargs : Additional arguments passed to the card template to customize the card.

config_filename : The name of the JSON configuration file. If None, a name is generated from the pipeline's name attribute.

• **push_to_hub_kwargs : Additional key word arguments passed along to the push_to_hub method.

Saves the pipeline's configuration and state to a directory.

This method creates a JSON configuration file that defines the pipeline's structure (name and steps). For each stateful step, it also saves a .safetensors file containing its state dictionary.

state_dict[[lerobot.processor.DataProcessorPipeline.state_dict]]

state_dict()

Source

Returns:

A dictionary mapping suffixless state keys to cloned step state dictionaries.

Return pipeline state tensors grouped by state key.

step_through[[lerobot.processor.DataProcessorPipeline.step_through]]

step_through(data: TInput)

Source

Parameters:

data : The input data.

Yields:

The EnvTransition object, starting with the initial state and then after each processing step.

Processes data step-by-step, yielding the transition at each stage.

This is a generator method useful for debugging and inspecting the intermediate state of the data as it passes through the pipeline.

transform_features[[lerobot.processor.DataProcessorPipeline.transform_features]]

transform_features(initial_features: dict[PipelineFeatureType, dict[str, PolicyFeature]])

Source

Parameters:

initial_features : A dictionary describing the initial features.

Returns:

The final feature description after all transformations.

Applies feature transformations from all steps sequentially.

This method propagates a feature description dictionary through each step's transform_features method, allowing the pipeline to statically determine the output feature specification without processing any real data.

unregister_after_step_hook[[lerobot.processor.DataProcessorPipeline.unregister_after_step_hook]]

unregister_after_step_hook(fn: Callable[[int, EnvTransition], None])

Source

Parameters:

fn : The exact function object that was previously registered.

Raises: ValueError

• ValueError -- If the hook is not found in the list.

Unregisters an 'after_step' hook.

unregister_before_step_hook[[lerobot.processor.DataProcessorPipeline.unregister_before_step_hook]]

unregister_before_step_hook(fn: Callable[[int, EnvTransition], None])

Source

Parameters:

fn : The exact function object that was previously registered.

Raises: ValueError

• ValueError -- If the hook is not found in the list.

Unregisters a 'before_step' hook.

PolicyProcessorPipeline[[lerobot.processor.DataProcessorPipeline]]

lerobot.processor.DataProcessorPipeline[[lerobot.processor.DataProcessorPipeline]]

lerobot.processor.DataProcessorPipeline(*args, **kwargs)