feat(msgs): typed CompressedImage sensor msg#2814
Conversation
…ges on any transport adds sensor_msgs.CompressedImage wrapping the existing dimos_lcm binding (jpeg/png, ts+frame_id preserved, rr.EncodedImage viz) and a CodecTransport that compresses Image->CompressedImage over any inner transport (lcm, zenoh, shm, webrtc). benchmark test compares raw image vs compressed on lcm+zenoh: 19.5x wire reduction, raw 720p over lcm drops most frames while comressed delivers all
Codecov Report❌ Patch coverage is @@ Coverage Diff @@
## main #2814 +/- ##
==========================================
+ Coverage 72.40% 72.42% +0.02%
==========================================
Files 998 1003 +5
Lines 89235 89665 +430
Branches 8122 8301 +179
==========================================
+ Hits 64610 64941 +331
- Misses 22427 22519 +92
- Partials 2198 2205 +7
Flags with carried forward coverage won't be shown. Click here to find out more.
... and 7 files with indirect coverage changes 🚀 New features to boost your workflow:
|
… compressed image transport go2 basic with CodecTransport on color_image (jpeg q80 on the wire istead of 2.76MB raw frames). readme gets a short compressed-images section with the publish path and how to run the raw-vs-compressed benchmark
the ubuntu-arm runner doesnt have libturbojpeg installed, same optional-dep treatment as test_cli.py / test_all_blueprints.py
…issing root cause of the arm failures: the tests job only apt-installed portaudio, x64 runners had libturbojpeg by accident from the runner image. install it explicitly and turn the silent skip into a CI assert (same treatment as the webrtc aiortc guard from #2048), skip stays for local no-syslib installs
collect rows across the lcm/zenoh params in a module fixture and print a single table (wire/median/max/fps/drops) like tool_benchmark does, plus a note on what the latency number actually measures
installing libturbojpeg on the arm runner unblocked _jpeg_case, which then hit the CI LFS download cap at collection time. treat unavailable data like unavailable turbojpeg (case returns None)
runs a real blueprint in replay mode with BenchSink consumer modules (per-frame arrival log + optional synthetic detector load) and a host sampler (cpu/rss/loopback rate). used for the raw-vs-codec report on unitree-go2; tool_ prefix keeps it out of pytest and the module registry
'codec' already means the memory2 storage codecs (JpegCodec/Lz4Codec) and the generic name promised flexibility the class doesnt have — its hardwired Image<->CompressedImage. new name matches the msg type and the existing JpegLcmTransport/WebRTCVideoTransport naming style
the import-time CI assert broke the self-hosted ros job which collects these files but deselects them by marker. new form: skip locally when the native lib is missing, run (and fail loudly) in CI — same intent, respects deselection like the aiortc guard test does
Team direction is option 5 (connections output CompressedImage directly); the transport wrapper was rejected as public API. Keep it inside tool_replay_bench.py so raw-vs-codec benchmark cells stay reproducible, move its tests alongside, drop the unitree-go2-compressed-image blueprint.
spomichter
changed the title
CompressedCodec is bench-only now: drop format/max_width/decode params and the double-wrap guard the bench never uses; rename its tests to test_replay_bench.py to match. Turbojpeg guard moves to conftest as a skipif_no_turbojpeg marker (skip locally, run-and-fail-loudly in CI).
spomichter
requested review from
Dreamsorcerer,
leshy,
mustafab0 and
paul-nechifor
as code owners
Greptile SummaryThis PR adds a typed compressed-image message and benchmark support. The main changes are:
Confidence Score: 4/5ROS bridge conversion and alpha-bearing PNG decoding can return invalid message data.
dimos/msgs/sensor_msgs/CompressedImage.py; dimos/protocol/pubsub/benchmark/tool_replay_bench.py Important Files Changed
Reviews (1): Last reviewed commit: "Merge branch 'main' into feat/compressed..." | Re-trigger Greptile |
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| @dataclass | ||
| class CompressedImage(Timestamped): | ||
| """Compressed image bytes (JPEG/PNG) — ROS sensor_msgs/CompressedImage.""" | ||
|
|
There was a problem hiding this comment.
This type flattens the ROS header into frame_id and ts, but it is not handled by the ROS bridge's complex-message conversion path. When a sensor_msgs.CompressedImage crosses that bridge, generic field copying cannot rebuild the ROS header or decode it back into the flattened fields, so ROS-backed publishers or subscribers can receive malformed messages or fail conversion.
| if arr.ndim == 2: | ||
| fmt = ImageFormat.GRAY16 if arr.dtype == np.uint16 else ImageFormat.GRAY | ||
| else: | ||
| fmt = ImageFormat.BGR |
There was a problem hiding this comment.
cv2.IMREAD_UNCHANGED returns a four-channel array for a PNG with alpha, but this branch marks every color PNG as BGR. Downstream image operations then see a three-channel format with four-channel data and can skip a needed conversion or fail on the unexpected shape.
| if arr.ndim == 2: | |
| fmt = ImageFormat.GRAY16 if arr.dtype == np.uint16 else ImageFormat.GRAY | |
| else: | |
| fmt = ImageFormat.BGR | |
| if arr.ndim == 2: | |
| fmt = ImageFormat.GRAY16 if arr.dtype == np.uint16 else ImageFormat.GRAY | |
| elif arr.shape[2] == 4: | |
| fmt = ImageFormat.BGRA | |
| else: | |
| fmt = ImageFormat.BGR |
Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!
|
|
||
| @rpc | ||
| def stop(self) -> None: | ||
| self._flush() |
There was a problem hiding this comment.
stop() flushes _buf without _lock, while the LCM receiver thread can append and flush the same buffer under that lock. If a callback is in flight during coordinator shutdown, the two flushes can duplicate, lose, or interleave JSONL records, making the benchmark's delivered-frame output inaccurate.
github-actions
Bot
added
the
ready-to-merge
| try: | ||
| db_path = get_data("go2_short.db") | ||
| except RuntimeError: | ||
| # 84MB LFS pull — refused by the CI LFS guard on standard runners |
There was a problem hiding this comment.
We either need this for the test, in which case it should be self_hosted, or we don't, in which case don't try and fetch it at all.
There was a problem hiding this comment.
I think me and Paul have differing opinions on this. But, I'd use an importorskip and rely on coverage to catch regressions in CI testing. It makes it easier for downstream packagers when testing.
|
|
||
|
|
||
| @dataclass | ||
| class CompressedImage(Timestamped): |
There was a problem hiding this comment.
you invented a message type? you cannot write custom msgs to sensor msgs like this. also likely no need to create a new msg type - see what ros ecosystem uses and respect their message format, I think it's some FoxgloveCompressedImage msg we already have in our types
There was a problem hiding this comment.
I would also recommend to integrate this end-to-end, so add a go2 build that doesn't decompress incoming stream, passes straight into rerun, with some module in the middle able to use image
There was a problem hiding this comment.
ROS has official compressed image type: https://docs.ros2.org/foxy/api/sensor_msgs/msg/CompressedImage.html I believe this impl is mostly aligned, the header field misalignment is exposed by greptile comments below.
There was a problem hiding this comment.
#string format # Specifies the format of the data
# # Acceptable values:
# # jpeg, png
this seems to be only jpeg/png but aparently people abuse and put h264 there since it's string
|
|
||
| from dimos.msgs.sensor_msgs.Image import Image | ||
|
|
||
| CompressionFormat = Literal["jpeg", "png"] |
leshy
left a comment
leshy
left a comment
There was a problem hiding this comment.
This definitely needs h264 support. h265 encoders are almost always in hardware and much cheaper
H264/265 are video codec. I don't think the abstraction fits here? Unless every frame is a key frame but it's very inefficient. |
every frame is a compressed video frame with occasional keyframes (same as any other h264 transport) it's a responsibility of a recevier to reconstruct full images if needed with a small keyframe aware reconstructor discussion on this here - #2831 would be something like also btw if h264 doesn't fit our abstraction - I'd argue the abstraction is wrong since obviously h264 is the correct format to pass video streams in the system (but I think our abstraction is ok) |
4 participants
Problem
dimos has no first-class compressed image type. jpeg support is bolted onto individual transports (
JpegLcmTransport,JpegShmTransport, nothing for zenoh/webrtc),JpegShmTransportloses timestamps on decode, and every new transport needs its own jpeg variant. meanwhile raw images are big — 720p rgb is 2.76MB per frame, ~40MB/s at camera rate. thats fine for shared memory on one host but it kills lcm the moment you leave it: one dropped udp fragment loses the whole frame.CompressedImageis a first-class msg (jpeg/png bytes + timestamp + frame_id, wraps thedimos_lcmbinding that already existed):depth/16-bit images can't go through jpeg — use
format="png"(lossless, covers GRAY16). float DEPTH raises on both.Benchmark report: raw Image vs CompressedImage
(numbers produced with the bench harness in this PR; the codec pin is the bench mechanism, the wins are the type's)
full grid run on this dev box (x86_64, 32 cpus), 6 runs x 10 frames per cell, sequential round-trips. what "latency" measures: one
broadcast()call until the subscriber callback fires, on localhost — so it includes the jpeg encode+decode cpu for the codec path but near-zero wire cost. thats why codec shows higher latency here: raw is two memcpys, codec pays ~5ms encode + ~5ms decode of real cpu. localhost is raw's best case — on a real 1Gbps link the raw frame pays ~22ms of transmit time alone vs <1ms for the jpeg, and over webrtc/wan its not close. what you buy for those ms: 8-40x less wire and the difference between frames arriving or not (see drops).jpeg codec cpu cost (median of 20, ms)
round-trip latency, median ms (LOST = zero frames ever arrived)
round-trip latency, p95 ms
dropped frames (of 60)
lcm raw 480p/720p lost every single frame on this box (untuned kernel udp buffers —
CI=1skips the lcm autoconf sysctl tuning, so this is what a fresh install sees). note the 1080p codec drops too: 315KB @ q75 is already past what untuned lcm handles reliably — thats what thequality/max_widthknobs are for (q50 = 69KB, zero drops).jpeg quality sweep (lcm, 720p)
sustained 14Hz 720p camera feed for 8s (the go2 workload)
this is the headline: publish a go2-style camera feed at 14Hz over lcm and the raw pipeline delivers 0.8 fps. the codec pipeline delivers all 112 frames.
throughput capacity, fps (1 / median latency)
codec capacity is cpu-bound (~1/encode+decode time) and the same on every transport — ~100fps at 720p, 7x a 14Hz camera. pshm raw shows why shm stays the on-host raw escape hatch: 1765fps at 720p, zero encode cost.
end-to-end replay benchmark (full nav stack, jetson-class 4-core profile: raw delivers 6.5 of 14 fps, compressed delivers all of it, with LESS total cpu) is in #2831 with charts.
notes: webrtc isnt in the grid because its loopback provider is an in-memory dict (stack overhead only, no real SCTP — thats P1 of the webrtc backend spec) and the cloudflare path is cred-gated. the vlm/agent blueprints (spatial, temporal-memory, agentic) are the biggest indirect winners — they re-encode jpeg per VLM call today, with option 5 they get wire bytes for free.
follow-ups (not this PR): option 5 migration per #2831 (
GO2Connection->Out[CompressedImage]+ consumers, needs a replay-dataset compat answer since recorded datasets store rawImage), migratingJpegLcmTransport/JpegShmTransportonto CompressedImage, webrtcImageDecimationemitting this type.Breaking Changes
None
How to Test
14 tests incl. real lcm+zenoh round-trips (skipped where native libturbojpeg is missing — and ci.yml now installs it so CI cant skip silently). benchmark alone, prints one grid at the end:
e2e replay bench:
python -m dimos.protocol.pubsub.benchmark.tool_replay_bench --blueprint unitree-go2-basic --mode codec --duration 30 --out /tmp/bench(re-ran post-trim: 473 frames / 33s = 14.3fps, full delivery). full suite + mypy run clean locally.Contributor License Agreement