LeRobot documentation

Processeurs d'environnement

LeRobot

You are viewing main version, which requires installation from source. If you'd like regular pip install, checkout the latest stable version (v0.6.1).

Join the Hugging Face community

and get access to the augmented documentation experience

Collaborate on models, datasets and Spaces

Faster examples with accelerated inference

Switch between documentation themes

to get started

Processeurs d'environnement

Les processeurs d’environnement constituent une couche essentielle de l’architecture de traitement des données de LeRobot, qui gère les transformations spécifiques à l’environnement, distinctes du traitement propre à la politique. Cette séparation des responsabilités permet un code plus propre, une meilleure modularité et une expérimentation plus aisée avec différents environnements et politiques.

Pourquoi des processeurs d'environnement ?

Lorsque l’on travaille avec différents environnements robotiques (LIBERO, MetaWorld, Aloha, etc.), chaque environnement possède souvent des formats de données, des systèmes de coordonnées et des conventions propres qui nécessitent une standardisation avant le traitement par la politique. Sans processeurs d’environnement, ces transformations seraient :

Code en dur dans le code de l'environnement - Rendre difficile l'expérimentation avec différentes représentations d'état
Dupliqué entre les politiques - Chaque politique devrait gérer les particularités spécifiques à l'environnement
Mélangé avec la logique de politique - Violation de la séparation des préoccupations et difficulté accrue du débogage

Les processeurs d'environnement résolvent ce problème en fournissant une couche de traitement dédiée entre les observations brutes de l'environnement et les entrées de la politique.

Le pipeline de traitement

Voici comment les données circulent à travers le pipeline de traitement complet lors de l'évaluation :

# In lerobot_eval.py rollout() function:

# 1. Raw environment observation (numpy arrays, various formats)
raw_observation = env.step(action)

# 2. Convert numpy to torch, normalize images [0,1]
observation = preprocess_observation(raw_observation)

# 3. Add the task description (for multi-task environments)
observation["task"] = list(env.call("task_description"))

# 4. ENVIRONMENT-SPECIFIC preprocessing (NEW!)
#    - Flatten robot states
#    - Rotate images to match dataset conventions
#    - Handle environment-specific coordinate systems
observation = env_preprocessor(observation)

# 5. POLICY-SPECIFIC preprocessing
#    - Normalize with dataset statistics
#    - Add batch dimensions
#    - Move to GPU
#    - Tokenize language instructions
observation = preprocessor(observation)

# 6. Policy inference
action = policy.select_action(observation)

# 7. POLICY-SPECIFIC postprocessing
#    - Unnormalize actions
#    - Remove batch dimensions
action = postprocessor(action)

# 8. ENVIRONMENT-SPECIFIC postprocessing (NEW!)
#    - Convert action formats if needed
#    - Apply environment-specific constraints
action_transition = {"action": action}
action_transition = env_postprocessor(action_transition)
action = action_transition["action"]

# 9. Execute in environment
env.step(action)

Les avantages

1. Séparation des responsabilités

Les processeurs d’environnement gèrent les transformations spécifiques au format de données de l’environnement, tandis que les processeurs de politique gèrent les transformations spécifiques aux exigences du modèle.

# ❌ Before: Mixed concerns
class LiberoVLAPolicy:
    def preprocess(self, obs):
        # Environment-specific: Flatten robot state (shouldn't be in policy!)
        state = self._flatten_robot_state(obs["robot_state"])
        # Policy-specific: Normalize with dataset stats
        state = self.normalizer(state)
        return state

# ✅ After: Clear separation
# Environment processor: Handles LIBERO's nested robot state
env_preprocessor = LiberoProcessorStep()  # Flattens robot_state

# Policy processor: Handles model requirements
policy_preprocessor = NormalizerProcessorStep(stats=dataset_stats)

2. Flexibilité et réutilisabilité

La même politique peut fonctionner avec différents processeurs d’environnement, et le même processeur d’environnement peut fonctionner avec différentes politiques :

# Use SmolVLA policy with LIBERO environment
libero_preprocessor, libero_postprocessor = make_env_pre_post_processors(
    env_cfg=libero_cfg,
    policy_cfg=smolvla_cfg,
)
smolvla_preprocessor, smolvla_postprocessor = make_pre_post_processors(smolvla_cfg)

# Or use ACT policy with the same LIBERO environment
libero_preprocessor, libero_postprocessor = make_env_pre_post_processors(
    env_cfg=libero_cfg,
    policy_cfg=act_cfg,
)
act_preprocessor, act_postprocessor = make_pre_post_processors(act_cfg)

3. Expérimentation plus simple

Souhaitez-vous essayer différentes représentations d'état pour LIBERO ? Il suffit de créer un nouveau processeur :

# Original: 8D state (pos + quat→axisangle + gripper)
@ProcessorStepRegistry.register("libero_processor")
class LiberoProcessorStep(ObservationProcessorStep):
    def _process_observation(self, obs):
        eef_pos = robot_state["eef"]["pos"]          # 3D
        eef_axisangle = quat2axisangle(quat)         # 3D
        gripper = robot_state["gripper"]["qpos"]     # 2D
        state = torch.cat([eef_pos, eef_axisangle, gripper], dim=-1)  # 8D
        return state

# Experiment: Add velocity for better control
@ProcessorStepRegistry.register("libero_velocity_processor")
class LiberoVelocityProcessorStep(ObservationProcessorStep):
    def _process_observation(self, obs):
        # Include velocities for 14D state
        eef_pos = robot_state["eef"]["pos"]          # 3D
        eef_axisangle = quat2axisangle(quat)         # 3D
        eef_vel = robot_state["eef"]["vel"]          # 3D  (NEW)
        gripper_pos = robot_state["gripper"]["qpos"] # 2D
        gripper_vel = robot_state["gripper"]["qvel"] # 3D  (NEW)
        state = torch.cat([eef_pos, eef_axisangle, eef_vel,
                          gripper_pos, gripper_vel], dim=-1)  # 14D
        return state

4. Code d’environnement plus propre

Les environnements exposent toutes les données disponibles sans avoir besoin de savoir ce que les modèles en aval utiliseront :

# LIBERO environment exposes full robot state
observation = {
    "pixels": {"image": img, "image2": img2},
    "robot_state": {
        "eef": {"pos": ..., "quat": ..., "vel": ..., "mat": ..., "axisangle": ...},
        "gripper": {"qpos": ..., "qvel": ...},
        "joints": {"pos": ..., "vel": ...}
    }
}

# Environment processor decides what to use
# Policy processor handles model-specific transformations

Utilisation des processeurs d'environnement

Fonction d'usine

La fonction make_env_pre_post_processors suit le même modèle que make_pre_post_processors pour les politiques :

from lerobot.envs import make_env_pre_post_processors, PushtEnv
from lerobot.envs.configs import LiberoEnv

# policy_cfg is your policy's config (e.g. cfg.policy in lerobot-eval); a few policies,
# such as XVLA, provide their own environment processors

# For LIBERO: Returns LiberoProcessorStep in preprocessor
libero_cfg = LiberoEnv(task="libero_spatial")
env_preprocessor, env_postprocessor = make_env_pre_post_processors(libero_cfg, policy_cfg=policy_cfg)

# For other environments: Returns identity processors (no-op)
pusht_cfg = PushtEnv()
env_preprocessor, env_postprocessor = make_env_pre_post_processors(pusht_cfg, policy_cfg=policy_cfg)

Implémentation dans envs/factory.py

def make_env_pre_post_processors(
    env_cfg: EnvConfig,
) -> tuple[
    PolicyProcessorPipeline[dict[str, Any], dict[str, Any]],
    PolicyProcessorPipeline[dict[str, Any], dict[str, Any]],
]:
    """
    Create preprocessor and postprocessor pipelines for environment observations.

    Args:
        env_cfg: The configuration of the environment.

    Returns:
        A tuple containing:
            - preprocessor: Pipeline that processes environment observations
            - postprocessor: Pipeline that processes environment outputs
    """
    # For LIBERO environments, add the LiberoProcessorStep to preprocessor
    if isinstance(env_cfg, LiberoEnv) or "libero" in env_cfg.type:
        preprocessor = PolicyProcessorPipeline(steps=[LiberoProcessorStep()])
    else:
        # For all other environments, return an identity preprocessor
        preprocessor = PolicyProcessorPipeline(steps=[])

    # Postprocessor is currently identity for all environments
    # Future: Could add environment-specific action transformations
    postprocessor = PolicyProcessorPipeline(steps=[])

    return preprocessor, postprocessor

Intégration dans l'évaluation

Dans lerobot_eval.py, les processeurs d’environnement sont créés une seule fois et utilisés tout au long :

def eval_main(cfg: EvalPipelineConfig):
    # Create environment
    envs = make_env(cfg.env, n_envs=cfg.eval.batch_size)

    # Create policy
    policy = make_policy(cfg=cfg.policy, env_cfg=cfg.env)

    # Create policy processors
    preprocessor, postprocessor = make_pre_post_processors(
        policy_cfg=cfg.policy,
        pretrained_path=cfg.policy.pretrained_path,
    )

    # Create environment processors (NEW!)
    env_preprocessor, env_postprocessor = make_env_pre_post_processors(env_cfg=cfg.env, policy_cfg=cfg.policy)

    # Run evaluation with both processor types
    eval_policy_all(
        envs=envs,
        policy=policy,
        env_preprocessor=env_preprocessor,      # Environment-specific
        env_postprocessor=env_postprocessor,    # Environment-specific
        preprocessor=preprocessor,              # Policy-specific
        postprocessor=postprocessor,            # Policy-specific
        n_episodes=cfg.eval.n_episodes,
    )

Exemple : Processeur d’environnement LIBERO

Le LiberoProcessorStep illustre un processeur d’environnement dans un contexte réel :

from lerobot.processor import ObservationProcessorStep

@dataclass
@ProcessorStepRegistry.register(name="libero_processor")
class LiberoProcessorStep(ObservationProcessorStep):
    """
    Processes LIBERO observations into the LeRobot format.

    **State Processing:**
    - Extracts end-effector position (3D)
    - Converts quaternion to axis-angle representation (3D)
    - Extracts gripper joint positions (2D)
    - Concatenates into 8D state vector

    **Image Processing:**
    - Rotates images 180° to match HuggingFaceVLA/libero convention
    """

    def _process_observation(self, observation):
        processed_obs = observation.copy()

        # Process images: Flip 180° for camera convention
        for key in list(processed_obs.keys()):
            if key.startswith("observation.images."):
                img = processed_obs[key]
                img = torch.flip(img, dims=[2, 3])  # Flip H and W
                processed_obs[key] = img

        # Process robot_state: Flatten to 8D vector
        if "observation.robot_state" in processed_obs:
            robot_state = processed_obs.pop("observation.robot_state")

            eef_pos = robot_state["eef"]["pos"]           # (B, 3)
            eef_quat = robot_state["eef"]["quat"]         # (B, 4)
            gripper_qpos = robot_state["gripper"]["qpos"] # (B, 2)

            # Convert quaternion to axis-angle
            eef_axisangle = self._quat2axisangle(eef_quat)  # (B, 3)

            # Concatenate into single state vector
            state = torch.cat((eef_pos, eef_axisangle, gripper_qpos), dim=-1)
            state = state.float()

            processed_obs["observation.state"] = state

        return processed_obs

Pourquoi ces transformations ?

Rotation des images : le jeu de données HuggingFaceVLA/libero contient des images tournées de 180° par rapport au simulateur LIBERO brut. Le processeur gère cette différence de convention afin que les politiques entraînées sur le jeu de données fonctionnent sans accroc.
Aplatissage de l'état : l'environnement LIBERO brut expose des dictionnaires imbriqués contenant toutes les informations d'état disponibles (position, quaternion, vitesse, représentation matricielle, etc.). Le processeur :
- Sélectionne les composants pertinents (pos, quat, gripper)
- Convertit un quaternion en axe-angle (plus adapté à l'apprentissage)
- Aplatit en un vecteur 8D unique que les politiques attendent
Flexibilité : l'environnement expose toujours toutes les données brutes. Si vous souhaitez tester différentes représentations d'état (par exemple, inclure les vitesses, utiliser une représentation matricielle au lieu de l'axe-angle), vous pouvez créer un nouveau processeur sans modifier le code de l'environnement.

Ajout de processeurs d'environnement pour de nouveaux environnements

Pour ajouter des processeurs d'environnement pour un nouvel environnement :

1. Créer l'étape du processeur

# In src/lerobot/processor/env_processor.py

@dataclass
@ProcessorStepRegistry.register(name="myenv_processor")
class MyEnvProcessorStep(ObservationProcessorStep):
    """Process observations from MyEnv."""

    def _process_observation(self, observation):
        processed = observation.copy()

        # Your environment-specific transformations
        if "myenv.specific.state" in processed:
            state = processed.pop("myenv.specific.state")
            # Transform to standard format
            processed["observation.state"] = self._transform_state(state)

        return processed

2. Mettez à jour votre sous-classe EnvConfig

# In src/lerobot/envs/factory.py

def make_env_pre_post_processors(env_cfg: EnvConfig):
    if isinstance(env_cfg, LiberoEnv) or "libero" in env_cfg.type:
        preprocessor = PolicyProcessorPipeline(steps=[LiberoProcessorStep()])
    elif isinstance(env_cfg, MyEnvConfig) or "myenv" in env_cfg.type:
        preprocessor = PolicyProcessorPipeline(steps=[MyEnvProcessorStep()])
    else:
        preprocessor = PolicyProcessorPipeline(steps=[])

    postprocessor = PolicyProcessorPipeline(steps=[])
    return preprocessor, postprocessor

3. Utilisation dans l'évaluation

Aucune modification n'est nécessaire ! Le script d'évaluation utilise automatiquement le processeur approprié :

lerobot-eval \
    --policy.path=lerobot/my_policy \
    --env.type=myenv \  # Automatically uses MyEnvProcessorStep
    --eval.n_episodes=10

Futur : Postprocesseurs d'environnement

Actuellement, les postprocesseurs sont des identités (sans effet) pour tous les environnements. Les cas d'utilisation futurs incluent :

Transformations de l'espace d'actions

@dataclass
class MyEnvActionPostprocessor(ProcessorStep):
    """Convert policy actions to environment-specific format."""

    def __call__(self, transition: EnvTransition) -> EnvTransition:
        action = transition["action"]

        # Example: Convert from Cartesian to joint space
        if self.action_space == "joint":
            action = self.ik_solver(action)

        # Example: Apply environment-specific safety limits
        action = torch.clamp(action, self.min_action, self.max_action)

        transition["action"] = action
        return transition

Conversions de systèmes de coordonnées

@dataclass
class CoordinateTransformPostprocessor(ProcessorStep):
    """Transform actions between coordinate systems."""

    def __call__(self, transition: EnvTransition) -> EnvTransition:
        action = transition["action"]

        # Example: Policy outputs in world frame, env expects base frame
        action = self.world_to_base_transform(action)

        transition["action"] = action
        return transition

Bonnes pratiques

Gardez les processeurs d’environnement simples : ils ne doivent traiter que les problèmes de format de données spécifiques à l’environnement, et non des transformations complexes liées à l’apprentissage.
Utiliser des processeurs de politique pour les exigences du modèle : la normalisation, le groupement par lots, l'affectation aux appareils et la tokenisation doivent être gérés par les processeurs de politique.
Exposer toutes les données des environnements : Laisser les processeurs décider de ce qu'ils doivent utiliser plutôt que de coder en dur des choix dans l'environnement.
Conventions de document : Documentez clairement les conventions de système de coordonnées, les orientations de caméra ou les formats de données que votre processeur gère.
Tester de manière indépendante : les processeurs d'environnement doivent être testables sans charger de politiques ou d'environnements complets.

Résumé

Les processeurs d'environnement assurent une séparation claire entre les transformations de données spécifiques à l'environnement et les exigences du modèle spécifiques à la politique. Cette architecture :

✅ Permet de mener facilement des expériences avec différentes représentations d'état
✅ Permet aux politiques de fonctionner de manière transparente dans différents environnements
✅ Maintient le code de l'environnement centré sur l'interface simulation/matériel
✅ Rends les pipelines de processeurs plus maintenables et débogables
✅ Respecte le principe de responsabilité unique

L’idée clé : les environnements définissent les formats de données, les processeurs les standardisent et les politiques consomment des données standardisées. Chaque couche a une responsabilité claire et ciblée.

Update on GitHub

←Processeurs pour robots et téléopérateurs Représentations d'actions→