native CompressedImage across all go2 blueprints

[spomichter]

Problem

decided to cahgne connections to output CompressedImage directly, producers pick the encoding, consumers decode only when they need pixels. This PR does that migration for every go2 blueprint. Stacked on #2814 (the CompressedImage type).

What changes

Producer — GO2Connection.color_image: Out[CompressedImage]. jpeg encode happens once at the publish edge (config.image_quality, default 75). frames that are already CompressedImage pass through untouched — which is also why old replay datasets keep working with zero conversion: replay feeds raw Image frames through the same edge and they get encoded like live frames. Go2FleetConnection inherits the change.

Consumers — decode at point of use, and CompressedImage.decode() is memoized (one decode per frame max, cache never pickles):

• go2-only modules flip in place: PerceiveLoopSkill, PersonFollowSkillContainer, TemporalMemory, Go2Memory, Go2Mid360Recorder, VLMAgent, VlmStreamTester, Go2TeleopModule
• modules shared with g1/xarm/desk graphs get port-retype subclasses so those robots keep raw Image until they migrate: CompressedSpatialMemory, CompressedNavigationSkillContainer, CompressedDetection3DModule, CompressedMarkerDetectionStreamModule, HostedTwistTeleopModule (retyped in place — hosted-go2 only)
• note for review: the port retypes need # type: ignore[assignment] (deliberate LSP violation — that IS the wiring mechanism). 5 of them, each commented.

Zero-decode paths (the structural wins over the codec wrapper):

• VLMAgent sends the jpeg wire bytes straight to the model (CompressedImage.agent_encode()) — the old per-query re-encode is gone
• quest teleop _push_jpeg forwards wire bytes to the headset — per-frame re-encode gone
• CameraVideoTrack (hosted teleop webrtc) decodes at recv(), so only frames actually sent to the operator pay a decode
• rerun viewer renders the jpeg via rr.EncodedImage, no pixel decode
• recorders store CompressedImage → new datasets ~39x smaller

Misc: sensor_msgs.CompressedImage added to zenoh latest-wins QoS; bench harness gets --sink decode|bytes and per-nic wire sampling; bench blueprint drops the viewer bridge (it was hijacking any open viewer on the box).

Native e2e benchmark (same protocol as #2831: go2_short replay ~14Hz 720p, 60s x 2 reps)

cell                     frames   fps     lag  cpu%  rss MB  wire/frame
basic  clean  bytes         855  14.3   -52ms     2    1836        71KB
basic  clean  decode        854  14.3   -51ms     2    1893        71KB
go2    clean  bytes         854  14.3   -51ms     4    4732        71KB
go2    clean  decode        855  14.3   -69ms     4    4967        71KB
go2    jetson bytes         853  14.3   -55ms     2    5455        71KB
go2    jetson decode        854  14.3   -53ms     2    5528        71KB
heavy  clean  decode        855  14.3   -52ms     6    5102        71KB
heavy  jetson decode        854  14.3   -39ms     3    5878        71KB

against the #2831 baselines (same box, same dataset):

cell	raw	codec pin	native
go2 / jetson fps	6.5	14.3	14.3
heavy / jetson fps	11.4	14.3	14.3
basic / clean fps	3.3	14.3	14.3
go2 / clean cpu	17%	4%	4%
heavy / jetson cpu	13%	3%	3%
wire per frame	2.76MB	~140KB (synthetic)	71KB (real frames)

native matches the codec pin on delivery and cpu and beats raw everywhere, with no wrapper in any blueprint — the encode lives in the connection where option 5 wants it. real dataset frames compress to 71KB at q75 (39x), better than the synthetic 720p estimate.

pending: network sweep (rate-limit + loss profiles via tc netem) — needs sudo on the bench box; raw/codec network baselines run from 3f2fc05ad where the harness still has --mode raw|codec.

Breaking Changes

• GO2Connection.color_image is now Out[CompressedImage] — anything outside this repo wiring ("color_image", Image) against a go2 graph must retype or decode.
• go2 blueprints referencing the shared perception modules now use the Compressed* subclasses.

How to Test

CI=1 uv run pytest   # full suite green locally (2507 passed)
PYTEST_VERSION=1 uv run python -m dimos.protocol.pubsub.benchmark.tool_replay_bench \
  --blueprint unitree-go2-basic --sink decode --duration 30 --out /tmp/bench

dimos --replay run unitree-go2 shows the camera in rerun via EncodedImage (no pixel decode in the viewer path).

Contributor License Agreement

• I have read and approved the CLA

[codecov]

❌ 1 Tests Failed:

Tests completed	Failed	Passed	Skipped
2847	1	2846	89

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

dimos.e2e_tests.test_dimsim_spatial_memory::test_go_to_the_bed

Flake rate in main: 17.27% (Passed 115 times, Failed 24 times)

Stack Traces | 525s run time

lcm_spy = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x7a72994111c0>
start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x7a7297deb600>
human_input = <function human_input.<locals>.send_human_input at 0x7a7297deb6a0>
dim_sim = <dimos.e2e_tests.dim_sim_client.DimSimClient object at 0x7a729917bb00>
explore_house = <function explore_house.<locals>.explore at 0x7a7297debba0>

    @pytest.mark.self_hosted_large
    def test_go_to_the_bed(lcm_spy, start_blueprint, human_input, dim_sim, explore_house) -> None:
        start_blueprint(
            "run",
            "unitree-go2-agentic",
            simulator="dimsim",
        )
        lcm_spy.save_topic(".../McpClient/on_system_modules/res")
        lcm_spy.wait_for_saved_topic(".../McpClient/on_system_modules/res", timeout=1200.0)
    
        explore_house()
    
        human_input("go to the bed")
    
>       lcm_spy.wait_until_odom_position(-3.567, -1.332, threshold=2, timeout=180)

dim_sim    = <dimos.e2e_tests.dim_sim_client.DimSimClient object at 0x7a729917bb00>
explore_house = <function explore_house.<locals>.explore at 0x7a7297debba0>
human_input = <function human_input.<locals>.send_human_input at 0x7a7297deb6a0>
lcm_spy    = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x7a72994111c0>
start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x7a7297deb600>

dimos/e2e_tests/test_dimsim_spatial_memory.py:32: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
dimos/e2e_tests/lcm_spy.py:167: in wait_until_odom_position
    self.wait_for_message_result(
        predicate  = <function LcmSpy.wait_until_odom_position.<locals>.predicate at 0x7a7297debd80>
        self       = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x7a72994111c0>
        threshold  = 2
        timeout    = 180
        x          = -3.567
        y          = -1.332
dimos/e2e_tests/lcm_spy.py:153: in wait_for_message_result
    wait_until(
        event      = <threading.Event at 0x7a72989b0fe0: unset>
        fail_message = 'Failed to get to position x=-3.567, y=-1.332'
        listener   = <function LcmSpy.wait_for_message_result.<locals>.listener at 0x7a7297debc40>
        predicate  = <function LcmSpy.wait_until_odom_position.<locals>.predicate at 0x7a7297debd80>
        self       = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x7a72994111c0>
        timeout    = 180
        topic      = '/odom#geometry_msgs.PoseStamped'
        type       = <class 'dimos.msgs.geometry_msgs.PoseStamped.PoseStamped'>
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

predicate = <bound method Event.is_set of <threading.Event at 0x7a72989b0fe0: unset>>

    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: Failed to get to position x=-3.567, y=-1.332

deadline   = 290068.968953767
interval   = 0.1
message    = 'Failed to get to position x=-3.567, y=-1.332'
predicate  = <bound method Event.is_set of <threading.Event at 0x7a72989b0fe0: unset>>
timeout    = 180

.../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.}