Procesadores para robots y teleoperadores

Esta guía muestra cómo construir y modificar pipelines de procesamiento que conectan teleoperadores (p. ej., teléfono) con robots y conjuntos de datos. Los pipelines estandarizan las conversiones entre diferentes espacios de acción/observación, lo que permite intercambiar teleoperaciones y robots sin reescribir el código de conexión.

Utilizamos los ejemplos de Phone a SO‑100 follower para concretar, pero los mismos patrones se aplican a otros robots.

Lo que aprenderás

• Control absoluto vs. relativo de EE: qué significa cada uno, ventajas y desventajas, y cómo elegirlo para tu tarea.
• Patrón de tres tuberías: cómo mapear acciones de teleoperación → acciones del conjunto de datos → comandos del robot, y observaciones del robot → observaciones del conjunto de datos.
• Adaptadores (to_transition / to_output): cómo convierten los diccionarios crudos en EnvTransition y viceversa para reducir el código repetitivo.
• Contratos de características del conjunto de datos: cómo los pasos declaran características mediante transform_features(...) y cómo agregarlas o fusionarlas para el registro.
• Elegir una representación: cuándo almacenar articulaciones, poses absolutas del extremo efector o deltas relativos del extremo efector, y cómo eso afecta al entrenamiento.
• Guía para personalizar el pipeline: cómo intercambiar robots/URDF de forma segura y ajustar límites, tamaños de paso y opciones como la inicialización de IK.

Control absoluto vs relativo de EE

Los ejemplos de esta guía usan poses absolutas del efector final (EE) porque son fáciles de interpretar. En la práctica, a menudo se prefieren los deltas relativos del EE o la posición de las articulaciones como características de aprendizaje.

Con los procesadores, eliges las características de aprendizaje que deseas utilizar para tu política. Esto podría ser posiciones/velocidades de articulaciones, EE absoluto o posiciones de EE relativas. También puedes elegir almacenar otras características, como pares de articulaciones, corrientes de motor, etc.

Tres pipelines

A menudo componemos tres pipelines. Dependiendo de su configuración, algunos pueden estar vacíos si los espacios de acción y observación ya coinciden. Cada uno de estos pipelines gestiona diferentes conversiones entre distintos espacios de acción y observación. A continuación se presenta una breve explicación de cada pipeline.

1. Pipeline 1: Espacio de acciones de teleoperación → espacio de acciones del conjunto de datos (pose del teléfono → objetivos de EE)
2. Pipeline 2: Espacio de acciones del conjunto de datos → espacio de comandos del robot (objetivos de EE → articulaciones)
3. Pipeline 3: Espacio de observación del robot → espacio de observación del conjunto de datos (articulaciones → pose del extremo efector)

A continuación se muestra un ejemplo de los tres pipelines que utilizamos en los ejemplos de seguidor de teléfono a SO-100:

phone_to_robot_ee_pose_processor = RobotProcessorPipeline[tuple[RobotAction, RobotObservation], RobotAction]( # teleop -> dataset action
    steps=[
        MapPhoneActionToRobotAction(platform=teleop_config.phone_os),
        EEReferenceAndDelta(
            kinematics=kinematics_solver, end_effector_step_sizes={"x": 0.5, "y": 0.5, "z": 0.5}, motor_names=list(robot.bus.motors.keys()),
        ),
        EEBoundsAndSafety(
            end_effector_bounds={"min": [-1.0, -1.0, -1.0], "max": [1.0, 1.0, 1.0]}, max_ee_step_m=0.20,
        ),
        GripperVelocityToJoint(),
    ],
    to_transition=robot_action_observation_to_transition,
    to_output=transition_to_robot_action,
)

robot_ee_to_joints_processor = RobotProcessorPipeline[tuple[RobotAction, RobotObservation], RobotAction]( # dataset action -> robot
    steps=[
        InverseKinematicsEEToJoints(
            kinematics=kinematics_solver, motor_names=list(robot.bus.motors.keys()), initial_guess_current_joints=True,
        ),
    ],
    to_transition=robot_action_observation_to_transition,
    to_output=transition_to_robot_action,
)

robot_joints_to_ee_pose = RobotProcessorPipeline[RobotObservation, RobotObservation]( # robot obs -> dataset obs
    steps=[
        ForwardKinematicsJointsToEE(kinematics=kinematics_solver, motor_names=list(robot.bus.motors.keys()))
    ],
    to_transition=observation_to_transition,
    to_output=transition_to_observation,
)

Por qué to_transition / to_output

Para convertir de robot/teleoperador a pipeline y viceversa, utilizamos los adaptadores de pipeline to_transition y to_output. Estos estandarizan las conversiones para reducir el código repetitivo y forman el puente entre los diccionarios crudos del robot y del teleoperador y el formato EnvTransition del pipeline. En los ejemplos de teléfono a seguidor SO-100, utilizamos los siguientes adaptadores:

• robot_action_observation_to_transition: transforma un par (action, robot observation) en una transición de pipeline (los pasos de EE e IK necesitan las posiciones actuales de las articulaciones del robot).
• transition_to_robot_action: transforma la transición de la pipeline en un diccionario de acción del robot.
• observation_to_transition: transforma el diccionario de observación del robot en una transición de la pipeline.
• transition_to_observation: transforma la transición de la pipeline en un diccionario de observación.

Consulte src/lerobot/processor/converters.py para más detalles.

Contratos de características de los conjuntos de datos

Las características del dataset se determinan por las claves guardadas en el dataset. Cada paso puede declarar qué características modifica en un contrato llamado transform_features(...). Una vez que construyes un procesador, este puede agregar todas estas características con aggregate_pipeline_dataset_features() y fusionar múltiples diccionarios de características con combine_feature_dicts(...).

A continuación se muestra un ejemplo de cómo declaramos características con el método transform_features en los ejemplos de seguidor de teléfono a SO-100:

    def transform_features(
        self, features: dict[PipelineFeatureType, dict[str, PolicyFeature]]
    ) -> dict[PipelineFeatureType, dict[str, PolicyFeature]]:
        # We only use the ee pose in the dataset, so we don't need the joint positions
        for n in self.motor_names:
            features[PipelineFeatureType.ACTION].pop(f"{n}.pos", None)
        # We specify the dataset features of this step that we want to be stored in the dataset
        for k in ["x", "y", "z", "wx", "wy", "wz", "gripper_pos"]:
            features[PipelineFeatureType.ACTION][f"ee.{k}"] = PolicyFeature(
                type=FeatureType.STATE, shape=(1,)
            )
        return features

Aquí declaramos qué PolicyFeatures modificamos en este paso, de modo que sepamos qué características podemos esperar al ejecutar el procesador. Estas características se pueden agregar y utilizar para crear las características del conjunto de datos.

A continuación se muestra un ejemplo de cómo agregamos y fusionamos características en el ejemplo de registro de teléfono a SO-100:

features=combine_feature_dicts(
        # Run the feature contract of the pipelines
        # This tells you how the features would look like after the pipeline steps
        aggregate_pipeline_dataset_features(
            pipeline=phone_to_robot_ee_pose_processor,
            initial_features=create_initial_features(action=phone.action_features), # <- Action features we can expect, these come from our teleop device (phone) and action processor
            use_videos=True,
        ),
        aggregate_pipeline_dataset_features(
            pipeline=robot_joints_to_ee_pose,
            initial_features=create_initial_features(observation=robot.observation_features), # <- Observation features we can expect, these come from our robot and observation processor
            use_videos=True,
            patterns=["observation.state.ee"], # <- Here you could optionally filter the features we want to store in the dataset, with a specific pattern

        ),
    ),

Cómo funciona:

• aggregate_pipeline_dataset_features(...): aplica transform_features en todo el pipeline y filtra por patrones (las imágenes se incluyen cuando use_videos=True, y las características de estado se incluyen cuando se especifica patterns).
• combine_feature_dicts(...): combina varios diccionarios de características.
• La grabación con record_loop(...) utiliza build_dataset_frame(...) para construir marcos coherentes con dataset.features antes de llamar a add_frame(...) para añadir el marco al conjunto de datos.

Orientaciones para personalizar pipelines de robots

Puede almacenar cualquiera de las siguientes características como su espacio de acción/observación:

• Posiciones articulares
• Posas EE absolutas
• Deltas relativos de EE
• Otras características: velocidad conjunta, pares, etc.

Elija lo que desea utilizar para el espacio de acciones y observaciones de su política y configure o modifique los pipelines y pasos en consecuencia.

Diferentes robots

• Puedes reutilizar fácilmente los pipelines, por ejemplo, para usar otro robot con teleoperación por teléfono, modificando los ejemplos y cambiando el RobotKinematics (URDF) y motor_names para usar tu propio robot con teleoperación por teléfono. Además, debes asegurarte de que target_frame_name apunte a tu pinza/muñeca.

La seguridad ante todo

• Al cambiar de pipelines, comience con límites estrictos e implemente medidas de seguridad al trabajar con robots reales.
• Se recomienda comenzar con la simulación y luego pasar a los robots reales.

¡Eso es todo! Esperamos que esta guía te ayude a empezar a personalizar tus pipelines de robots. Si te encuentras con algún problema en cualquier momento, únete a nuestra comunidad de Discord para recibir soporte.