Processadores para Robôs e Teleoperadores

Este guia mostra como construir e modificar pipelines de processamento que conectam teleoperadores (por exemplo, telefone) a robôs e conjuntos de dados. Os pipelines padronizam as conversões entre diferentes espaços de ação/observação, permitindo trocar teleops e robôs sem reescrever o código de integração.

Usamos os exemplos de Phone para SO‑100 follower para concretude, mas os mesmos padrões se aplicam a outros robôs.

O que você vai aprender

• Controle absoluto vs. relativo de EE: o que cada um significa, os prós e contras e como escolher para sua tarefa.
• Padrão de três pipelines: como mapear ações de teleoperação → ações do dataset → comandos do robô, e observações do robô → observações do dataset.
• Adaptadores (to_transition / to_output): Como eles convertem dicionários brutos em EnvTransition e de volta, para reduzir o código repetitivo.
• Contratos de recursos do dataset: como as etapas declaram recursos via transform_features(...) e como agregá-los/uni-los para gravação.
• Escolhendo uma representação: quando armazenar juntas, poses absolutas do EE ou deltas relativos do EE — e como isso afeta o treinamento.
• Orientações para personalização do pipeline: como trocar robôs/URDFs com segurança e ajustar limites, tamanhos de passo e opções como inicialização de IK.

Controle absoluto vs relativo de EE

Os exemplos neste guia usam poses absolutas do efector final (EE) porque são fáceis de raciocinar. Na prática, deltas relativos do EE ou a posição das juntas são frequentemente preferidos como características de aprendizado.

Com os processadores, você escolhe os recursos de aprendizado que deseja usar para sua política. Isso pode incluir posições/velocidades das juntas, EE absoluto ou posições relativas do EE. Você também pode escolher armazenar outros recursos, como torques das juntas, correntes dos motores, etc.

Três pipelines

Frequentemente, compomos três pipelines. Dependendo da sua configuração, alguns podem ficar vazios se os espaços de ação e observação já corresponderem. Cada um desses pipelines lida com diferentes conversões entre diferentes espaços de ação e observação. Abaixo, há uma breve explicação de cada pipeline.

1. Pipeline 1: Espaço de ação de teleop → espaço de ação do dataset (pose do telefone → alvos do EE)
2. Pipeline 2: Espaço de ação do conjunto de dados → espaço de comando do robô (alvos de EE → juntas)
3. Pipeline 3: Espaço de observação do robô → espaço de observação do conjunto de dados (juntas → pose do EE)

Abaixo está um exemplo dos três pipelines que usamos nos exemplos de seguidor de telefone para 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 que to_transition / to_output

Para converter de robot/teleoperator para pipeline e vice-versa, usamos os adaptadores de pipeline to_transition e to_output. Eles padronizam as conversões para reduzir código repetitivo e formam a ponte entre os dicionários brutos do robot e do teleoperator e o formato EnvTransition do pipeline. Nos exemplos de telefone para seguidor SO-100, usamos os seguintes adaptadores:

• robot_action_observation_to_transition: transforma um par (action, robot observation) em uma transição de pipeline (as etapas de EE e IK precisam das posições atuais das juntas do robô).
• transition_to_robot_action: transforma a transição do pipeline em um dicionário de ação do robô.
• observation_to_transition: transforma o dicionário de observação do robô em uma transição de pipeline.
• transition_to_observation: transforma a transição do pipeline em um dicionário de observação.

Consulte src/lerobot/processor/converters.py para mais detalhes.

Contratos de recursos de dataset

Os recursos do dataset são determinados pelas chaves salvas no dataset. Cada etapa pode declarar quais recursos ela modifica em um contrato chamado transform_features(...). Uma vez que você constrói um processador, o processador pode então agregar todos esses recursos com aggregate_pipeline_dataset_features() e mesclar múltiplos dicionários de recursos com combine_feature_dicts(...).

Abaixo está um exemplo de como declaramos recursos com o método transform_features nos exemplos de seguidor de telefone para 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

Aqui declaramos quais PolicyFeatures modificamos nesta etapa, para que saibamos quais recursos podemos esperar ao executar o processador. Esses recursos podem, em seguida, ser agregados e usados para criar os recursos do dataset.

Abaixo está um exemplo de como agregamos e mesclamos recursos no exemplo de registro de telefone para 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

        ),
    ),

Como funciona:

• aggregate_pipeline_dataset_features(...): aplica transform_features em todo o pipeline e filtra por padrões (imagens são incluídas quando use_videos=True, e recursos de estado são incluídos quando patterns é especificado).
• combine_feature_dicts(...): combina vários dicionários de recursos.
• Gravar com record_loop(...) usa build_dataset_frame(...) para construir frames consistentes com dataset.features antes de chamarmos add_frame(...) para adicionar o frame ao dataset.

Orientações para personalizar pipelines de robôs

Você pode armazenar qualquer um dos seguintes recursos como seu espaço de ação/observação:

• Posições das articulações
• Poses absolutas de EE
• Deltas relativos de EE
• Outras funcionalidades: velocidade conjunta, torques, etc.

Escolha o que deseja usar para o espaço de ação e o espaço de observação da sua política e configure/modifique os pipelines e as etapas de acordo.

Robôs diferentes

• Você pode reutilizar facilmente pipelines, por exemplo, para usar outro robô com teleop por telefone, modificando os exemplos e substituindo o RobotKinematics (URDF) e o motor_names para usar seu próprio robô com teleop por telefone. Além disso, você deve garantir que o target_frame_name aponte para sua garra/pulso.

Segurança em primeiro lugar

• Ao alterar pipelines, comece com limites estritos e implemente etapas de segurança ao trabalhar com robôs reais.
• É aconselhável começar com simulação e, em seguida, passar para robôs reais.

É isso! Esperamos que este guia ajude você a começar a personalizar seus pipelines de robôs. Se você encontrar algum problema em qualquer momento, entre na nossa comunidade do Discord para obter suporte.