fix(go2): record camera calibration

[ruthwikdasyam]

Contribution path

• Small, safe change that does not need a tracking issue

Problem

Go2 recordings include camera images but omit the corresponding CameraInfo stream.
Consumers therefore need a local calibration file or internal GO2Connection.camera_info_static knowledge to interpret images, so exported recordings are not self-contained.

Solution

Add a typed camera_info input to both Go2 recorder variants: Go2Memory and Go2Mid360Recorder.

How to Test

uv run pytest dimos/robot/unitree/go2/test_recorders.py -q

Also validated the neighboring Go2 blueprint tests, blueprint registry generation, focused mypy/ruff checks, and pre-commit run --all-files.

AI assistance

OpenCode with GPT-5.6 Sol implemented the change, added the regression test, ran validation, and prepared this draft PR under human direction.

Checklist

• I have read and approved the CLA.

[codecov]

❌ 2 Tests Failed:

Tests completed	Failed	Passed	Skipped
4871	2	4869	87

View the full list of 2 ❄️ flaky test(s)

dimos.e2e_tests.test_manipulation_planning_groups::test_dual_arm_plans_and_dispatches_both_arms_through_control_coordinator

Flake rate in main: 100.00% (Passed 0 times, Failed 3 times)

Stack Traces | 120s run time

lcm_spy = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x7ae9ea644ad0>
start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x7ae9e2a462a0>

    def test_dual_arm_plans_and_dispatches_both_arms_through_control_coordinator(
        lcm_spy: LcmSpy,
        start_blueprint: Callable[..., DimosCliCall],
    ) -> None:
        """Plan one generated plan over both arms and dispatch through one trajectory task."""
>       _start_openarm_mock_planner(start_blueprint, lcm_spy)

lcm_spy    = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x7ae9ea644ad0>
start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x7ae9e2a462a0>

dimos/e2e_tests/test_manipulation_planning_groups.py:190: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
dimos/e2e_tests/test_manipulation_planning_groups.py:151: in _start_openarm_mock_planner
    lcm_spy.wait_for_saved_topic(JOINT_STATE_TOPIC, timeout=120.0)
        lcm_spy    = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x7ae9ea644ad0>
        start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x7ae9e2a462a0>
dimos/e2e_tests/lcm_spy.py:97: in wait_for_saved_topic
    wait_until(
        condition  = <function LcmSpy.wait_for_saved_topic.<locals>.condition at 0x7ae9e2a46840>
        self       = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x7ae9ea644ad0>
        timeout    = 120.0
        topic      = '/coordinator_joint_state#sensor_msgs.JointState'
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

predicate = <function LcmSpy.wait_for_saved_topic.<locals>.condition at 0x7ae9e2a46840>

    def wait_until(
        predicate: Callable[[], bool],
        *,
        timeout: float,
        interval: float = 0.1,
        message: str | None = None,
    ) -> None:
        """Poll ``predicate`` until it returns truthy or ``timeout`` elapses."""
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            if predicate():
                return
            time.sleep(interval)
>       raise TimeoutError(message or f"Timed out after {timeout}s waiting for condition")
E       TimeoutError: Timeout waiting for topic /coordinator_joint_state#sensor_msgs.JointState

deadline   = 4400777.505425674
interval   = 0.1
message    = 'Timeout waiting for topic /coordinator_joint_state#sensor_msgs.JointState'
predicate  = <function LcmSpy.wait_for_saved_topic.<locals>.condition at 0x7ae9e2a46840>
timeout    = 120.0

.../utils/testing/waiting.py:35: TimeoutError

dimos.e2e_tests.test_manipulation_planning_groups::test_single_arm_plans_and_executes_through_control_coordinator

Flake rate in main: 25.00% (Passed 12 times, Failed 4 times)

Stack Traces | 120s run time

lcm_spy = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x7ae9e861c1a0>
start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x7ae9e2a45d00>

    def test_single_arm_plans_and_executes_through_control_coordinator(
        lcm_spy: LcmSpy,
        start_blueprint: Callable[..., DimosCliCall],
    ) -> None:
        """Plan with one arm and execute through its trajectory task."""
>       _start_openarm_mock_planner(start_blueprint, lcm_spy)

lcm_spy    = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x7ae9e861c1a0>
start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x7ae9e2a45d00>

dimos/e2e_tests/test_manipulation_planning_groups.py:159: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
dimos/e2e_tests/test_manipulation_planning_groups.py:151: in _start_openarm_mock_planner
    lcm_spy.wait_for_saved_topic(JOINT_STATE_TOPIC, timeout=120.0)
        lcm_spy    = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x7ae9e861c1a0>
        start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x7ae9e2a45d00>
dimos/e2e_tests/lcm_spy.py:97: in wait_for_saved_topic
    wait_until(
        condition  = <function LcmSpy.wait_for_saved_topic.<locals>.condition at 0x7ae9e2a45f80>
        self       = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x7ae9e861c1a0>
        timeout    = 120.0
        topic      = '/coordinator_joint_state#sensor_msgs.JointState'
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

predicate = <function LcmSpy.wait_for_saved_topic.<locals>.condition at 0x7ae9e2a45f80>

    def wait_until(
        predicate: Callable[[], bool],
        *,
        timeout: float,
        interval: float = 0.1,
        message: str | None = None,
    ) -> None:
        """Poll ``predicate`` until it returns truthy or ``timeout`` elapses."""
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            if predicate():
                return
            time.sleep(interval)
>       raise TimeoutError(message or f"Timed out after {timeout}s waiting for condition")
E       TimeoutError: Timeout waiting for topic /coordinator_joint_state#sensor_msgs.JointState

deadline   = 4400657.311378627
interval   = 0.1
message    = 'Timeout waiting for topic /coordinator_joint_state#sensor_msgs.JointState'
predicate  = <function LcmSpy.wait_for_saved_topic.<locals>.condition at 0x7ae9e2a45f80>
timeout    = 120.0

.../utils/testing/waiting.py:35: TimeoutError

To view more test analytics, go to the Test Analytics Dashboard
_{📋 Got 3 mins? Take this short survey to help us improve Test Analytics.}

[greptile-apps]

Greptile Summary

Adds CameraInfo inputs to the standard Go2 recorder and the Go2/Mid-360 recorder so image calibration metadata can be included in recordings. The declarations match the existing GO2Connection output name and type used by automatic stream wiring.

T-Rex validation blocked

The focused blueprint-construction check could not import the Go2 blueprints because the Python package structlog is missing. This prevented runtime construction of the recorder blueprints and verification of their camera-info connection.

Confidence Score: 5/5

No product defect was established in the changed recorder declarations.

The final review contains no confirmed finding. Static inspection shows both recorder inputs use the same unremapped camera_info stream name and CameraInfo type as the Go2 connection output; runtime construction remains unavailable because of the missing package.

Files Needing Attention: No source file requires a corrective change. Restore the complete Python runtime dependency set before running the focused Go2 blueprint wiring check.

T-Rex Logs

What T-Rex did

• T-Rex authored the focused blueprint wiring script to import both Go2 blueprints, locate the GO2Connection producer and the recorder consumers, and verify matching stream names, types, and directions.
• T-Rex attempted to run the wiring-check script with the recording option, but the run halted during blueprint import with ModuleNotFoundError: No module named 'structlog', signaling a missing runtime dependency and preventing blueprint construction or exercise.
• T-Rex produced a finding-proof for a posted P2 finding, and the reviewer comment provides the details of that finding.

_{Ran code and verified through T-Rex}

Comments Outside Diff (1)

1. dimos/robot/unitree/go2/blueprints/smart/unitree_go2.py, line 104 (link)

   CameraInfo blueprint wiring could not be executed

   • Bug

     • CameraInfo blueprint wiring could not be executed

       The focused runtime check could not import the actual blueprints because the environment lacks structlog after earlier missing dependencies were resolved. Static inspection indicates both recorder camera_info inputs should be auto-connected to GO2Connection.camera_info by the shared unremapped ("camera_info", CameraInfo) stream key, but the actual build was not completed. Restore the runtime dependency set and run the included blueprint check before relying on recorded calibration.

   • Cause
     • The validation Python environment is incomplete: importing dimos.msgs.sensor_msgs.CameraInfo ultimately imports dimos.utils.logging_config, which raises ModuleNotFoundError: No module named 'structlog'.
   • Fix
     • Install the repository's required runtime dependencies and rerun trex-artifacts/go2-camera-info-wiring-check.py against both blueprints.

   _{Ran code and verified through T-Rex}

_{Reviews (1): Last reviewed commit: "test(go2): remove recorder metadata cove..." | Re-trigger Greptile}