fix(so_follower): check the observation for None before copying it (#4255)

Four kinematic processor steps read the observation as

    observation = self.transition.get(TransitionKey.OBSERVATION).copy()
    if observation is None:
        raise ValueError("Joints observation is require for computing robot kinematics")

so `.copy()` runs first and the guard below it is unreachable. A transition
without an observation raises `AttributeError: 'NoneType' object has no
attribute 'copy'` instead of the intended message.

That transition is not hypothetical: `RobotProcessorPipeline.process_action`
builds one with `create_transition(action=action)`, which sets
`TransitionKey.OBSERVATION` to None.

Reads the value first, checks it, then copies. Affects EEReferenceAndDelta,
InverseKinematicsEEToJoints, GripperVelocityToJoint and InverseKinematicsRLStep.
Adds a parametrised regression test covering all four; each fails with the
AttributeError if the fix is reverted.

Co-authored-by: Steven Palma <imstevenpmwork@ieee.org>
This commit is contained in:
Lin Junrong
2026-07-31 22:12:18 +08:00
committed by GitHub
parent 29fcf057dc
commit 732a12108e
2 changed files with 56 additions and 8 deletions
@@ -17,9 +17,14 @@
import pytest
from lerobot.configs import FeatureType, PipelineFeatureType, PolicyFeature
from lerobot.processor.converters import create_transition
from lerobot.robots.so_follower.robot_kinematic_processor import (
EEReferenceAndDelta,
ForwardKinematicsJointsToEEAction,
ForwardKinematicsJointsToEEObservation,
GripperVelocityToJoint,
InverseKinematicsEEToJoints,
InverseKinematicsRLStep,
)
MOTOR_NAMES = ["shoulder_pan", "shoulder_lift", "elbow_flex", "wrist_flex", "wrist_roll", "gripper"]
@@ -43,3 +48,38 @@ def test_fk_feature_schema(step_cls, bucket, feature_type):
out = step_cls(kinematics=None, motor_names=MOTOR_NAMES).transform_features(features)[bucket]
assert set(out) == EE_KEYS
assert {feature.type for feature in out.values()} == {feature_type}
EE_ACTION = dict.fromkeys(EE_KEYS, 0.0)
@pytest.mark.parametrize(
("step", "action"),
[
(
EEReferenceAndDelta(kinematics=None, end_effector_step_sizes={}, motor_names=MOTOR_NAMES),
dict(EE_ACTION),
),
(InverseKinematicsEEToJoints(kinematics=None, motor_names=MOTOR_NAMES), dict(EE_ACTION)),
(GripperVelocityToJoint(), {**EE_ACTION, "ee.gripper_vel": 0.0}),
(InverseKinematicsRLStep(kinematics=None, motor_names=MOTOR_NAMES), dict(EE_ACTION)),
],
ids=[
"ee_reference_and_delta",
"inverse_kinematics_ee_to_joints",
"gripper_velocity_to_joint",
"inverse_kinematics_rl_step",
],
)
def test_missing_observation_raises_value_error(step, action):
"""A transition without an observation must surface the documented ValueError.
`RobotProcessorPipeline.process_action` builds its transition with
`create_transition(action=...)`, which sets `TransitionKey.OBSERVATION` to None.
These steps used to call `.copy()` on that before the None check, so the guard
below them was unreachable and an AttributeError escaped instead.
"""
transition = create_transition(action=action)
with pytest.raises(ValueError, match="Joints observation"):
step(transition)