diff --git a/dimos/message_utils/__init__.py b/dimos/message_utils/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/dimos/message_utils/image.py b/dimos/message_utils/image.py new file mode 100644 index 00000000000..08cfb6296bf --- /dev/null +++ b/dimos/message_utils/image.py @@ -0,0 +1,151 @@ +# Copyright 2026 Dimensional Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Read and convert generated image buffers.""" + +from typing import Any + +import imagecodecs +import numpy as np +from numpy.typing import NDArray + +from dimos.messages.builtin_interfaces.msg import Time +from dimos.messages.sensor_msgs.msg import CompressedImage, Image +from dimos.messages.std_msgs.msg import Header + +_FORMATS = { + "rgb8": ("u1", 3), + "bgr8": ("u1", 3), + "rgba8": ("u1", 4), + "bgra8": ("u1", 4), + "mono8": ("u1", 1), + "mono16": ("u2", 1), + "16UC1": ("u2", 1), + "32FC1": ("f4", 1), + "64FC1": ("f8", 1), + "16SC1": ("i2", 1), + "32FC3": ("f4", 3), +} + + +def _image_from_array( + pixels: NDArray[Any], *, encoding: str, header: Header | None = None +) -> Image: + """Copy an array into a generated image with an explicit pixel encoding.""" + if encoding not in _FORMATS: + raise ValueError(f"unsupported image encoding {encoding!r}") + kind, channels = _FORMATS[encoding] + expected = np.dtype(kind) + if pixels.dtype.kind != expected.kind or pixels.dtype.itemsize != expected.itemsize: + raise ValueError(f"array dtype does not match {encoding}") + if (channels == 1 and pixels.ndim != 2) or ( + channels != 1 and (pixels.ndim != 3 or pixels.shape[2] != channels) + ): + raise ValueError(f"array shape does not match {encoding}") + contiguous = np.array(pixels, copy=True, order="C") + return Image( + header=header if header is not None else Header(stamp=Time(sec=0, nanosec=0), frame_id=""), + width=pixels.shape[1], + height=pixels.shape[0], + encoding=encoding, + is_bigendian=int(not pixels.dtype.isnative if np.little_endian else pixels.dtype.isnative), + step=pixels.shape[1] * channels * pixels.dtype.itemsize, + data=np.asarray(contiguous.view(np.uint8).ravel(), dtype=np.uint8), + ) + + +def image_view(msg: Image) -> NDArray[Any]: + """Borrow a read-only array, retaining the message and respecting row padding/endian.""" + if msg.encoding not in _FORMATS: + raise ValueError(f"unsupported image encoding {msg.encoding!r}") + kind, channels = _FORMATS[msg.encoding] + dtype = np.dtype((">" if msg.is_bigendian else "<") + kind) + row_bytes = msg.width * channels * dtype.itemsize + if msg.step < row_bytes or len(msg.data) != msg.height * msg.step: + raise ValueError("image data length/step does not match its dimensions") + shape = (msg.height, msg.width) if channels == 1 else (msg.height, msg.width, channels) + strides = ( + (msg.step, dtype.itemsize) + if channels == 1 + else (msg.step, channels * dtype.itemsize, dtype.itemsize) + ) + view = np.ndarray(shape, dtype=dtype, buffer=msg.data, strides=strides) + view.setflags(write=False) + return view + + +def image_from_compressed(message: CompressedImage) -> Image: + """Decode JPEG/PNG or the JXL extension into explicitly encoded generated pixels.""" + import cv2 + + format_name = message.format.lower() + if "jxl" in format_name: + jxl_pixels = imagecodecs.jpegxl_decode(bytes(message.data)) + if jxl_pixels.ndim == 2: + encoding = { + np.dtype("float32"): "32FC1", + np.dtype("uint16"): "mono16", + np.dtype("uint8"): "mono8", + }.get(jxl_pixels.dtype) + elif jxl_pixels.ndim == 3 and jxl_pixels.shape[2] in (3, 4): + encoding = "rgb8" if jxl_pixels.shape[2] == 3 else "rgba8" + else: + encoding = None + if encoding is None: + raise ValueError("unsupported decoded JXL layout") + if encoding == "mono16" and format_name.startswith("16uc1;"): + encoding = "16UC1" + return _image_from_array(jxl_pixels, encoding=encoding, header=message.header) + if not any(name in format_name for name in ("jpeg", "jpg", "png")): + raise ValueError(f"unsupported compressed image format {message.format!r}") + pixels = cv2.imdecode(np.frombuffer(bytes(message.data), dtype=np.uint8), cv2.IMREAD_UNCHANGED) + if pixels is None: + raise ValueError("invalid compressed image data") + if pixels.ndim == 2: + encoding = "mono16" if pixels.dtype.itemsize == 2 else "mono8" + elif pixels.ndim == 3 and pixels.shape[2] == 3: + encoding = "bgr8" + elif pixels.ndim == 3 and pixels.shape[2] == 4: + encoding = "bgra8" + else: + raise ValueError("unsupported decoded image layout") + if encoding == "mono16" and format_name.startswith("16uc1;"): + encoding = "16UC1" + return _image_from_array(pixels, encoding=encoding, header=message.header) + + +def image_to_rgb(message: Image) -> NDArray[np.uint8]: + """Copy a display image as RGB8; 16-bit grayscale uses its high byte. + + Alpha is discarded. Floating-point depth needs an explicit visualization + scale and is rejected rather than assigning pixel values to meters. + """ + import cv2 + + pixels = image_view(message) + if message.encoding == "rgb8": + return pixels.copy() + if message.encoding in ("mono16", "16UC1"): + return np.asarray( + cv2.cvtColor((pixels / 256).astype(np.uint8), cv2.COLOR_GRAY2RGB), dtype=np.uint8 + ) + codes = { + "bgr8": cv2.COLOR_BGR2RGB, + "rgba8": cv2.COLOR_RGBA2RGB, + "bgra8": cv2.COLOR_BGRA2RGB, + "mono8": cv2.COLOR_GRAY2RGB, + } + if message.encoding not in codes: + raise ValueError(f"Cannot convert {message.encoding!r} to RGB8") + return np.asarray(cv2.cvtColor(pixels, codes[message.encoding]), dtype=np.uint8) diff --git a/dimos/message_utils/pointcloud.py b/dimos/message_utils/pointcloud.py new file mode 100644 index 00000000000..1e630828eab --- /dev/null +++ b/dimos/message_utils/pointcloud.py @@ -0,0 +1,90 @@ +# Copyright 2026 Dimensional Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Read generated point-cloud layouts and extract positions and colors.""" + +from typing import Any + +import numpy as np +from numpy.typing import NDArray + +from dimos.messages.sensor_msgs.msg import PointCloud2 + +_FIELD_KINDS = {1: "i1", 2: "u1", 3: "i2", 4: "u2", 5: "i4", 6: "u4", 7: "f4", 8: "f8"} + + +def pointcloud_view(message: PointCloud2) -> NDArray[Any]: + """Borrow read-only structured (height, width) points, including row/field padding. + + Fields retain their declared counts and endianness. The returned array keeps + the message's storage alive. Use ``view.copy()`` for independent mutable data. + """ + if message.width and not message.point_step: + raise ValueError("nonempty point cloud requires a positive point_step") + if message.row_step < message.width * message.point_step: + raise ValueError("point cloud row_step is smaller than its row of points") + if len(message.data) != message.height * message.row_step: + raise ValueError("point cloud data length does not match height * row_step") + names: list[str] = [] + formats: list[Any] = [] + offsets: list[int] = [] + for field in message.fields: + if not field.name or field.name in names: + raise ValueError("point cloud field names must be nonempty and unique") + if field.datatype not in _FIELD_KINDS or not field.count: + raise ValueError(f"invalid point cloud datatype/count for {field.name!r}") + dtype = np.dtype((">" if message.is_bigendian else "<") + _FIELD_KINDS[field.datatype]) + if field.offset + dtype.itemsize * field.count > message.point_step: + raise ValueError(f"point cloud field {field.name!r} exceeds point_step") + names.append(field.name) + formats.append(dtype if field.count == 1 else (dtype, (field.count,))) + offsets.append(field.offset) + dtype = np.dtype( + {"names": names, "formats": formats, "offsets": offsets, "itemsize": message.point_step} + ) + view = np.ndarray( + (message.height, message.width), + dtype=dtype, + buffer=message.data, + strides=(message.row_step, message.point_step), + ) + view.setflags(write=False) + return view + + +def pointcloud_xyz(message: PointCloud2) -> NDArray[np.float64]: + """Copy scalar x/y/z fields into an N-by-3 float64 array, preserving point order.""" + points = pointcloud_view(message) + for name in ("x", "y", "z"): + if name not in (points.dtype.names or ()) or points[name].ndim != 2: + raise ValueError(f"point cloud requires a scalar {name!r} field") + return np.stack([points[name].ravel() for name in ("x", "y", "z")], axis=-1).astype(np.float64) + + +def pointcloud_rgb(message: PointCloud2) -> NDArray[np.uint8] | None: + """Copy RGB bytes from the standard packed rgb/rgba field, if present. + + FLOAT32 fields carry color bits, not numeric floating-point color values. + Both representations preserve the declared byte order and row padding. + """ + points = pointcloud_view(message) + names = points.dtype.names or () + name = next((name for name in ("rgb", "rgba") if name in names), None) + if name is None: + return None + field = points[name] + if field.ndim != 2 or field.dtype.kind not in "fu" or field.dtype.itemsize != 4: + raise ValueError("packed point cloud color must be a scalar FLOAT32 or UINT32 field") + packed = field.view(np.dtype(">u4" if message.is_bigendian else "> shift) & 255 for shift in (16, 8, 0)], axis=-1).astype(np.uint8) diff --git a/dimos/visualization/rerun/bridge.py b/dimos/visualization/rerun/bridge.py index 543e3488b2f..819cafa663d 100644 --- a/dimos/visualization/rerun/bridge.py +++ b/dimos/visualization/rerun/bridge.py @@ -17,6 +17,7 @@ from __future__ import annotations from collections.abc import Callable +from contextlib import ExitStack from dataclasses import field import functools from operator import attrgetter @@ -52,6 +53,16 @@ from dimos.core.core import rpc from dimos.core.global_config import global_config from dimos.core.module import Module, ModuleConfig +from dimos.core.transport import PubSubTransport +from dimos.messages.sensor_msgs.msg import ( + CameraInfo as CdrCameraInfo, + CompressedImage as CdrCompressedImage, + Image as CdrImage, +) +from dimos.messages.tf2_msgs.msg import TFMessage as CdrTFMessage +from dimos.msgs.geometry_msgs.Quaternion import Quaternion +from dimos.msgs.geometry_msgs.Transform import Transform +from dimos.msgs.geometry_msgs.Vector3 import Vector3 from dimos.msgs.sensor_msgs.CameraInfo import CameraInfo from dimos.msgs.sensor_msgs.Image import Image from dimos.msgs.tf2_msgs.TFMessage import TfFrameTree, TFMessage @@ -62,6 +73,7 @@ from dimos.protocol.service.lcmservice import autoconf from dimos.utils.generic import get_local_ips from dimos.utils.logging_config import setup_logger +from dimos.visualization.rerun.cdr import camera_pinhole, generated_archetypes, message_frame_id from dimos.visualization.rerun.constants import ( RERUN_ENABLE_WEB, RERUN_GRPC_PORT, @@ -278,7 +290,12 @@ class Config(ModuleConfig): honored. """ - pubsubs: list[SubscribeAllCapable[Any, Any]] = field(default_factory=lambda: [LCM()]) + # Construct the selected backend only when the bridge starts. Creating a + # blueprint or importing a camera module must not open multicast sockets. + pubsubs: list[SubscribeAllCapable[Any, Any]] | None = None + # Explicit subscriptions for the proposal's generated CDR channels. Each + # transport belongs to this bridge; legacy discovery/defaults stay intact. + cdr_transports: dict[str, PubSubTransport[Any]] = field(default_factory=dict) visual_override: dict[Glob | str, VisualOverride] = field(default_factory=dict) static: dict[str, Callable[[Any], Any]] = field(default_factory=dict) @@ -342,7 +359,7 @@ def __init__(self, **kwargs: Any) -> None: self._last_log = {} self._override_cache: dict[str, Callable[[Any], RerunData | None]] = {} self._frame_attached: dict[str, str] = {} - self._camera_infos: dict[str, CameraInfo] = {} + self._camera_infos: dict[str, CameraInfo | CdrCameraInfo] = {} self._image_entities: set[str] = set() self._min_intervals: dict[str, float] = {} self._tf_lock = threading.Lock() @@ -409,7 +426,7 @@ def final_convert(msg: Any) -> RerunData | None: return msg if isinstance(msg, RerunConvertible): return msg.to_rerun() - return None + return generated_archetypes(msg, entity_path) # compose all converters def composed(msg: Any) -> RerunData | None: @@ -456,13 +473,43 @@ def _on_message(self, msg: Any, topic: Any) -> None: return self._last_log[entity_path] = now + if self._tf_tree is not None and isinstance(msg, CdrTFMessage): + # The existing frame-tree layout helper consumes local numerical values. + transforms = [ + Transform( + translation=Vector3( + t.transform.translation.x, + t.transform.translation.y, + t.transform.translation.z, + ), + rotation=Quaternion( + t.transform.rotation.x, + t.transform.rotation.y, + t.transform.rotation.z, + t.transform.rotation.w, + ), + frame_id=t.header.frame_id, + child_frame_id=t.child_frame_id, + ts=t.header.stamp.sec + t.header.stamp.nanosec / 1_000_000_000, + ) + for t in msg.transforms + ] + with self._tf_lock: + for entry in generated_archetypes(msg, entity_path) or []: + rr.log(entry[0], entry[1]) + self._tf_tree.update(transforms) + return + if self._tf_tree is not None and isinstance(msg, TFMessage): with self._tf_lock: for path, archetype in msg.to_rerun(self._tf_tree): rr.log(path, archetype) return - if isinstance(msg, CameraInfo) and entity_path not in self.config.visual_override: + if ( + isinstance(msg, (CameraInfo, CdrCameraInfo)) + and entity_path not in self.config.visual_override + ): self._log_camera_info(entity_path, msg) return @@ -471,16 +518,26 @@ def _on_message(self, msg: Any, topic: Any) -> None: if not rerun_data: return + is_image = isinstance(msg, (Image, CdrImage, CdrCompressedImage)) + if is_image: + self._image_entities.add(entity_path) + # TFMessage for example returns list of (entity_path, archetype) tuples if is_rerun_multi(rerun_data): for entry in rerun_data: rr.log(entry[0], entry[1], static=isinstance(entry, RerunEntry) and entry.static) # Pin the base entity to the message frame once, so child entities # follow it the way a single archetype does below. - frame_id = getattr(msg, "frame_id", None) + frame_id = message_frame_id(msg) if frame_id and self._frame_attached.get(entity_path) != frame_id: - rr.log(entity_path, rr.Transform3D(parent_frame=f"tf#/{frame_id}")) + if not any( + entry[0] == entity_path and isinstance(entry[1], rr.Transform3D) + for entry in rerun_data + ): + rr.log(entity_path, rr.Transform3D(parent_frame=f"tf#/{frame_id}")) self._frame_attached[entity_path] = frame_id + if is_image and frame_id in self._camera_infos: + rr.log(entity_path, self._camera_pinhole(self._camera_infos[frame_id])) else: rr.log(entity_path, cast("Archetype", rerun_data)) if isinstance(msg, Image): @@ -488,14 +545,18 @@ def _on_message(self, msg: Any, topic: Any) -> None: # if source msg carries a frame_id, attach the entity to that TF frame # should skip if archetype is a Transform3D if not isinstance(rerun_data, rr.Transform3D): - frame_id = getattr(msg, "frame_id", None) + frame_id = message_frame_id(msg) if frame_id and self._frame_attached.get(entity_path) != frame_id: rr.log(entity_path, rr.Transform3D(parent_frame=f"tf#/{frame_id}")) self._frame_attached[entity_path] = frame_id - if isinstance(msg, Image) and frame_id in self._camera_infos: - rr.log(entity_path, self._camera_infos[frame_id].to_rerun_pinhole()) + if is_image and frame_id in self._camera_infos: + rr.log(entity_path, self._camera_pinhole(self._camera_infos[frame_id])) + + @staticmethod + def _camera_pinhole(info: CameraInfo | CdrCameraInfo) -> Archetype: + return camera_pinhole(info) if isinstance(info, CdrCameraInfo) else info.to_rerun_pinhole() - def _log_camera_info(self, entity_path: str, info: CameraInfo) -> None: + def _log_camera_info(self, entity_path: str, info: CameraInfo | CdrCameraInfo) -> None: """A CameraInfo is the pinhole of every image in its optical frame. Rerun draws the frustum only when the Pinhole sits on the image entity, @@ -504,13 +565,14 @@ def _log_camera_info(self, entity_path: str, info: CameraInfo) -> None: """ import rerun as rr - if not info.frame_id: - rr.log(entity_path, info.to_rerun_pinhole()) + frame = message_frame_id(info) + if not frame: + rr.log(entity_path, self._camera_pinhole(info)) return - self._camera_infos[info.frame_id] = info + self._camera_infos[frame] = info for image_path, frame_id in self._frame_attached.items(): - if frame_id == info.frame_id and image_path in self._image_entities: - rr.log(image_path, info.to_rerun_pinhole()) + if frame_id == frame and image_path in self._image_entities: + rr.log(image_path, self._camera_pinhole(info)) @rpc def start(self) -> None: @@ -596,6 +658,32 @@ def start(self) -> None: self.register_disposable(Disposable(pubsub.stop)) # type: ignore[union-attr] self._log_static() + for topic, transport in self.config.cdr_transports.items(): + self._subscribe_cdr(topic, transport) + + def _subscribe_cdr(self, topic: str, transport: PubSubTransport[Any]) -> None: + cleanup = ExitStack() + cleanup.callback(transport.stop) + try: + unsubscribe = transport.subscribe(functools.partial(self._on_cdr_message, topic=topic)) + cleanup.callback(unsubscribe) + except BaseException: + cleanup.close() + raise + self.register_disposable(Disposable(cleanup.close)) + + def _on_cdr_message(self, msg: Any, *, topic: str) -> None: + import rerun as rr + + stamp = getattr(getattr(msg, "header", None), "stamp", None) + if stamp is not None: + nanos = int(stamp.sec) * 1_000_000_000 + int(stamp.nanosec) + rr.set_time("source", timestamp=np.datetime64(nanos, "ns")) + try: + self._on_message(msg, "/" + topic.strip("/")) + finally: + if stamp is not None: + rr.disable_timeline("source") def _subscribe(self, pubsub: SubscribeAllCapable[Any, Any]) -> Callable[[], None]: """Subscribe to the named topics, or to everything when none are named. diff --git a/dimos/visualization/rerun/cdr.py b/dimos/visualization/rerun/cdr.py new file mode 100644 index 00000000000..920de372993 --- /dev/null +++ b/dimos/visualization/rerun/cdr.py @@ -0,0 +1,254 @@ +# Copyright 2026 Dimensional Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Viewer adapters live outside the generated plain value classes.""" + +from __future__ import annotations + +from dataclasses import asdict +import json +from typing import TYPE_CHECKING, Any + +import numpy as np + +from dimos.message_utils.image import ( + image_from_compressed, + image_to_rgb, + image_view, +) +from dimos.message_utils.pointcloud import pointcloud_rgb, pointcloud_xyz +from dimos.messages.dimos_msgs.msg import EntityMarkers +from dimos.messages.geometry_msgs.msg import PointStamped, PoseStamped +from dimos.messages.nav_msgs.msg import Odometry, Path +from dimos.messages.sensor_msgs.msg import CameraInfo, CompressedImage, Image, PointCloud2 +from dimos.messages.tf2_msgs.msg import TFMessage + +if TYPE_CHECKING: + import rerun as rr + + from dimos.visualization.rerun.bridge import RerunMulti + + +def message_frame_id(message: Any) -> str | None: + header = getattr(message, "header", None) + frame = ( + getattr(header, "frame_id", None) + if header is not None + else getattr(message, "frame_id", None) + ) + return frame if isinstance(frame, str) else None + + +def camera_pinhole(info: CameraInfo, optical_frame: str | None = None) -> rr.Pinhole: + import rerun as rr + + frame = info.header.frame_id if optical_frame is None else optical_frame + return rr.Pinhole( + focal_length=[info.k[0], info.k[4]], + principal_point=[info.k[2], info.k[5]], + width=info.width, + height=info.height, + image_plane_distance=1.0, + parent_frame=f"tf#/{frame}" if frame else None, + ) + + +def image_archetype(message: Image | CompressedImage) -> rr.Image | rr.DepthImage | rr.EncodedImage: + import rerun as rr + + if isinstance(message, CompressedImage): + name = message.format.lower() + if "jxl" in name or ("png" in name and name.startswith("16uc1;")): + return image_archetype(image_from_compressed(message)) + if "jpeg" in name or "jpg" in name: + return rr.EncodedImage(contents=bytes(message.data), media_type="image/jpeg") + if "png" in name: + return rr.EncodedImage(contents=bytes(message.data), media_type="image/png") + raise ValueError(f"unsupported compressed image format {message.format!r}") + if message.encoding in ("16UC1", "32FC1", "mono16"): + pixels = image_view(message) + pixels = np.ascontiguousarray(pixels, dtype=pixels.dtype.newbyteorder("=")) + if message.encoding != "mono16": + return rr.DepthImage(pixels, meter=1000.0 if message.encoding == "16UC1" else 1.0) + return rr.Image( + bytes=pixels.tobytes(), + width=message.width, + height=message.height, + datatype=np.uint16, + color_model="L", + ) + return rr.Image( + bytes=image_to_rgb(message).tobytes(), + width=message.width, + height=message.height, + datatype=np.uint8, + color_model="RGB", + ) + + +def cloud_archetype( + message: PointCloud2, + *, + voxel_size: float = 0.05, + mode: str = "points", + bottom_cutoff: float | None = None, + height_colors: bool = False, + color: tuple[int, int, int] | None = None, +) -> rr.Points3D | rr.Boxes3D: + """Render plain cloud values, keeping color/geometry filters aligned. + + Existing bridge defaults use screen-space dots. Offline map tools explicitly + select world-space spheres/boxes and the existing height-based turbo colors. + """ + from matplotlib import colormaps + import rerun as rr + + points, colors = pointcloud_xyz(message), pointcloud_rgb(message) + keep = np.isfinite(points).all(axis=1) + if bottom_cutoff is not None: + keep &= points[:, 2] >= bottom_cutoff + points = points[keep] + if colors is not None: + colors = colors[keep] + elif height_colors and len(points): + z = points[:, 2] + levels = ((z - z.min()) / (z.max() - z.min() + 1e-8) * 255).astype(np.uint8) + colors = (colormaps["turbo"](levels)[:, :3] * 255).astype(np.uint8) + if color is not None: + colors = np.asarray(color, dtype=np.uint8) + if mode == "boxes": + return rr.Boxes3D(centers=points, half_sizes=[voxel_size / 2] * 3, colors=colors) + return rr.Points3D( + points, + colors=colors, + radii=rr.components.Radius.ui_points(2.0) if mode == "points" else voxel_size / 2, + ) + + +def entity_points(message: EntityMarkers) -> rr.Points3D: + """Render generated world entities with the existing labels, colors and radii.""" + import rerun as rr + + colors = { + "person": (255, 100, 100, 255), + "object": (100, 255, 100, 255), + "location": (100, 100, 255, 255), + } + return rr.Points3D( + positions=[[m.position.x, m.position.y, m.position.z] for m in message.markers], + labels=[f"{m.entity_id}: {m.label[:40]}" for m in message.markers], + colors=[colors.get(m.entity_type, (200, 200, 200, 255)) for m in message.markers], + radii=[0.15] * len(message.markers), + ) + + +def path_archetype( + message: Path, + *, + color: tuple[int, int, int] = (0, 255, 128), + z_offset: float = 0.5, + radii: float = 0.05, +) -> rr.LineStrips3D: + import rerun as rr + + if not message.poses: + return rr.LineStrips3D([]) + points = [ + [pose.pose.position.x, pose.pose.position.y, pose.pose.position.z + z_offset] + for pose in message.poses + ] + return rr.LineStrips3D([points], colors=[color], radii=radii) + + +def generated_archetypes(message: Any, entity: str) -> RerunMulti | None: + """Render stamped geometry and its exact fields; other domains remain explicit.""" + import rerun as rr + + if type(message) in (Image, CompressedImage, PointCloud2): + archetype = ( + cloud_archetype(message) if type(message) is PointCloud2 else image_archetype(message) + ) + entries: RerunMulti = [(entity, archetype)] + frame = message_frame_id(message) + if frame: + entries.append((entity, rr.Transform3D(parent_frame=f"tf#/{frame}"))) + return entries + if type(message) is CameraInfo: + entries = [(entity, camera_pinhole(message))] + elif type(message) is TFMessage: + entries = [] + for value in message.transforms: + t, q = value.transform.translation, value.transform.rotation + entries.append( + ( + "tf#/" + rr.escape_entity_path_part(value.child_frame_id), + rr.Transform3D( + translation=[t.x, t.y, t.z], + quaternion=rr.Quaternion(xyzw=[q.x, q.y, q.z, q.w]), + parent_frame="tf#/" + value.header.frame_id, + child_frame="tf#/" + value.child_frame_id, + ), + ) + ) + elif type(message) is Path: + entries = [(entity, path_archetype(message))] + if message.header.frame_id: + entries.append((entity, rr.Transform3D(parent_frame=f"tf#/{message.header.frame_id}"))) + elif type(message) is EntityMarkers: + entries = [(entity, entity_points(message))] + if message.header.frame_id: + entries.append((entity, rr.Transform3D(parent_frame=f"tf#/{message.header.frame_id}"))) + elif type(message) in (PoseStamped, Odometry): + pose = message.pose.pose if type(message) is Odometry else message.pose + point, rotation = pose.position, pose.orientation + entries = [ + ( + entity, + rr.Transform3D( + translation=[point.x, point.y, point.z], + quaternion=rr.Quaternion(xyzw=[rotation.x, rotation.y, rotation.z, rotation.w]), + parent_frame=f"tf#/{message.header.frame_id}" + if message.header.frame_id + else None, + child_frame=f"tf#/{message.child_frame_id}" + if type(message) is Odometry and message.child_frame_id + else None, + ), + ), + (entity, rr.TransformAxes3D(0.35)), + ] + elif type(message) is PointStamped: + point = message.point + entries = [(entity, rr.Points3D([[point.x, point.y, point.z]]))] + if message.header.frame_id: + entries.append((entity, rr.Transform3D(parent_frame=f"tf#/{message.header.frame_id}"))) + else: + return None + fields = asdict( + message, + dict_factory=lambda items: {key: value for key, value in items if key != "__msgtype__"}, + ) + entries.append( + (entity + "/message", rr.TextDocument(json.dumps(fields, indent=2, default=_json_value))) + ) + return entries + + +def _json_value(value: Any) -> Any: + """Represent generated numerical arrays without changing their source values.""" + if isinstance(value, np.ndarray): + return value.tolist() + if isinstance(value, np.generic): + return value.item() + raise TypeError(f"Cannot render {type(value).__name__} as JSON") diff --git a/dimos/visualization/rerun/test_cdr.py b/dimos/visualization/rerun/test_cdr.py new file mode 100644 index 00000000000..2f087fb714c --- /dev/null +++ b/dimos/visualization/rerun/test_cdr.py @@ -0,0 +1,339 @@ +# Copyright 2026 Dimensional Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Generated values stay plain while the viewer owns rendering and subscriptions.""" + +import json +from unittest.mock import Mock + +from dimos_cdrgen.registry import decode, encode +import numpy as np +import pytest +import rerun as rr + +from dimos.core.global_config import GlobalConfig +from dimos.core.transport import ZenohTransport +from dimos.messages.builtin_interfaces.msg import Time +from dimos.messages.dimos_msgs.msg import EntityMarker, EntityMarkers +from dimos.messages.geometry_msgs.msg import ( + Point, + PointStamped, + PoseWithCovariance, + Quaternion, + Transform, + TransformStamped, + Twist, + TwistWithCovariance, + Vector3, +) +from dimos.messages.nav_msgs.msg import Odometry +from dimos.messages.sensor_msgs.msg import ( + CameraInfo, + Image, + PointCloud2, + PointField, + RegionOfInterest, +) +from dimos.messages.std_msgs.msg import Header +from dimos.messages.tf2_msgs.msg import TFMessage +from dimos.protocol.rpc.zenohrpc import ZenohRPC +from dimos.visualization.rerun.bridge import Config, RerunBridgeModule, _resolve_pubsubs +from dimos.visualization.rerun.cdr import ( + cloud_archetype, + generated_archetypes, + image_archetype, +) +from dimos.visualization.vis_module import vis_module +from dimos.web.relay_bridge.tests.cdr_fixtures import build_messages + + +def image_from_array(pixels, *, encoding, header=None): + return Image( + header=header or Header(stamp=Time(sec=0, nanosec=0), frame_id=""), + width=pixels.shape[1], + height=pixels.shape[0], + encoding=encoding, + is_bigendian=int(pixels.dtype.byteorder == ">"), + step=pixels.strides[0], + data=np.frombuffer(pixels.tobytes(), dtype=np.uint8), + ) + + +def camera_info_from_intrinsics(fx, fy, cx, cy, width, height, *, header): + return CameraInfo( + distortion_model="plumb_bob", + d=np.zeros(5, dtype=np.float64), + r=np.eye(3, dtype=np.float64).ravel(), + p=np.zeros(12, dtype=np.float64), + binning_x=0, + binning_y=0, + roi=RegionOfInterest(x_offset=0, y_offset=0, height=0, width=0, do_rectify=False), + header=header, + width=width, + height=height, + k=np.asarray([fx, 0, cx, 0, fy, cy, 0, 0, 1], dtype=np.float64), + ) + + +def pointcloud_from_xyz_rgb(points, colors, *, header): + data = np.empty(len(points), dtype=[("x", " None: + message = build_messages()[0][1] + entries = generated_archetypes(message, "world/pose") + assert entries is not None + assert [path for path, _ in entries] == ["world/pose", "world/pose", "world/pose/message"] + transform, axes, text = [archetype for _, archetype in entries] + np.testing.assert_allclose( + transform.translation.as_arrow_array().to_pylist(), [[1.5, -2.5, 0.25]] + ) + np.testing.assert_allclose( + transform.quaternion.as_arrow_array().to_pylist(), [[0, 0, 0.6, 0.8]] + ) + assert transform.parent_frame.as_arrow_array().to_pylist() == ["tf#/map"] + assert isinstance(axes, rr.TransformAxes3D) + fields = json.loads(text.text.as_arrow_array().to_pylist()[0]) + assert fields["header"] == { + "stamp": {"sec": 1757400000, "nanosec": 500000123}, + "frame_id": "map", + } + assert "__msgtype__" not in fields + assert not hasattr(message, "to_rerun") + + +def test_point_frame_and_unknown_values() -> None: + pose = build_messages()[0][1] + msg = PointStamped(header=pose.header, point=pose.pose.position) + entries = generated_archetypes(msg, "world/point") + assert entries is not None + assert isinstance(entries[0][1], rr.Points3D) + assert entries[1][1].parent_frame.as_arrow_array().to_pylist() == ["tf#/map"] + assert generated_archetypes({"x": 1}, "world/no") is None + + +@pytest.mark.parametrize("nanoseconds", [0, 1757400000500000123]) +def test_source_time_is_scoped_and_reset_after_render_failure(bridge, mocker, nanoseconds) -> None: + msg = build_messages()[0][1] + msg.header.stamp = Time(sec=nanoseconds // 1_000_000_000, nanosec=nanoseconds % 1_000_000_000) + render = mocker.patch.object(bridge, "_on_message", side_effect=ValueError("bad renderer")) + set_time = mocker.patch.object(rr, "set_time") + clear_time = mocker.patch.object(rr, "disable_timeline") + with pytest.raises(ValueError, match="bad renderer"): + bridge._on_cdr_message(msg, topic="/pose") + render.assert_called_once_with(msg, "/pose") + set_time.assert_called_once_with("source", timestamp=np.datetime64(nanoseconds, "ns")) + clear_time.assert_called_once_with("source") + + +def test_subscription_cleanup_order_and_failed_start(bridge) -> None: + operations: list[str] = [] + transport = Mock(spec=ZenohTransport) + transport.subscribe.return_value = lambda: operations.append("unsubscribe") + transport.stop.side_effect = lambda: operations.append("stop") + bridge._subscribe_cdr("pose", transport) + bridge.stop() + assert operations == ["unsubscribe", "stop"] + failed = Mock(spec=ZenohTransport) + failed.subscribe.side_effect = RuntimeError("native start failed") + with pytest.raises(RuntimeError, match="native start failed"): + bridge._subscribe_cdr("pose", failed) + failed.stop.assert_called_once_with() + + +def test_bridge_default_renderer_handles_generated_pose(bridge, mocker) -> None: + log = mocker.patch.object(rr, "log") + bridge._on_message(build_messages()[0][1], "/pose") + assert [call.args[0] for call in log.call_args_list] == [ + "world/pose", + "world/pose", + "world/pose/message", + ] + assert isinstance(log.call_args_list[0].args[1], rr.Transform3D) + + +@pytest.mark.parametrize("shape", [(1, 2), (2, 1)]) +def test_single_row_or_column_rgb_keeps_dimensions_and_color_bytes(shape) -> None: + pixels = np.array([[[255, 0, 0], [0, 255, 0]]], dtype=np.uint8).reshape(*shape, 3) + image = decode(encode(image_from_array(pixels, encoding="rgb8")), Image) + archetype = image_archetype(image) + assert archetype.format.as_arrow_array().to_pylist()[0]["height"] == shape[0] + assert archetype.format.as_arrow_array().to_pylist()[0]["width"] == shape[1] + assert archetype.buffer.as_arrow_array().to_pylist() == [[255, 0, 0, 0, 255, 0]] + + +def test_big_endian_depth_renderer_preserves_values_and_units() -> None: + pixels = np.array([[1, 256, 65535]], dtype=">u2") + archetype = image_archetype(image_from_array(pixels, encoding="16UC1")) + assert isinstance(archetype, rr.DepthImage) + assert archetype.meter.as_arrow_array().to_pylist() == [1000.0] + raw = bytes(archetype.buffer.as_arrow_array().to_pylist()[0]) + np.testing.assert_array_equal(np.frombuffer(raw, dtype=np.uint16), pixels.ravel()) + + +@pytest.mark.parametrize("camera_first", [False, True]) +def test_calibration_attaches_in_both_arrival_orders(bridge, mocker, camera_first) -> None: + header = build_messages()[0][1].header + image = image_from_array(np.zeros((1, 2, 3), dtype=np.uint8), encoding="rgb8", header=header) + calibration = camera_info_from_intrinsics(10, 20, 1, 0.5, 2, 1, header=header) + log = mocker.patch.object(rr, "log") + samples = [(calibration, "/calibration"), (image, "/image")] + for message, topic in samples if camera_first else reversed(samples): + bridge._on_message(message, topic) + pinholes = [call for call in log.call_args_list if isinstance(call.args[1], rr.Pinhole)] + assert len(pinholes) == 1 + assert pinholes[0].args[0] == "world/image" + assert pinholes[0].args[1].parent_frame.as_arrow_array().to_pylist() == ["tf#/map"] + + +def test_cloud_renderer_filters_nonfinite_positions_with_matching_colors() -> None: + header = build_messages()[0][1].header + cloud = pointcloud_from_xyz_rgb( + np.array([[1, 2, 3], [float("nan"), 0, 0], [4, 5, 6]]), + np.array([[255, 0, 0], [0, 255, 0], [0, 0, 255]], dtype=np.uint8), + header=header, + ) + archetype = cloud_archetype(cloud) + assert archetype.positions.as_arrow_array().to_pylist() == [[1, 2, 3], [4, 5, 6]] + assert archetype.colors.as_arrow_array().to_pylist() == [0xFF0000FF, 0x0000FFFF] + + +@pytest.mark.parametrize("backend", ["zenoh", "lcm"]) +def test_blueprint_does_not_construct_pubsubs_and_resolution_keeps_backend(mocker, backend) -> None: + lcm = mocker.patch("dimos.visualization.rerun.bridge.LCM") + zenoh = mocker.patch("dimos.visualization.rerun.bridge.Zenoh") + config = Config(g=GlobalConfig(transport=backend)) + assert config.pubsubs is None + vis_module("rerun") + lcm.assert_not_called() + zenoh.assert_not_called() + resolved = _resolve_pubsubs(config) + expected = zenoh if backend == "zenoh" else lcm + expected.assert_called_once_with() + assert resolved == [expected.return_value] + + +def test_generated_entity_marker_adapter_keeps_labels_colors_frame_and_exact_fields(): + header = Header(frame_id="world", stamp=Time(sec=1700000000, nanosec=123456789)) + message = EntityMarkers( + header=header, + markers=[ + EntityMarker( + entity_id="E1", + label="person", + entity_type="person", + position=Point(x=1, y=2, z=0.3), + ) + ], + ) + message = decode(encode(message), EntityMarkers) + entries = generated_archetypes(message, "memory/entities") + assert entries is not None + points, frame, fields = [value for _, value in entries] + np.testing.assert_allclose(points.positions.as_arrow_array().to_pylist(), [[1, 2, 0.3]]) + assert points.labels.as_arrow_array().to_pylist() == ["E1: person"] + assert points.colors.as_arrow_array().to_pylist() == [0xFF6464FF] + assert frame.parent_frame.as_arrow_array().to_pylist() == ["tf#/world"] + assert ( + json.loads(fields.text.as_arrow_array().to_pylist()[0])["header"]["stamp"]["nanosec"] + == 123456789 + ) + + +@pytest.mark.parametrize("mode", ["points", "boxes", "spheres"]) +def test_fixed_display_color_overrides_sensor_rgb_after_invalid_points_filter(mode): + cloud = pointcloud_from_xyz_rgb( + np.array([[1, 2, 3], [np.nan, 0, 0]], dtype=np.float32), + np.array([[255, 0, 0], [0, 255, 0]], dtype=np.uint8), + header=Header(stamp=Time(sec=0, nanosec=0), frame_id="world"), + ) + before = encode(cloud) + value = cloud_archetype(cloud, mode=mode, color=(76, 220, 41)) + positions = value.centers if mode == "boxes" else value.positions + assert positions.as_arrow_array().to_pylist() == [[1, 2, 3]] + assert value.colors.as_arrow_array().to_pylist() == [0x4CDC29FF] + assert encode(cloud) == before + + +def test_generated_odometry_exposes_pose_axes_and_full_covariance_fields(): + pose = build_messages()[0][1] + value = Odometry( + pose.header, + "base", + PoseWithCovariance(pose.pose, np.zeros(36)), + TwistWithCovariance(Twist(Vector3(0.0, 0.0, 0.0), Vector3(0.0, 0.0, 0.0)), np.zeros(36)), + ) + value.pose.covariance[0] = 0.25 + entries = generated_archetypes(value, "world/odometry") + assert entries is not None + assert isinstance(entries[0][1], rr.Transform3D) + assert isinstance(entries[1][1], rr.TransformAxes3D) + assert entries[0][1].translation.as_arrow_array().to_pylist() == [[1.5, -2.5, 0.25]] + fields = json.loads(entries[-1][1].text.as_arrow_array().to_pylist()[0]) + assert fields["pose"]["covariance"][0] == 0.25 + assert fields["child_frame_id"] == "base" + assert entries[0][1].child_frame.as_arrow_array().to_pylist() == ["tf#/base"] + assert fields["header"]["stamp"]["nanosec"] == 500000123 + assert value.pose.covariance[0] == 0.25 + + +def test_generated_tf_preserves_frame_tree_layout_and_transform_values(bridge, mocker): + bridge.config.tf_axes = 0.35 + bridge._tf_tree = bridge._new_tf_tree() + log = mocker.patch.object(rr, "log") + value = TFMessage( + [ + TransformStamped( + Header(Time(0, 1), "world"), + "camera", + Transform(Vector3(1.0, 2.0, 3.0), Quaternion(0.0, 0.0, 0.0, 1.0)), + ) + ] + ) + bridge._on_message(value, "tf") + assert bridge._tf_tree.placements() == { + "world": "world/tf/world", + "camera": "world/tf/world/camera", + } + transform = next(call.args[1] for call in log.call_args_list if call.args[0] == "tf#/camera") + assert transform.translation.as_arrow_array().to_pylist() == [[1.0, 2.0, 3.0]] + assert transform.child_frame.as_arrow_array().to_pylist() == ["tf#/camera"] + assert value.transforms[0].header.stamp == Time(0, 1) diff --git a/dimos/visualization/rerun/test_cdr_e2e.py b/dimos/visualization/rerun/test_cdr_e2e.py new file mode 100644 index 00000000000..bafeef18d47 --- /dev/null +++ b/dimos/visualization/rerun/test_cdr_e2e.py @@ -0,0 +1,102 @@ +# Copyright 2026 Dimensional Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Run the reviewer example and inspect its live bridge and native MCAP importer.""" + +from pathlib import Path +import subprocess +import sys + +import pytest + + +def test_live_bridge_and_native_mcap_import_cover_scene( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + # The pinned Rerun CLI disables analytics when CI is present. This changes + # only the child environment, not the user's persistent viewer settings. + monkeypatch.setenv("CI", "1") + monkeypatch.delenv("FORCE_RERUN_ANALYTICS", raising=False) + example = Path(__file__).resolve().parents[3] / "examples/message-viewer/demo_cdr_rerun.py" + output = tmp_path / "pose.rrd" + command = [sys.executable, "-I", str(example), str(output)] + subprocess.run(command, check=True, capture_output=True, text=True, timeout=30) + contents = output.read_bytes() + assert contents.startswith(b"RRF") + rerun = Path(sys.executable).parent / "rerun" + result = subprocess.run( + [str(rerun), "rrd", "print", "-vvv", str(output)], + check=True, + capture_output=True, + text=True, + timeout=30, + ) + for marker in ( + "TransformAxes3D:axis_length", + "[[1.5, -2.5, 0.25]]", + "[[0.0, 0.0, 0.6, 0.8]]", + "tf#/map", + "tf#/camera", + "2023-11-14T22:13:20.123456789", + '"nanosec": 123456789', + "Exact message values", + "Points3D:positions", + "Points3D:colors", + "Boxes3D:centers", + "Image:buffer", + "Pinhole:image_from_camera", + "DepthImage:meter", + "EncodedImage:blob", + "world/path", + "world/odometry", + "world/entities", + "demo person", + "world/legacy_pose", + "world/late_image", + "world/cloud_boxes", + "world/cloud_height", + "Generated RGB pixels", + ): + assert marker in result.stdout, marker + recording = output.with_suffix(".mcap") + assert recording.read_bytes().startswith(b"\x89MCAP") + native = output.with_name(output.stem + "-native.rrd") + subprocess.run( + [str(rerun), "mcap", "convert", str(recording), "-o", str(native)], + check=True, + capture_output=True, + text=True, + timeout=30, + ) + imported = subprocess.run( + [str(rerun), "rrd", "print", "-vvv", str(native)], + check=True, + capture_output=True, + text=True, + timeout=30, + ) + for marker in ( + "/pose", + "/tf", + "Image:buffer", + "Points3D:positions", + "Pinhole:image_from_camera", + "InstancePoses3D", + "2023-11-14T22:13:20.123456789", + ): + assert marker in imported.stdout, marker + repeated = subprocess.run(command, capture_output=True, text=True, timeout=30) + assert repeated.returncode != 0 + assert "FileExistsError" in repeated.stderr + assert output.read_bytes() == contents diff --git a/dimos/visualization/rerun/test_cdr_entities.py b/dimos/visualization/rerun/test_cdr_entities.py new file mode 100644 index 00000000000..9cfd9c3676d --- /dev/null +++ b/dimos/visualization/rerun/test_cdr_entities.py @@ -0,0 +1,48 @@ +# Copyright 2026 Dimensional Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Tests for refactored temporal memory components.""" + +from dimos_cdrgen.registry import decode as cdr_decode, encode as cdr_encode +import rerun as rr + +from dimos.messages.builtin_interfaces.msg import Time +from dimos.messages.dimos_msgs.msg import EntityMarker, EntityMarkers +from dimos.messages.geometry_msgs.msg import Point +from dimos.messages.std_msgs.msg import Header +from dimos.visualization.rerun.cdr import entity_points + + +def test_markers_to_rerun() -> None: + markers = EntityMarkers( + markers=[ + EntityMarker( + entity_id="E1", + label="person walking", + entity_type="person", + position=Point(x=1.0, y=2.0, z=0.3), + ), + EntityMarker( + entity_id="E2", + label="wooden table", + entity_type="object", + position=Point(x=3.0, y=4.0, z=0.3), + ), + ], + header=Header(stamp=Time(sec=0, nanosec=0), frame_id=""), + ) + archetype = entity_points(cdr_decode(cdr_encode(markers), EntityMarkers)) + # Should return rr.Points3D + + assert isinstance(archetype, rr.Points3D) diff --git a/dimos/visualization/rerun/test_cdr_image_values.py b/dimos/visualization/rerun/test_cdr_image_values.py new file mode 100644 index 00000000000..3e56d4bac7c --- /dev/null +++ b/dimos/visualization/rerun/test_cdr_image_values.py @@ -0,0 +1,57 @@ +# Copyright 2025-2026 Dimensional Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import imagecodecs +import numpy as np +import pytest +import rerun as rr + +from dimos.messages.builtin_interfaces.msg import Time +from dimos.messages.sensor_msgs.msg import CompressedImage +from dimos.messages.std_msgs.msg import Header +from dimos.visualization.rerun.cdr import image_archetype + + +def test_png_color_is_rendered_as_encoded_image(): + pixels = np.array([[[255, 0, 0], [0, 255, 0]]], dtype=np.uint8) + message = CompressedImage( + header=Header(stamp=Time(sec=0, nanosec=0), frame_id="camera"), + format="png", + data=np.frombuffer(imagecodecs.png_encode(pixels), dtype=np.uint8), + ) + archetype = image_archetype(message) + assert isinstance(archetype, rr.EncodedImage) + assert bytes(archetype.blob.as_arrow_array().to_pylist()[0]) == bytes(message.data) + + +@pytest.mark.parametrize("codec", ["png", "jxl"]) +def test_compressed_uint16_depth_keeps_values_and_units(codec): + pixels = np.arange(80, dtype=np.uint16).reshape(10, 8) + encoded = ( + imagecodecs.png_encode(pixels) + if codec == "png" + else imagecodecs.jpegxl_encode(pixels, lossless=True) + ) + message = CompressedImage( + header=Header(frame_id="depth", stamp=Time(sec=0, nanosec=0)), + format=f"16UC1; {codec}", + data=np.frombuffer(encoded, dtype=np.uint8), + ) + archetype = image_archetype(message) + assert isinstance(archetype, rr.DepthImage) + assert archetype.meter.as_arrow_array().to_pylist() == [1000.0] + np.testing.assert_array_equal( + np.frombuffer(bytes(archetype.buffer.as_arrow_array().to_pylist()[0]), dtype=np.uint16), + pixels.ravel(), + ) diff --git a/dimos/web/cdr_codec.py b/dimos/web/cdr_codec.py new file mode 100644 index 00000000000..459565fe887 --- /dev/null +++ b/dimos/web/cdr_codec.py @@ -0,0 +1,63 @@ +# Copyright 2026 Dimensional Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Generated CDR payloads and self-contained ROS2 schemas for web channels.""" + +from collections.abc import Mapping +import re +from typing import Any + +from dimos_cdrgen.registry import encode, message_types, schema + +CDR_V1_SUFFIX = ".cdr.v1" +_TYPE_NAME = re.compile(r"^[A-Za-z][A-Za-z0-9_]*/msg/[A-Za-z][A-Za-z0-9_]*$") + + +def cdr_type_name(message_type: type[Any]) -> str | None: + name = getattr(message_type, "__msgtype__", None) + if ( + isinstance(name, str) + and _TYPE_NAME.fullmatch(name) + and message_types().get(name) is message_type + ): + return name + return None + + +def export_schema(message_type: type[Any]) -> dict[str, str]: + name = cdr_type_name(message_type) + if name is None: + raise ValueError(f"{message_type.__qualname__} has no generated CDR schema") + return {"type": name, "definition": schema(name)} + + +def check_cdr_params(params: Mapping[str, Any]) -> None: + schema = params.get("cdr") + if not ( + isinstance(schema, Mapping) + and isinstance(schema.get("type"), str) + and _TYPE_NAME.fullmatch(schema["type"]) + and isinstance(schema.get("definition"), str) + ): + raise ValueError( + "*.cdr.v1 channels need params['cdr'] = {type, definition}; cockpit() fills it in" + ) + + +def encode_cdr_v1(msg: Any, params: Mapping[str, Any]) -> bytes: + if cdr_type_name(type(msg)) != params["cdr"]["type"]: + raise ValueError( + f"message type does not match declared channel type {params['cdr']['type']}" + ) + return encode(msg) diff --git a/dimos/web/cockpit.py b/dimos/web/cockpit.py index 22dbe908c44..9049f922b80 100644 --- a/dimos/web/cockpit.py +++ b/dimos/web/cockpit.py @@ -51,7 +51,8 @@ from dimos.msgs.geometry_msgs.PointStamped import PointStamped from dimos.msgs.nav_msgs.Path import Path from dimos.msgs.sensor_msgs.PointCloud2 import PointCloud2 -from dimos.web.codecs import is_generic_lcm_encoding +from dimos.web.cdr_codec import export_schema as export_cdr_schema +from dimos.web.codecs import is_generic_cdr_encoding, is_generic_lcm_encoding from dimos.web.lcm_codec import default_encoding, export_schema, schema_class_for from dimos.web.relay_bridge.manifest import ( MANIFEST_VERSION, @@ -168,7 +169,8 @@ class Channel: JSON-shaped types and dataclasses (rx) / JSON scalars, lists and dicts (tx decode), or `.lcm.v1` for rx DimOS messages with a dimos_lcm schema (the frame is `lcm_encode()`, the schema rides - params["lcm"]). None picks the default: `.lcm.v1` for rx + params["lcm"]), or explicit `/msg/.cdr.v1` for installed + generated values (complete ROS2 schemas ride params["cdr"]). None picks the default: `.lcm.v1` for rx DimOS messages, "json.v1" otherwise. A dir="tx" channel is a generic browser publish input: it must declare @@ -278,6 +280,10 @@ def _request_of(channel: Channel) -> ChannelRequest: # Deep plain copy: the manifest and specs must not alias the (frozen) # authoring record's nested values. params = _thaw_params(channel.params or {}) + if is_generic_cdr_encoding(channel.encoding): + if "cdr" in params: + raise ValueError("params key 'cdr' is reserved for the CDR schema") + params["cdr"] = export_cdr_schema(channel.message_type) if is_generic_lcm_encoding(channel.encoding): if "lcm" in params: raise ValueError("params key 'lcm' is reserved for the LCM schema") diff --git a/dimos/web/codecs.py b/dimos/web/codecs.py index e0129908846..fee05c93c96 100644 --- a/dimos/web/codecs.py +++ b/dimos/web/codecs.py @@ -35,10 +35,11 @@ def decode_goal(value: dict[str, Any], context: PublishContext) -> PoseStamped: Encoders take the message (plus, optionally, the channel's params mapping) and return `bytes`, an `EncodedPayload` when the frame needs header meta, or None -to skip the sample. Two codecs apply without registration: json.v1 for +to skip the sample. Codecs apply without registration: json.v1 for JSON-shaped message types, and `.lcm.v1` (dimos/web/lcm_codec.py) for DimOS messages with a dimos_lcm schema (the frame is `msg.lcm_encode()`, -the schema rides the channel's params["lcm"]). Anything else potentially +the schema rides the channel's params["lcm"]), and explicit `/msg/.cdr.v1` +for installed generated CDR values with complete params["cdr"] schemas. Anything else potentially large (images, arrays, bytes) needs an explicit @web_encoder. """ @@ -52,6 +53,13 @@ def decode_goal(value: dict[str, Any], context: PublishContext) -> PoseStamped: import types from typing import Any, TypeVar, Union, get_args, get_origin, get_type_hints +from dimos.web.cdr_codec import ( + CDR_V1_SUFFIX, + cdr_type_name, + check_cdr_params, + encode_cdr_v1, + export_schema as export_cdr_schema, +) from dimos.web.lcm_codec import ( LCM_V1_SUFFIX, check_lcm_params, @@ -421,6 +429,11 @@ def resolve_decoder(encoding: str, message_type: type[Any]) -> DecoderDef: ) +def is_generic_cdr_encoding(encoding: str) -> bool: + """True for a generic CDR encoder unless a registered encoder owns the id.""" + return encoding.endswith(CDR_V1_SUFFIX) and encoding not in _encoders + + def is_generic_lcm_encoding(encoding: str) -> bool: """True for a *.lcm.v1 id the generic LCM codec serves (a registered @web_encoder with that id keeps its own params and encoder).""" @@ -446,6 +459,16 @@ def resolve_encoder(encoding: str, message_type: type[Any]) -> EncoderDef: "and qualified name" ) return definition + if encoding.endswith(CDR_V1_SUFFIX): + if encoding != f"{cdr_type_name(message_type)}{CDR_V1_SUFFIX}": + raise ValueError( + f"encoding {encoding!r} does not match the installed CDR type " + f"{message_type.__qualname__}" + ) + export_cdr_schema(message_type) + return EncoderDef( + encoding, message_type, encode_cdr_v1, takes_params=True, check_params=check_cdr_params + ) if encoding.endswith(LCM_V1_SUFFIX): if encoding != f"{lcm_type_name(message_type)}{LCM_V1_SUFFIX}": raise ValueError( diff --git a/dimos/web/relay_bridge/module_test_support.py b/dimos/web/relay_bridge/module_test_support.py index 7bb849106c3..51fdf38cbfc 100644 --- a/dimos/web/relay_bridge/module_test_support.py +++ b/dimos/web/relay_bridge/module_test_support.py @@ -27,9 +27,11 @@ import threading import time from typing import Any +from unittest.mock import Mock import pytest +from dimos.protocol.rpc.zenohrpc import ZenohRPC from dimos.web.relay_bridge import relay_bridge_module from dimos.web.relay_bridge.protocol import PROTOCOL_VERSION, DataFrame, IceServer, Msg, RtcTrack from dimos.web.relay_bridge.relay_bridge_module import RelayBridgeModule @@ -261,6 +263,10 @@ async def fake_fetch(base_url: str, **kwargs: Any) -> RelayInfo: monkeypatch.setattr(relay_bridge_module, "fetch_relay_info", fake_fetch) monkeypatch.setattr(RelayClient, "connect", fake_connect) + # Module construction starts RPC before the relay itself. Keep this + # no-network harness off native sockets as well as off WebTransport. + monkeypatch.setattr(ZenohRPC, "start", Mock()) + monkeypatch.setattr(ZenohRPC, "serve_module_rpc", Mock()) def make_bridge( diff --git a/dimos/web/relay_bridge/test_relay_bridge_authoring.py b/dimos/web/relay_bridge/test_relay_bridge_authoring.py index 8b59103dc61..2be80cec126 100644 --- a/dimos/web/relay_bridge/test_relay_bridge_authoring.py +++ b/dimos/web/relay_bridge/test_relay_bridge_authoring.py @@ -27,6 +27,7 @@ import struct from typing import Any +from dimos_cdrgen.registry import encode as encode_cdr from langchain_core.messages import AIMessage import numpy as np import pytest @@ -36,6 +37,7 @@ from dimos.core.module import Module, ModuleConfig from dimos.core.resource_monitor.stats import ProcessStats, WorkerStats from dimos.core.stream import In, Out +from dimos.messages.geometry_msgs.msg import PoseStamped as GeneratedPoseStamped from dimos.msgs.geometry_msgs.PointStamped import PointStamped from dimos.msgs.geometry_msgs.Pose import Pose from dimos.msgs.geometry_msgs.PoseStamped import PoseStamped @@ -68,6 +70,7 @@ Subs, ) from dimos.web.relay_bridge.relay_bridge_module import RelayBridgeModule, with_relay_bridge +from dimos.web.relay_bridge.tests.cdr_fixtures import build_messages # Composition helpers live at module level: under PEP 563 (`from __future__ @@ -212,6 +215,41 @@ def test_authored_specs_drive_generated_port(monkeypatch) -> None: stop_module(module) +def test_generated_cdr_is_encoded_only_for_subscribed_viewers(monkeypatch) -> None: + blueprint = cockpit( + channels=[ + Channel( + "generated_pose", + GeneratedPoseStamped, + encoding="geometry_msgs/msg/PoseStamped.cdr.v1", + max_hz=1000.0, + ) + ] + ) + module, clients = start_authored(monkeypatch, blueprint, wire=("generated_pose",)) + try: + transport = transport_of(module, "generated_pose") + msg = build_messages()[0][1] + transport.publish(msg) + flush_loop(module) + assert clients[0].frames == [] + push(module, clients[0], Subs(chs=["generated_pose"], n=1)) + assert wait_until(lambda: transport.subscribers) + transport.publish(msg) + assert wait_until(lambda: clients[0].frames) + channel, payload, delivery, meta = clients[0].frames[0] + assert (channel, payload, delivery, meta) == ( + "generated_pose", + encode_cdr(msg), + "reliable", + None, + ) + push(module, clients[0], Subs(chs=[], n=2)) + assert wait_until(lambda: not transport.subscribers) + finally: + stop_module(module) + + def test_two_jpeg_channels_use_independent_quality(monkeypatch) -> None: # The second camera the old single-int _jpeg_quality could not express. blueprint = cockpit( diff --git a/dimos/web/relay_bridge/tests/cdr_fixtures.py b/dimos/web/relay_bridge/tests/cdr_fixtures.py new file mode 100644 index 00000000000..b0719e5bcc7 --- /dev/null +++ b/dimos/web/relay_bridge/tests/cdr_fixtures.py @@ -0,0 +1,139 @@ +# Copyright 2026 Dimensional Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Regenerate browser fixtures with: python -m dimos.web.relay_bridge.tests.cdr_fixtures.""" + +import base64 +import json +from typing import Any + +from dimos_cdrgen.definitions import Definitions +from dimos_cdrgen.registry import encode as cdr_encode +import numpy as np + +from dimos.messages.builtin_interfaces.msg import Time +from dimos.messages.dimos_msgs.msg import LineSegment3D, LineSegments3D +from dimos.messages.geometry_msgs.msg import Point, Pose, PoseStamped, Quaternion, Vector3 +from dimos.messages.sensor_msgs.msg import Image, Imu, JointState +from dimos.messages.std_msgs.msg import Empty, Header, Int64, String +from dimos.web.cdr_codec import CDR_V1_SUFFIX, export_schema +from dimos.web.relay_bridge.locate import find_web_dir + + +def build_messages() -> list[tuple[str, Any]]: + header = Header(stamp=Time(sec=1757400000, nanosec=500000123), frame_id="map") + return [ + ( + "pose_stamped", + PoseStamped( + header=header, + pose=Pose( + position=Point(x=1.5, y=-2.5, z=0.25), + orientation=Quaternion(z=0.6, w=0.8, x=0.0, y=0.0), + ), + ), + ), + ( + "custom_segments", + LineSegments3D( + header=header, + segments=[ + LineSegment3D( + start=Point(x=1, y=0.0, z=0.0), end=Point(y=2, x=0.0, z=0.0), weight=4 + ) + ], + ), + ), + ( + "image", + Image( + header=header, + width=2, + height=1, + encoding="rgb8", + step=6, + data=np.array([0, 1, 2, 253, 254, 255], dtype=np.uint8), + is_bigendian=0, + ), + ), + ( + "imu", + Imu( + header=header, + orientation_covariance=np.asarray(range(9), dtype=np.float64), + orientation=Quaternion(x=0.0, y=0.0, z=0.0, w=1.0), + angular_velocity=Vector3(x=0.0, y=0.0, z=0.0), + angular_velocity_covariance=np.zeros(9, dtype=np.float64), + linear_acceleration=Vector3(x=0.0, y=0.0, z=0.0), + linear_acceleration_covariance=np.zeros(9, dtype=np.float64), + ), + ), + ( + "joints", + JointState( + header=header, + name=["joint_a", "joint_b"], + position=np.array([1.5, -2.5], dtype=np.float64), + velocity=np.array([], dtype=np.float64), + effort=np.array([], dtype=np.float64), + ), + ), + ("int64", Int64(data=-9007199254740993)), + ("empty", Empty()), + ("string", String(data="robot 🦾 café")), + ] + + +def build_vectors() -> list[dict[str, Any]]: + definitions = {message.name: message for message in Definitions([], installed=True).resolve()} + + def walk(msg: Any) -> dict[str, Any]: + result: dict[str, Any] = {} + for field in definitions[msg.__msgtype__].fields: + value = getattr(msg, field.name) + convert = ( + walk + if field.type.nested + else ( + str + if field.type.name in ("int64", "uint64") + else lambda x: x.item() if isinstance(x, np.generic) else x + ) + ) + result[field.name] = ( + [convert(item) for item in value] if field.type.is_array else convert(value) + ) + return result + + return [ + { + "name": name, + "encoding": msg.__msgtype__ + CDR_V1_SUFFIX, + "schema": export_schema(type(msg)), + "value": walk(msg), + "payload_b64": base64.b64encode(cdr_encode(msg)).decode(), + "big_endian_b64": base64.b64encode(cdr_encode(msg, little_endian=False)).decode(), + } + for name, msg in build_messages() + ] + + +def main() -> None: + path = find_web_dir() / "sdk" / "tests" / "fixtures" / "cdr_frames.json" + path.write_text(json.dumps({"vectors": build_vectors()}, indent=2, ensure_ascii=False) + "\n") + print(path) + + +if __name__ == "__main__": + main() diff --git a/dimos/web/test_cdr_codec.py b/dimos/web/test_cdr_codec.py new file mode 100644 index 00000000000..9297de60208 --- /dev/null +++ b/dimos/web/test_cdr_codec.py @@ -0,0 +1,100 @@ +# Copyright 2026 Dimensional Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Web CDR schemas and bytes come from the installed owner, not another generator.""" + +import base64 +import json +import pickle +import subprocess +import sys +from typing import Any + +from dimos_cdrgen.registry import decode, encode +import pytest + +from dimos.messages.geometry_msgs.msg import Point, PoseStamped +from dimos.web.cdr_codec import check_cdr_params, encode_cdr_v1, export_schema +from dimos.web.cockpit import Channel, cockpit +from dimos.web.codecs import resolve_encoder +from dimos.web.relay_bridge.locate import find_web_dir +from dimos.web.relay_bridge.tests.cdr_fixtures import build_messages, build_vectors + + +def test_cdr_channel_embeds_schema_and_preserves_owned_encoder() -> None: + channel = Channel( + "pose", PoseStamped, encoding="geometry_msgs/msg/PoseStamped.cdr.v1", params={"note": "x"} + ) + blueprint = cockpit(channels=[channel]) + atom = blueprint.blueprints[0] + wire = atom.kwargs["manifest"]["channels"][0] + assert wire["params"] == {"note": "x", "cdr": export_schema(PoseStamped)} + assert "MSG: std_msgs/Header" in wire["params"]["cdr"]["definition"] + assert "MSG: builtin_interfaces/Time" in wire["params"]["cdr"]["definition"] + assert dict(channel.params) == {"note": "x"} + spec = pickle.loads(pickle.dumps(blueprint)).blueprints[0].kwargs["channels"][0] + msg = build_messages()[0][1] + assert spec.encoder is encode_cdr_v1 + assert spec.encoder(msg, spec.params) == encode(msg) + assert next(stream for stream in atom.streams if stream.name == "pose").type is PoseStamped + + +def test_cdr_channel_rejects_foreign_type_and_reserved_schema() -> None: + with pytest.raises(ValueError, match="does not match"): + resolve_encoder("geometry_msgs/msg/Point.cdr.v1", PoseStamped) + with pytest.raises(ValueError, match="reserved"): + cockpit( + channels=[ + Channel( + "pose", + PoseStamped, + encoding="geometry_msgs/msg/PoseStamped.cdr.v1", + params={"cdr": {}}, + ) + ] + ) + with pytest.raises(ValueError, match="declared channel type"): + encode_cdr_v1(Point(x=0, y=0, z=0), {"cdr": export_schema(PoseStamped)}) + with pytest.raises(ValueError, match="no generated CDR schema"): + export_schema(dict) + + +@pytest.mark.parametrize("params", [{}, {"cdr": {}}, {"cdr": {"type": "p.T", "definition": ""}}]) +def test_missing_schema_is_rejected(params: dict[str, Any]) -> None: + with pytest.raises(ValueError, match="params"): + check_cdr_params(params) + + +def test_browser_vectors_are_actual_generated_messages_in_both_byte_orders() -> None: + vectors = json.loads((find_web_dir() / "sdk/tests/fixtures/cdr_frames.json").read_text())[ + "vectors" + ] + assert vectors == build_vectors() + for (_, msg), vector in zip(build_messages(), vectors, strict=True): + for key, little in [("payload_b64", True), ("big_endian_b64", False)]: + payload = base64.b64decode(vector[key]) + restored = decode(payload, type(msg)) + assert type(restored) is type(msg) + assert encode(restored, little_endian=little) == payload + + +def test_codec_import_has_no_generated_or_numpy_side_effects() -> None: + subprocess.run( + [ + sys.executable, + "-c", + "import sys; import dimos.web.cdr_codec; assert 'numpy' not in sys.modules; assert 'dimos.messages' not in sys.modules", + ], + check=True, + ) diff --git a/docs/usage/visualization.md b/docs/usage/visualization.md index 4b3fe350685..9a98cc605b3 100644 --- a/docs/usage/visualization.md +++ b/docs/usage/visualization.md @@ -147,3 +147,57 @@ When a `RerunBridgeModule` is already part of your blueprint, you typically don' Rerun on `dev` is **module-driven**: modules decide what to log, and `Blueprint.build()` sets up the shared viewer + default layout. To use a custom websocket port, set `--rerunwebsocketserver.port`; the bridge in the same blueprint takes the viewer URL from that server. `--rerunbridgemodule.websocket-url` points a bridge at a websocket server elsewhere. Set `host` on each module to override its bind address; both default to the shared `listen_host`. + +## Generated CDR viewer demo (WIP) + +The CDR viewer is under development. This example runs the real Rerun bridge +with local Zenoh transports. It needs no robot or ROS installation. + +From the repository root: + +```sh skip +DIMOS_ALLOW_MISSING_COCKPIT=1 uv sync --locked --extra visualization --extra web +uv run --extra visualization --extra web python -I examples/message-viewer/demo_cdr_rerun.py cdr-viewer.rrd +uv run --extra visualization rerun cdr-viewer.rrd +uv run --extra visualization rerun cdr-viewer.mcap +``` + +Use a new output filename for each run. The example creates the live RRD and +`cdr-viewer.mcap`. Both destinations must be new. The build flag permits +installation without prebuilt browser assets; message generation still runs. + +The live recording uses `RerunBridgeModule.start()`, its typed subscriptions and +rendering callbacks. A receipt signal waits for each message before shutdown. +The MCAP contains the source scene encoded as CDR with complete ROS2 schemas. +It is example input for playback, not a capture by the native recorder. + +Open the MCAP directly with Rerun's native importer. To save its imported data: + +```sh skip +uv run --extra visualization rerun mcap convert cdr-viewer.mcap -o cdr-viewer-native.rrd +``` + +Rerun 0.32 converts supported standard ROS2 types into visualizations. Other +messages, including this scene's Path and custom EntityMarkers, are available +as reflected fields; they do not automatically acquire the live bridge's path +or labeled-entity display. No dimOS Python renderer is used for this import. +Existing `dimos mem rerun` behavior for legacy stores, including `--seconds` +and `--root`, is unchanged. Use the native Rerun command for CDR MCAP files. + +| PR capability | Demo scene | What to check | +| --- | --- | --- | +| TF and source metadata | `tf`, `pose` | Map/camera frames, pose axes at `[1.5, -2.5, 0.25]`, `nanosec: 123456789` in the text tab | +| Geometry adapters | `point`, `odometry`, `path`, `entities` | Point, three-waypoint path, odometry axes, labeled person and object | +| Camera calibration | `image`, `late_image` | Pinhole on each image, whether calibration arrives first or last | +| Sensor buffers | `image`, `depth`, `compressed`, `compressed_depth` | Four RGB pixels, big-endian uint16 depth, PNG image and JXL depth in the sensor tab | +| Cloud rendering | `cloud`, `cloud_boxes`, `cloud_height` | RGB points, boxes and height-colored spheres | +| Legacy rendering | `legacy_pose` | Existing pose rendering alongside generated messages | +| Native MCAP import | `cdr-viewer.mcap` | Standard images, points, calibration and TF; reflected fields for types without a native visualization | + +The example anchors its `world` TF frame to the viewer root. Business producers +must supply their own TF tree. It does not demonstrate robot graph migration or +automatic CDR type discovery. The existing integration test runs this example, +checks the live RRD, then converts the MCAP with the official CLI and checks its imported components. + +For the live browser SDK and manifest path, run the +[Web bridge demo](/docs/web/bridge.md#generated-cdr-messages-wip). diff --git a/docs/web/bridge.md b/docs/web/bridge.md index 651b513e5bf..41b68e161d2 100644 --- a/docs/web/bridge.md +++ b/docs/web/bridge.md @@ -40,6 +40,83 @@ The relay flags are `dimos run` options. Most of the rest are fields of the brid ## Channels +### Generated CDR messages (WIP) + +An installed generated type can be sent as CDR with its complete ROS2 definition +in the manifest. Declare the encoding explicitly in this proposal: + +```python +from dimos.messages.geometry_msgs.msg import PoseStamped +from dimos.web.cockpit import Channel, cockpit + +viewer = cockpit(channels=[Channel( + "measured_pose", PoseStamped, + encoding="geometry_msgs/msg/PoseStamped.cdr.v1", +)]) +``` + +Compose `viewer` with the module producing `measured_pose: Out[PoseStamped]`. +Generated ports use the shared CDR codec on the selected LCM or Zenoh transport. The manifest embeds PoseStamped, Header and Time definitions from +the installed message owner; the browser needs neither ROS nor another generated +message package. Registered custom web encoders continue to take precedence. +Existing LCM/JSON channels and default transport selection are unchanged here. + +The SDK registry resolves the channel schema once and decodes its CDR frames: + +```ts +import { createDecoderRegistry } from "@dimos/sdk"; + +const decode = createDecoderRegistry().resolve(channel); +const value = decode?.(payload, frameHeader).value; +``` + +`channel` is the adopted manifest record, `payload` the frame's `Uint8Array` and +`frameHeader` its SDK header. Invalid schemas are unsupported; malformed CDR +frames fail decoding. The raw cockpit table previews small messages. Messages +with variable-length arrays require a visible panel subscription so a preview +does not pull a large image/cloud buffer. A decoder alone does not add a typed +camera, map or control panel. + +To run a real generated-message source, cockpit bridge and local relay, from +the repository root in terminal 1: + +```sh skip +DIMOS_ALLOW_MISSING_COCKPIT=1 uv sync --locked --extra visualization --extra web +uv run --extra web python -I examples/message-viewer/demo_cdr_browser.py +``` + +In terminal 2: + +```sh skip +cd web/sdk +deno task fixture --host 127.0.0.1 +``` + +Open `http://127.0.0.1:5174/demo_cdr.html?live=1`. Wait for `connected`, +`Received 8/8 channels`, increasing frame counts and zero decode errors. +Select each channel. Check its fields and expand its embedded ROS2 definition: + +- `pose_stamped`: position `[1.5, -2.5, 0.25]` and `nanosec: 500000123`. +- `custom_segments`: custom nested message fields. +- `image`: two near-black/near-white pixels, with bytes `[0, 1, 2, 253, 254, 255]`. +- `imu`, `joints`: fixed and variable arrays. +- `int64`: `-9007199254740993n`, with no JavaScript number rounding. +- `empty`, `string`: an empty message and Unicode text. + +This path uses actual generated Python messages, typed Zenoh transport, +`cockpit()` manifest creation, relay framing, SDK subscriptions and CDR decoding. +The page explicitly subscribes to all demo channels, including arrays. It does +not add domain-specific cockpit panels or browser-to-robot CDR publishing. +Stop terminal 1 with Ctrl-C. Use `--port` to select a different relay port and +open the exact URL printed by the program. + +Without `?live=1`, the same page uses fixed packets to inspect little- and +big-endian decoding. Generate `web/sdk/tests/fixtures/cdr_frames.json` with +`python -m dimos.web.relay_bridge.tests.cdr_fixtures`. SDK tests check both byte +orders; the fixture does not validate a live connection. + +### Existing channels + A channel is one stream crossing the wire in one direction with one encoding. `rx` channels flow from the robot to the browser, `tx` channels from the browser to the robot. The bridge has four built-in ports: | Channel | Type | Encoding | Delivery | diff --git a/examples/message-viewer/demo_cdr_browser.py b/examples/message-viewer/demo_cdr_browser.py new file mode 100644 index 00000000000..417f30cc060 --- /dev/null +++ b/examples/message-viewer/demo_cdr_browser.py @@ -0,0 +1,183 @@ +# Copyright 2026 Dimensional Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Publish generated messages through the real cockpit bridge and local relay. + +Keep publishing for the browser page until Ctrl-C. +""" + +from __future__ import annotations + +import argparse +from contextlib import ExitStack +import threading +from typing import Any + +import numpy as np + +from dimos.core.transport import ZenohTransport +from dimos.messages.builtin_interfaces.msg import Time +from dimos.messages.dimos_msgs.msg import LineSegment3D, LineSegments3D +from dimos.messages.geometry_msgs.msg import Point, Pose, PoseStamped, Quaternion, Vector3 +from dimos.messages.sensor_msgs.msg import Image, Imu, JointState +from dimos.messages.std_msgs.msg import Empty, Header, Int64, String +from dimos.protocol.pubsub.impl.zenohpubsub import Topic +from dimos.protocol.rpc.zenohrpc import ZenohRPC +from dimos.web.cockpit import Channel, cockpit +from dimos.web.relay_bridge.config import RelayBridgeConfig + + +def build_messages() -> list[tuple[str, Any]]: + header = Header(stamp=Time(sec=1757400000, nanosec=500000123), frame_id="map") + return [ + ( + "pose_stamped", + PoseStamped( + header=header, + pose=Pose( + position=Point(x=1.5, y=-2.5, z=0.25), + orientation=Quaternion(z=0.6, w=0.8, x=0.0, y=0.0), + ), + ), + ), + ( + "custom_segments", + LineSegments3D( + header=header, + segments=[ + LineSegment3D( + start=Point(x=1, y=0.0, z=0.0), end=Point(y=2, x=0.0, z=0.0), weight=4 + ) + ], + ), + ), + ( + "image", + Image( + header=header, + width=2, + height=1, + encoding="rgb8", + step=6, + data=np.array([0, 1, 2, 253, 254, 255], dtype=np.uint8), + is_bigendian=0, + ), + ), + ( + "imu", + Imu( + header=header, + orientation_covariance=np.asarray(range(9), dtype=np.float64), + orientation=Quaternion(x=0.0, y=0.0, z=0.0, w=1.0), + angular_velocity=Vector3(x=0.0, y=0.0, z=0.0), + angular_velocity_covariance=np.zeros(9, dtype=np.float64), + linear_acceleration=Vector3(x=0.0, y=0.0, z=0.0), + linear_acceleration_covariance=np.zeros(9, dtype=np.float64), + ), + ), + ( + "joints", + JointState( + header=header, + name=["joint_a", "joint_b"], + position=np.array([1.5, -2.5], dtype=np.float64), + velocity=np.array([], dtype=np.float64), + effort=np.array([], dtype=np.float64), + ), + ), + ("int64", Int64(data=-9007199254740993)), + ("empty", Empty()), + ("string", String(data="robot 🦾 café")), + ] + + +class LoopbackRPC(ZenohRPC): + def __init__(self, **kwargs: Any) -> None: + super().__init__( + mode="peer", + listen=["tcp/127.0.0.1:0"], + scouting=False, + multicast=False, + gossip=False, + **kwargs, + ) + + +def run(port: int) -> None: + messages = dict(build_messages()) + channels = [ + Channel(name, type(value), encoding=value.__msgtype__ + ".cdr.v1", max_hz=5) + for name, value in messages.items() + ] + (atom,) = cockpit(channels=channels).blueprints + config = dict( + mode="peer", listen=["tcp/127.0.0.1:0"], scouting=False, multicast=False, gossip=False + ) + stopped = threading.Event() + with ExitStack() as cleanup: + module = atom.module( + **atom.kwargs, + robot_id="cdr-demo", + robot_name="CDR demo", + rpc_transport=LoopbackRPC, + local_port=port, + open_browser=False, + rtc=False, + ) + cleanup.callback(module.stop) + publishers = {} + for name, value in messages.items(): + receiver = ZenohTransport(Topic(name, type(value)), **config) + cleanup.callback(receiver.stop) + module.inputs[name].transport = receiver + publishers[name] = ZenohTransport(Topic(name, type(value)), **config) + cleanup.callback(publishers[name].stop) + module.start() + + def publish() -> None: + while not stopped.is_set(): + for name, value in messages.items(): + publishers[name].broadcast(None, value) + stopped.wait(0.2) + + thread = threading.Thread(target=publish, name="cdr-demo-publisher", daemon=True) + thread.start() + + def stop_publisher() -> None: + stopped.set() + thread.join(timeout=5) + + cleanup.callback(stop_publisher) + print("From web/sdk run: deno task fixture --host 127.0.0.1", flush=True) + print( + f"Open http://127.0.0.1:5174/demo_cdr.html?live=1&relay={module._url}", + flush=True, + ) + try: + stopped.wait() + except KeyboardInterrupt: + pass + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--port", type=int, default=RelayBridgeConfig.model_fields["local_port"].default + ) + args = parser.parse_args() + run(args.port) + + +if __name__ == "__main__": + main() diff --git a/examples/message-viewer/demo_cdr_rerun.py b/examples/message-viewer/demo_cdr_rerun.py new file mode 100644 index 00000000000..db8c576200d --- /dev/null +++ b/examples/message-viewer/demo_cdr_rerun.py @@ -0,0 +1,400 @@ +# Copyright 2026 Dimensional Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Demonstrate live Rerun bridge rendering and native CDR MCAP viewing without a robot.""" + +from __future__ import annotations + +import argparse +from contextlib import ExitStack +from copy import deepcopy +from functools import partial +from pathlib import Path +import socket +import threading +from typing import Any + +from dimos_cdrgen.registry import encode, schema +import imagecodecs +from mcap.writer import Writer +import numpy as np +import rerun as rr +import rerun.blueprint as rrb + +from dimos.core.transport import ZenohTransport +from dimos.messages.builtin_interfaces.msg import Time +from dimos.messages.dimos_msgs.msg import EntityMarker, EntityMarkers +from dimos.messages.geometry_msgs.msg import ( + Point, + PointStamped, + Pose, + PoseStamped, + PoseWithCovariance, + Quaternion, + Transform, + TransformStamped, + Twist, + TwistWithCovariance, + Vector3, +) +from dimos.messages.nav_msgs.msg import Odometry, Path as MessagePath +from dimos.messages.sensor_msgs.msg import ( + CameraInfo, + CompressedImage, + Image, + PointCloud2, + PointField, + RegionOfInterest, +) +from dimos.messages.std_msgs.msg import Header +from dimos.messages.tf2_msgs.msg import TFMessage +from dimos.msgs.geometry_msgs.PoseStamped import PoseStamped as LegacyPoseStamped +from dimos.protocol.pubsub.impl.zenohpubsub import Topic +from dimos.protocol.pubsub.patterns import Glob +from dimos.protocol.rpc.zenohrpc import ZenohRPC +from dimos.visualization.rerun.bridge import RerunBridgeModule +from dimos.visualization.rerun.cdr import cloud_archetype + + +def scene_messages() -> dict[str, Any]: + message = PoseStamped( + header=Header(stamp=Time(sec=1700000000, nanosec=123456789), frame_id="map"), + pose=Pose( + position=Point(x=1.5, y=-2.5, z=0.25), orientation=Quaternion(x=0, y=0, z=0.6, w=0.8) + ), + ) + pixels = np.array([[[255, 0, 0], [0, 255, 0]], [[0, 0, 255], [255, 255, 255]]], dtype=np.uint8) + image = Image( + header=message.header, + width=2, + height=2, + encoding="rgb8", + is_bigendian=0, + step=6, + data=pixels.ravel(), + ) + calibration = CameraInfo( + distortion_model="plumb_bob", + d=np.zeros(5, dtype=np.float64), + r=np.eye(3, dtype=np.float64).ravel(), + p=np.zeros(12, dtype=np.float64), + binning_x=0, + binning_y=0, + roi=RegionOfInterest(x_offset=0, y_offset=0, height=0, width=0, do_rectify=False), + header=message.header, + width=2, + height=2, + k=np.array([2, 0, 0.5, 0, 2, 0.5, 0, 0, 1], dtype=np.float64), + ) + points = np.array( + [ + (-0.25, -0.25, 1, 0xFF0000), + (0.25, -0.25, 1, 0x00FF00), + (-0.25, 0.25, 1, 0x0000FF), + (0.25, 0.25, 1, 0xFFFFFF), + ], + dtype=[("x", " rrb.Blueprint: + return rrb.Blueprint( + rrb.Tabs( + rrb.Spatial3DView( + origin="world", + name="Geometry, TF and colored cloud", + contents=[ + "world/pose/**", + "world/point/**", + "world/odometry/**", + "world/path/**", + "world/entities/**", + "world/cloud/**", + "world/cloud_boxes/**", + "world/cloud_height/**", + "world/legacy_pose/**", + "world/tf/**", + "tf#/**", + ], + ), + rrb.Grid( + rrb.Spatial2DView(origin="world/image", name="Generated RGB pixels"), + rrb.Spatial2DView(origin="world/depth", name="Big endian depth"), + rrb.Spatial2DView(origin="world/compressed", name="PNG pixels"), + rrb.Spatial2DView(origin="world/compressed_depth", name="JXL depth"), + rrb.Spatial2DView(origin="world/late_image", name="Calibration arriving last"), + grid_columns=3, + ), + rrb.TextDocumentView(origin="world/pose/message", name="Exact message values"), + ) + ) + + +class DemoBridge(RerunBridgeModule): + """A real bridge with a receipt signal so the demo can flush before exiting.""" + + def __init__(self, **kwargs: Any) -> None: + super().__init__(**kwargs) + self.received = threading.Event() + self.errors: list[Exception] = [] + + def _on_cdr_message(self, msg: Any, *, topic: str) -> None: + try: + super()._on_cdr_message(msg, topic=topic) + except Exception as error: + self.errors.append(error) + finally: + self.received.set() + + +class LoopbackRPC(ZenohRPC): + def __init__(self, **kwargs: Any) -> None: + super().__init__( + mode="peer", + listen=["tcp/127.0.0.1:0"], + scouting=False, + multicast=False, + gossip=False, + **kwargs, + ) + + +def run(output: Path) -> None: + # Reserve destinations before starting sockets. Existing recordings are never overwritten. + mcap = output.with_suffix(".mcap") + with ExitStack() as cleanup: + for destination in (output, mcap): + cleanup.enter_context(destination.open("xb")) + config: dict[str, Any] = dict( + mode="peer", listen=["tcp/127.0.0.1:0"], scouting=False, multicast=False, gossip=False + ) + messages = scene_messages() + # Select a free local Rerun port instead of attaching to another user's viewer. + with socket.socket() as reserve: + reserve.bind(("127.0.0.1", 0)) + port = reserve.getsockname()[1] + bridge = DemoBridge( + pubsubs=[], + cdr_transports={ + topic: ZenohTransport(Topic(topic, type(value)), **config) + for topic, value in messages.items() + }, + rpc_transport=LoopbackRPC, + rerun_open="none", + rerun_web=False, + connect_url=f"rerun+http://127.0.0.1:{port}/proxy", + tf_axes=0.25, + blueprint=viewer_layout, + visual_override={ + Glob("world/cloud_boxes"): partial(cloud_archetype, mode="boxes"), + Glob("world/cloud_height"): partial( + cloud_archetype, mode="spheres", height_colors=True + ), + }, + ) + cleanup.callback(bridge.stop) + bridge.start() + rr.save(output) + cleanup.callback(rr.disconnect) + rr.send_blueprint(viewer_layout()) + rr.log("world", rr.Transform3D(child_frame="tf#/world"), static=True) + with mcap.open("wb") as stream: + writer = Writer(stream) + writer.start(profile="ros2") + try: + for index, (topic, value) in enumerate(messages.items()): + with ExitStack() as sender: + transport = ZenohTransport(Topic(topic, type(value)), **config) + sender.callback(transport.stop) + bridge.received.clear() + transport.broadcast(None, value) + if not bridge.received.wait(timeout=10): + raise TimeoutError(f"Bridge did not receive {topic}") + if bridge.errors: + raise bridge.errors[0] + schema_id = writer.register_schema( + value.__msgtype__, "ros2msg", schema(value.__msgtype__).encode() + ) + channel = writer.register_channel(topic, "cdr", schema_id) + timestamp = 1_700_000_000_123_456_789 + index * 1_000_000 + writer.add_message( + channel, log_time=timestamp, publish_time=timestamp, data=encode(value) + ) + print(f"Bridge rendered {topic}: {value.__msgtype__}") + finally: + writer.finish() + # Existing messages still take the bridge's ordinary rendering path. + bridge._on_message(LegacyPoseStamped(position=(0, -1, 0), frame_id="map"), "/legacy_pose") + recording = rr.get_global_data_recording() + if recording is not None: + recording.flush() + print(f"Live bridge: rerun {output}") + print(f"Native MCAP view: rerun {mcap}") + print(f"Reusable recording: {mcap}") + print("Source timestamp: 1700000000.123456789; pose [1.5, -2.5, 0.25].") + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "output", + type=Path, + help="new live .rrd filename; also creates a CDR .mcap for native Rerun viewing", + ) + args = parser.parse_args() + run(args.output) + + +if __name__ == "__main__": + main() diff --git a/pyproject.toml b/pyproject.toml index a8e49902d9c..55c9c0d3cf7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -33,6 +33,7 @@ include = ["dimos*"] # (MANIFEST.in already prunes these from the sdist; this keeps their stray # .py modules out of the wheel too). exclude = [ + "dimos.web.relay_bridge.tests*", "dimos.web.websocket_vis.node_modules*", "dimos.web.command-center-extension*", "dimos.web.dimos_interface.src*", @@ -741,7 +742,8 @@ runtime-evaluated-base-classes = ["dimos.core.module.Module"] [tool.mypy] files = ["dimos/"] -mypy_path = "stubs/" +# Editable installs expose generated messages through an import hook; mypy needs the source path. +mypy_path = ["stubs/", "build/message-source/python"] python_version = "3.12" incremental = true strict = true diff --git a/web/cockpit/src/subscriptions.test.ts b/web/cockpit/src/subscriptions.test.ts index 420179092e5..fec2c09a5eb 100644 --- a/web/cockpit/src/subscriptions.test.ts +++ b/web/cockpit/src/subscriptions.test.ts @@ -66,6 +66,23 @@ const voxels = spec({ ch: "global_map", encoding: "voxels.zlib.v1", delivery: "l const map3dPanel = panel({ id: "map3d", kind: "map3d", channels: ["global_map", "odom"] }); describe("subscribableChannels", () => { + it("previews small CDR values but only subscribes CDR sequences through a panel", () => { + const small = spec({ + ch: "scalar", + encoding: "demo_msgs/msg/Scalar.cdr.v1", + params: { cdr: { type: "demo_msgs/msg/Scalar", definition: "float64 value\n" } }, + }); + const bulk = spec({ + ch: "bulk", + encoding: "demo_msgs/msg/Bulk.cdr.v1", + params: { cdr: { type: "demo_msgs/msg/Bulk", definition: "uint8[] data\n" } }, + }); + const view = panel({ id: "view", kind: "video", channels: ["bulk"] }); + expect(channelSubscribable(small, [])).toBe(true); + expect(channelSubscribable(bulk, [])).toBe(false); + expect(channelSubscribable(bulk, [view])).toBe(true); + expect(channelSubscribable({ ...bulk, params: {} }, [view])).toBe(false); + }); it("keeps only channels with a decoder (undecodable ones waste bandwidth)", () => { expect(subscribableChannels([odom, jpeg, future], [videoPanel])).toEqual([odom, jpeg]); expect(subscribableChannels([future], [])).toEqual([]); diff --git a/web/cockpit/src/subscriptions.ts b/web/cockpit/src/subscriptions.ts index 5ff52e5b0ab..87dba75644b 100644 --- a/web/cockpit/src/subscriptions.ts +++ b/web/cockpit/src/subscriptions.ts @@ -5,7 +5,13 @@ // manifest and view change. import type { ChannelSpec, DecoderRegistry, Session } from "@dimos/sdk"; -import { createDecoderRegistry, isLcmSchema, TRACK_ENCODING } from "@dimos/sdk"; +import { + cdrHasSequences, + createDecoderRegistry, + isCdrSchema, + isLcmSchema, + TRACK_ENCODING, +} from "@dimos/sdk"; import type { PanelSpec } from "@dimos/shared"; import { getPanel } from "./panels/registry.tsx"; @@ -30,6 +36,9 @@ const PANEL_ONLY_ENCODINGS = new Set([ * subscribes like json.v1 so the table can show it. */ function panelOnly(spec: ChannelSpec): boolean { if (PANEL_ONLY_ENCODINGS.has(spec.encoding)) return true; + if (/\.cdr\.v1$/.test(spec.encoding) && isCdrSchema(spec.params.cdr)) { + return cdrHasSequences(spec.params.cdr); + } const lcm = spec.params.lcm; return isLcmSchema(lcm) && Object.values(lcm.structs).some((rows) => diff --git a/web/deno.lock b/web/deno.lock index 0fe06125699..f87e68507f3 100644 --- a/web/deno.lock +++ b/web/deno.lock @@ -8,6 +8,8 @@ "jsr:@std/internal@^1.0.12": "1.0.12", "npm:@fontsource-variable/inter@^5.3.0": "5.3.0", "npm:@fontsource-variable/jetbrains-mono@^5.3.0": "5.3.0", + "npm:@foxglove/rosmsg2-serialization@3.1.2": "3.1.2", + "npm:@foxglove/rosmsg@5.0.5": "5.0.5", "npm:@peculiar/x509@2": "2.0.0", "npm:@types/node@^24.10.1": "24.12.2", "npm:@types/react-dom@^19.2.0": "19.2.3_@types+react@19.2.17", @@ -327,6 +329,30 @@ "@fontsource-variable/jetbrains-mono@5.3.0": { "integrity": "sha512-F32xpS2NsGYoQi2ADSkKTgpJj7ozajsGgDJ8woTnqjmIB+dxDIqImjl4pXZVEExu8UFZ2ndhmX18EBS/hdz3Lw==" }, + "@foxglove/cdr@3.5.1": { + "integrity": "sha512-Fubm04/xC1SkeNaviwUhbFk8nAiiNvIqlmP5MW7r6IVGFa35rsA0E3/Rg/pGdE77ZXf4XNwShZcQoEmje62ddQ==" + }, + "@foxglove/message-definition@0.5.0": { + "integrity": "sha512-B/2GvWn4wRSsPDywcmy6uNXgAfZRhiOAnmy3qAaS6MHVqiHQqs/RoDsoW6Gw9sXSweJpeH5qAqAihvm1pR9Thg==" + }, + "@foxglove/rosmsg2-serialization@3.1.2": { + "integrity": "sha512-KKtmHuuIQy6I0JLUi01LFhV7hrZb5xd+LegtuGz2+9L9pftBxtVVKhoxO6B/kQREGocIUOLP8494kxLF58LMWw==", + "dependencies": [ + "@foxglove/cdr", + "@foxglove/message-definition", + "@foxglove/rostime" + ] + }, + "@foxglove/rosmsg@5.0.5": { + "integrity": "sha512-+qY6xW+4HdbPI3ItAlFFMjXgAiplXyRsv+kqdBk8OcWGH38BmEd7s6v/VnCkbYovAoB5rfR5giuKfmPRNzwbrA==", + "dependencies": [ + "@foxglove/message-definition", + "md5-typescript" + ] + }, + "@foxglove/rostime@1.1.3": { + "integrity": "sha512-Gr3Xy4sNbo2NcXg0tBHb/jVUqIcDyZ1kgHj9dB7K+Y1A7AQ0oSCmSXIuf8oqeteSIs0pl8BJOWEwAtLNC99Izg==" + }, "@jridgewell/gen-mapping@0.3.13": { "integrity": "sha512-2kkt/7niJ6MgEPxF0bYdQ6etZaA+fQvDcLKckhy1yIQOzaoKjBBjSj63/aLVjYE3qhRt5dvM+uUyfCg6UKCBbA==", "dependencies": [ @@ -939,6 +965,9 @@ "@jridgewell/sourcemap-codec" ] }, + "md5-typescript@1.0.5": { + "integrity": "sha512-ovAc4EtiNt2dY8JPhPr/wkC9h4U5k/nuClNVcG0Ga3V1rMlYpAY24ZaaymFXJlz+ccJ6UMPo3FSaVKe7czBsXw==" + }, "meshoptimizer@1.1.1": { "integrity": "sha512-oRFNWJRDA/WTrVj7NWvqa5HqE1t9MYDj2VaWirQCzCCrAd2GHrqR/sQezCxiWATPNlKTcRaPRHPJwIRoPBAp5g==" }, @@ -1229,6 +1258,8 @@ "sdk": { "packageJson": { "dependencies": [ + "npm:@foxglove/rosmsg2-serialization@3.1.2", + "npm:@foxglove/rosmsg@5.0.5", "npm:@types/node@^24.10.1", "npm:@types/react-dom@^19.2.0", "npm:@types/react@^19.2.0", diff --git a/web/sdk/fixture/demo_cdr.html b/web/sdk/fixture/demo_cdr.html new file mode 100644 index 00000000000..1ecaca6dc93 --- /dev/null +++ b/web/sdk/fixture/demo_cdr.html @@ -0,0 +1,78 @@ + + + + + + DimOS generated CDR demo + + + +

Generated messages in the browser

+

+ Python-generated CDR bytes, decoded by the DimOS browser SDK from their embedded ROS2 + definitions. Add ?live=1 to receive actual relay frames. Without it, this page + uses fixed packets for byte-order checks. No ROS installation or browser message package. +

+ + +
+
+

+ +

+      
+ Embedded ROS2 definition +

+      
+
+ + + diff --git a/web/sdk/fixture/demo_cdr.ts b/web/sdk/fixture/demo_cdr.ts new file mode 100644 index 00000000000..1f67527578c --- /dev/null +++ b/web/sdk/fixture/demo_cdr.ts @@ -0,0 +1,148 @@ +import frames from "../tests/fixtures/cdr_frames.json"; +import { connect, createDecoderRegistry } from "@dimos/sdk"; +import type { ChannelSpec } from "@dimos/shared"; + +function renderFixture(): void { + const registry = createDecoderRegistry(); + const messages = document.querySelector("#message")!; + const endian = document.querySelector("#endian")!; + for (const [index, vector] of frames.vectors.entries()) { + messages.add(new Option(vector.name, String(index))); + } + function render(): void { + const vector = frames.vectors[Number(messages.value)]; + const channel: ChannelSpec = { + ch: vector.name, + dir: "rx", + encoding: vector.encoding, + delivery: "reliable", + maxHz: 1, + params: { cdr: vector.schema }, + publish: "none", + requiredScope: null, + }; + const encoded = endian.value === "big" ? vector.big_endian_b64 : vector.payload_b64; + const payload = Uint8Array.from(atob(encoded), (c) => c.charCodeAt(0)); + const decoded = registry.resolve(channel)!(payload, { + ch: vector.name, + seq: 1, + ts: 0, + delivery: "reliable", + }); + document.querySelector("#status")!.textContent = + `Decoded ${payload.length} CDR bytes · ${endian.value} endian`; + document.querySelector("#type")!.textContent = vector.schema.type; + document.querySelector("#schema")!.textContent = vector.schema.definition; + document.querySelector("#value")!.textContent = JSON.stringify( + decoded.value, + (_key, value) => + typeof value === "bigint" + ? `${value}n` + : ArrayBuffer.isView(value) + ? Array.from(value as unknown as ArrayLike) + : value, + 2, + ); + const canvas = document.querySelector("#image")!; + canvas.hidden = vector.name !== "image"; + if (!canvas.hidden) { + const image = decoded.value as { data: Uint8Array }; + const rgba = new Uint8ClampedArray(8); + for (let i = 0; i < 2; i++) { + rgba.set(image.data.subarray(i * 3, i * 3 + 3), i * 4); + rgba[i * 4 + 3] = 255; + } + canvas.getContext("2d")!.putImageData(new ImageData(rgba, 2, 1), 0, 0); + } + } + messages.onchange = render; + endian.onchange = render; + render(); +} + +function renderLive(): void { + const messages = document.querySelector("#message")!; + const endian = document.querySelector("#endian")!; + const status = document.querySelector("#status")!; + const values = new Map(); + const subscriptions = new Map void>(); + endian.disabled = true; + const session = connect({ + robot: "cdr-demo", + url: new URLSearchParams(location.search).get("relay") ?? undefined, + }); + function render(): void { + const manifest = session.status.get().manifest; + const channel = manifest?.channels.find((item) => item.ch === messages.value); + const slot = values.get(messages.value); + status.textContent = `Live relay: ${session.status.get().transport.phase}. ` + + `Received ${values.size}/${subscriptions.size} channels. ` + + [...values].map(([name, data]) => `${name}: ${data.frames} frames, ${data.errors} errors`) + .join("; "); + document.querySelector("#type")!.textContent = channel?.encoding ?? "Waiting for manifest"; + document.querySelector("#schema")!.textContent = JSON.stringify(channel?.params.cdr, null, 2) ?? + ""; + document.querySelector("#value")!.textContent = JSON.stringify( + slot?.value, + (_key, value: unknown) => typeof value === "bigint" ? `${value}n` : value, + 2, + ) ?? "Waiting for a live frame"; + const canvas = document.querySelector("#image")!; + canvas.hidden = true; + const image = slot?.value; + if ( + image && typeof image === "object" && "data" in image && image.data instanceof Uint8Array && + "width" in image && typeof image.width === "number" && "height" in image && + typeof image.height === "number" && + "step" in image && typeof image.step === "number" && "encoding" in image && + image.encoding === "rgb8" && + image.width > 0 && image.width <= 512 && image.height > 0 && image.height <= 512 && + image.data.length === image.height * image.step && image.step >= image.width * 3 + ) { + canvas.width = image.width; + canvas.height = image.height; + const rgba = new Uint8ClampedArray(image.width * image.height * 4); + for (let y = 0; y < image.height; y++) { + for (let x = 0; x < image.width; x++) { + const offset = (y * image.width + x) * 4; + rgba.set(image.data.subarray(y * image.step + x * 3, y * image.step + x * 3 + 3), offset); + rgba[offset + 3] = 255; + } + } + canvas.getContext("2d")!.putImageData(new ImageData(rgba, image.width, image.height), 0, 0); + canvas.hidden = false; + } + } + session.status.subscribe(() => { + for (const channel of session.status.get().manifest?.channels ?? []) { + if (subscriptions.has(channel.ch)) continue; + messages.add(new Option(channel.ch, channel.ch)); + subscriptions.set( + channel.ch, + session.subscribe(channel.ch, (snapshot) => { + if (snapshot.slot) { + values.set(channel.ch, { + value: snapshot.slot.value, + frames: snapshot.stats.frames, + errors: snapshot.stats.decodeErrors, + }); + } + render(); + }), + ); + } + render(); + }); + messages.onchange = render; + globalThis.addEventListener("pagehide", () => { + for (const stop of subscriptions.values()) stop(); + session.close(); + }, { once: true }); + render(); +} + +if (new URLSearchParams(location.search).has("live")) { + renderLive(); +} else { + renderFixture(); +} diff --git a/web/sdk/package.json b/web/sdk/package.json index d3521895cda..a67fa16b3e4 100644 --- a/web/sdk/package.json +++ b/web/sdk/package.json @@ -21,5 +21,9 @@ "typescript": "~5.9.2", "vite": "^7.1.0", "vitest": "^3.2.4" + }, + "dependencies": { + "@foxglove/rosmsg": "5.0.5", + "@foxglove/rosmsg2-serialization": "3.1.2" } } diff --git a/web/sdk/src/decoders/cdr.test.ts b/web/sdk/src/decoders/cdr.test.ts new file mode 100644 index 00000000000..73270c2120d --- /dev/null +++ b/web/sdk/src/decoders/cdr.test.ts @@ -0,0 +1,67 @@ +import { describe, expect, it } from "vitest"; +import frames from "../../tests/fixtures/cdr_frames.json"; +import { cdrDecoder, cdrDecoderFor, cdrPreview } from "./cdr.ts"; +import { createDecoderRegistry } from "./index.ts"; +import type { ChannelSpec } from "@dimos/shared"; +const HEADER = { ch: "message", seq: 1, ts: 0, delivery: "latest" as const }; +const bytes = (value: string) => Uint8Array.from(atob(value), (c) => c.charCodeAt(0)); +function plain(value: unknown): unknown { + if (typeof value === "bigint") return String(value); + if (ArrayBuffer.isView(value)) return Array.from(value as unknown as ArrayLike, plain); + if (Array.isArray(value)) return value.map(plain); + if (value !== null && typeof value === "object") { + return Object.fromEntries( + Object.entries(value).map(([key, item]) => [key, plain(item)]), + ); + } + return value; +} +describe("generated CDR browser decoding", () => { + it("resolves a CDR manifest once and refuses conflicting type identities", () => { + const vector = frames.vectors[0]; + const spec: ChannelSpec = { + ch: "pose", + dir: "rx", + encoding: vector.encoding, + delivery: "reliable", + maxHz: 1, + params: { cdr: vector.schema }, + publish: "none", + requiredScope: null, + }; + const registry = createDecoderRegistry(); + const decode = registry.resolve(spec)!; + expect(registry.resolve(spec)).toBe(decode); + expect(plain(decode(bytes(vector.payload_b64), HEADER).value)).toEqual(vector.value); + expect(registry.resolve({ ...spec, encoding: "other_msgs/msg/Pose.cdr.v1" })).toBeUndefined(); + expect(registry.resolve({ ...spec, params: {} })).toBeUndefined(); + expect(registry.resolve({ ...spec })).not.toBe(decode); + }); + for (const vector of frames.vectors) { + it(`decodes ${vector.name} in both byte orders from its embedded schema`, () => { + const decode = cdrDecoder(vector.schema); + for (const encoded of [vector.payload_b64, vector.big_endian_b64]) { + expect(plain(decode(bytes(encoded), HEADER).value)).toEqual(vector.value); + } + }); + } + it("rejects truncation, invalid encapsulation and trailing bytes", () => { + const vector = frames.vectors[0]; + const decode = cdrDecoder(vector.schema); + const payload = bytes(vector.payload_b64); + expect(() => decode(payload.subarray(0, payload.length - 1), HEADER)).toThrow(); + expect(() => decode(new Uint8Array([...payload, 0]), HEADER)).toThrow(/trailing/); + payload[0] = 255; + expect(() => decode(payload, HEADER)).toThrow(/header/); + }); + it("declines missing and malformed schemas", () => { + expect(cdrDecoderFor(undefined)).toBeNull(); + expect(cdrDecoderFor({ type: "p/msg/T", definition: "not a definition !" })).toBeNull(); + expect(cdrDecoderFor({ type: "p.T", definition: "float64 x" })).toBeNull(); + }); + it("bounds previews of large images and arrays", () => { + const text = cdrPreview({ data: new Uint8Array(1000000), label: "x".repeat(10000) }); + expect(text.length).toBeLessThan(550); + expect(text).toContain("Uint8Array(1000000)"); + }); +}); diff --git a/web/sdk/src/decoders/cdr.ts b/web/sdk/src/decoders/cdr.ts new file mode 100644 index 00000000000..10c244c780d --- /dev/null +++ b/web/sdk/src/decoders/cdr.ts @@ -0,0 +1,106 @@ +// Decode generated messages from the manifest's complete ROS2 definition. +import { parse } from "@foxglove/rosmsg"; +import { MessageReader } from "@foxglove/rosmsg2-serialization"; +import type { Decoder } from "./index.ts"; + +export interface CdrSchema { + type: string; + definition: string; +} +export const CDR_ENCODING_RE = /\.cdr\.v1$/; +export const CDR_PREVIEW_MAX_CHARS = 512; +const PREVIEW_MAX_ITEMS = 8; +const PREVIEW_MAX_STRING = 64; + +export function isCdrSchema(value: unknown): value is CdrSchema { + if (value === null || typeof value !== "object") return false; + const schema = value as Record; + return typeof schema.type === "string" && + /^[A-Za-z][A-Za-z0-9_]*\/msg\/[A-Za-z][A-Za-z0-9_]*$/.test(schema.type) && + typeof schema.definition === "string"; +} + +export function cdrDecoder(schema: CdrSchema): Decoder { + const definitions = parse(schema.definition, { ros2: true }); + const reader = new MessageReader(definitions); + return (payload) => { + // Generated codecs use encapsulated XCDR1 with no trailing bytes. + if ( + payload.length < 4 || payload[0] !== 0 || payload[1] > 1 || + payload[2] !== 0 || payload[3] !== 0 + ) throw new Error("invalid XCDR1 header"); + const value = reader.readMessage(payload); + if (reader.lastReadByteLength() !== payload.byteLength) throw new Error("trailing CDR bytes"); + return { value, preview: cdrPreview(value) }; + }; +} + +export function cdrDecoderFor(value: unknown): Decoder | null { + if (!isCdrSchema(value)) return null; + try { + return cdrDecoder(value); + } catch { + return null; + } +} + +/** Variable-length arrays require an explicit panel subscription in the cockpit. */ +export function cdrHasSequences(schema: CdrSchema): boolean { + return parse(schema.definition, { ros2: true }).some((definition) => + definition.definitions.some((field) => field.isArray && field.arrayLength === undefined) + ); +} + +export function cdrPreview(value: unknown, budget = CDR_PREVIEW_MAX_CHARS): string { + const parts: string[] = []; + let len = 0; + let full = false; + const push = (s: string) => { + parts.push(s); + len += s.length; + if (len >= budget) full = true; + }; + const items = (n: number, at: (i: number) => unknown) => { + for (let i = 0; i < n && i < PREVIEW_MAX_ITEMS && !full; i++) { + if (i > 0) push(", "); + walk(at(i)); + } + if (n > PREVIEW_MAX_ITEMS && !full) push(", ..."); + push("]"); + }; + const walk = (v: unknown): void => { + if (full) return; + if (v === null || v === undefined) { + push("null"); + } else if (typeof v === "string") { + push( + JSON.stringify(v.length > PREVIEW_MAX_STRING ? `${v.slice(0, PREVIEW_MAX_STRING)}...` : v), + ); + } else if (typeof v === "number" || typeof v === "boolean" || typeof v === "bigint") { + push(String(v)); + } else if (ArrayBuffer.isView(v)) { + const a = v as unknown as ArrayLike; + push(`${v.constructor.name}(${a.length})[`); + items(a.length, (i) => a[i]); + } else if (Array.isArray(v)) { + push("["); + items(v.length, (i) => v[i]); + } else if (typeof v === "object") { + push("{"); + let first = true; + for (const [key, item] of Object.entries(v)) { + if (full) break; + if (!first) push(", "); + first = false; + push(`${key}: `); + walk(item); + } + push("}"); + } else { + push(String(v)); + } + }; + walk(value); + const text = parts.join(""); + return full ? `${text.slice(0, budget)} ... (truncated)` : text; +} diff --git a/web/sdk/src/decoders/index.ts b/web/sdk/src/decoders/index.ts index dd8ce12e0d3..86e8a105d75 100644 --- a/web/sdk/src/decoders/index.ts +++ b/web/sdk/src/decoders/index.ts @@ -1,5 +1,5 @@ // Payload decoder registry. resolve(spec) finds a manifest channel's decoder: -// a registered encoding id, the *.json.vN convention, or a *.lcm.v1 decoder +// a registered encoding id, the *.json.vN convention, or a *.lcm.v1/*.cdr.v1 decoder // compiled from the schema the robot put in the channel's params. An encoding // without a decoder is not an error: the channel renders as "unsupported" // (forward compatibility with newer bridges). Binary decoders (h264.v1, ...) @@ -9,6 +9,7 @@ import type { ChannelSpec, FrameHeader } from "@dimos/shared"; import { TRACK_ENCODING } from "@dimos/shared/manifest"; +import { CDR_ENCODING_RE, cdrDecoderFor, isCdrSchema } from "./cdr.ts"; import { costmapDecoder } from "./costmap.ts"; import { jpegDecoder } from "./jpeg.ts"; import { jsonDecoder } from "./json.ts"; @@ -27,6 +28,7 @@ export class DecoderRegistry { #decoders = new Map(); // Compiled *.lcm.v1 decoders per adopted manifest record (null: params.lcm // unusable); entries die with the manifest object that carried them. + #cdr = new WeakMap(); #lcm = new WeakMap(); /** Duplicate registration is an error unless `replace` is set. */ @@ -52,6 +54,17 @@ export class DecoderRegistry { resolve(spec: ChannelSpec): Decoder | undefined { const known = this.get(spec.encoding); if (known !== undefined) return known; + if (CDR_ENCODING_RE.test(spec.encoding)) { + if (!isCdrSchema(spec.params.cdr) || spec.encoding !== `${spec.params.cdr.type}.cdr.v1`) { + return undefined; + } + let compiled = this.#cdr.get(spec); + if (compiled === undefined) { + compiled = cdrDecoderFor(spec.params.cdr); + this.#cdr.set(spec, compiled); + } + return compiled ?? undefined; + } if (!LCM_ENCODING_RE.test(spec.encoding)) return undefined; let compiled = this.#lcm.get(spec); if (compiled === undefined) { diff --git a/web/sdk/src/index.ts b/web/sdk/src/index.ts index f7efa3c7846..e4d6d348247 100644 --- a/web/sdk/src/index.ts +++ b/web/sdk/src/index.ts @@ -42,3 +42,5 @@ export type { } from "@dimos/shared"; export { TRACK_ENCODING } from "@dimos/shared/manifest"; export type { Manifest } from "@dimos/shared/manifest"; + +export { cdrDecoder, cdrHasSequences, isCdrSchema } from "./decoders/cdr.ts"; diff --git a/web/sdk/tests/fixtures/cdr_frames.json b/web/sdk/tests/fixtures/cdr_frames.json new file mode 100644 index 00000000000..9cb2c72f5f3 --- /dev/null +++ b/web/sdk/tests/fixtures/cdr_frames.json @@ -0,0 +1,236 @@ +{ + "vectors": [ + { + "name": "pose_stamped", + "encoding": "geometry_msgs/msg/PoseStamped.cdr.v1", + "schema": { + "type": "geometry_msgs/msg/PoseStamped", + "definition": "# A Pose with reference coordinate frame and timestamp\n\nstd_msgs/Header header\nPose pose\n================================================================================\nMSG: builtin_interfaces/Time\n# This message communicates ROS Time defined here:\n# https://design.ros2.org/articles/clock_and_time.html\n\n# The seconds component, valid over all int32 values.\nint32 sec\n\n# The nanoseconds component, valid in the range [0, 1e9), to be added to the seconds component. \n# e.g.\n# The time -1.7 seconds is represented as {sec: -2, nanosec: 3e8}\n# The time 1.7 seconds is represented as {sec: 1, nanosec: 7e8}\nuint32 nanosec\n================================================================================\nMSG: geometry_msgs/Point\n# This contains the position of a point in free space\nfloat64 x\nfloat64 y\nfloat64 z\n================================================================================\nMSG: geometry_msgs/Pose\n# A representation of pose in free space, composed of position and orientation.\n\nPoint position\nQuaternion orientation\n================================================================================\nMSG: geometry_msgs/Quaternion\n# This represents an orientation in free space in quaternion form.\n\nfloat64 x 0\nfloat64 y 0\nfloat64 z 0\nfloat64 w 1\n================================================================================\nMSG: std_msgs/Header\n# Standard metadata for higher-level stamped data types.\n# This is generally used to communicate timestamped data\n# in a particular coordinate frame.\n\n# Two-integer timestamp that is expressed as seconds and nanoseconds.\nbuiltin_interfaces/Time stamp\n\n# Transform frame with which this data is associated.\nstring frame_id\n" + }, + "value": { + "header": { + "stamp": { + "sec": 1757400000, + "nanosec": 500000123 + }, + "frame_id": "map" + }, + "pose": { + "position": { + "x": 1.5, + "y": -2.5, + "z": 0.25 + }, + "orientation": { + "x": 0.0, + "y": 0.0, + "z": 0.6, + "w": 0.8 + } + } + }, + "payload_b64": "AAEAAMDLv2h7Zc0dBAAAAG1hcAAAAAAAAAD4PwAAAAAAAATAAAAAAAAA0D8AAAAAAAAAAAAAAAAAAAAAMzMzMzMz4z+amZmZmZnpPw==", + "big_endian_b64": "AAAAAGi/y8AdzWV7AAAABG1hcAA/+AAAAAAAAMAEAAAAAAAAP9AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAP+MzMzMzMzM/6ZmZmZmZmg==" + }, + { + "name": "custom_segments", + "encoding": "dimos_msgs/msg/LineSegments3D.cdr.v1", + "schema": { + "type": "dimos_msgs/msg/LineSegments3D", + "definition": "# Copyright 2026 Dimensional Inc.\n# SPDX-License-Identifier: Apache-2.0\n\n# Weighted segments in a common coordinate frame.\nstd_msgs/Header header\nLineSegment3D[] segments\n================================================================================\nMSG: builtin_interfaces/Time\n# This message communicates ROS Time defined here:\n# https://design.ros2.org/articles/clock_and_time.html\n\n# The seconds component, valid over all int32 values.\nint32 sec\n\n# The nanoseconds component, valid in the range [0, 1e9), to be added to the seconds component. \n# e.g.\n# The time -1.7 seconds is represented as {sec: -2, nanosec: 3e8}\n# The time 1.7 seconds is represented as {sec: 1, nanosec: 7e8}\nuint32 nanosec\n================================================================================\nMSG: dimos_msgs/LineSegment3D\n# Copyright 2026 Dimensional Inc.\n# SPDX-License-Identifier: Apache-2.0\n\ngeometry_msgs/Point start\ngeometry_msgs/Point end\nfloat64 weight 1.0\n================================================================================\nMSG: geometry_msgs/Point\n# This contains the position of a point in free space\nfloat64 x\nfloat64 y\nfloat64 z\n================================================================================\nMSG: std_msgs/Header\n# Standard metadata for higher-level stamped data types.\n# This is generally used to communicate timestamped data\n# in a particular coordinate frame.\n\n# Two-integer timestamp that is expressed as seconds and nanoseconds.\nbuiltin_interfaces/Time stamp\n\n# Transform frame with which this data is associated.\nstring frame_id\n" + }, + "value": { + "header": { + "stamp": { + "sec": 1757400000, + "nanosec": 500000123 + }, + "frame_id": "map" + }, + "segments": [ + { + "start": { + "x": 1, + "y": 0.0, + "z": 0.0 + }, + "end": { + "x": 0.0, + "y": 2, + "z": 0.0 + }, + "weight": 4 + } + ] + }, + "payload_b64": "AAEAAMDLv2h7Zc0dBAAAAG1hcAABAAAAAAAAAAAAAAAAAPA/AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAEAAAAAAAAAAAAAAAAAAABBA", + "big_endian_b64": "AAAAAGi/y8AdzWV7AAAABG1hcAAAAAABAAAAAD/wAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAQAAAAAAAAAAAAAAAAAAAAEAQAAAAAAAA" + }, + { + "name": "image", + "encoding": "sensor_msgs/msg/Image.cdr.v1", + "schema": { + "type": "sensor_msgs/msg/Image", + "definition": "# This message contains an uncompressed image\n# (0, 0) is at top-left corner of image\n\nstd_msgs/Header header # Header timestamp should be acquisition time of image\n # Header frame_id should be optical frame of camera\n # origin of frame should be optical center of cameara\n # +x should point to the right in the image\n # +y should point down in the image\n # +z should point into to plane of the image\n # If the frame_id here and the frame_id of the CameraInfo\n # message associated with the image conflict\n # the behavior is undefined\n\nuint32 height # image height, that is, number of rows\nuint32 width # image width, that is, number of columns\n\n# The legal values for encoding are in file include/sensor_msgs/image_encodings.hpp\n# If you want to standardize a new string format, join\n# ros-users@lists.ros.org and send an email proposing a new encoding.\n\nstring encoding # Encoding of pixels -- channel meaning, ordering, size\n # taken from the list of strings in include/sensor_msgs/image_encodings.hpp\n\nuint8 is_bigendian # is this data bigendian?\nuint32 step # Full row length in bytes\nuint8[] data # actual matrix data, size is (step * rows)\n================================================================================\nMSG: builtin_interfaces/Time\n# This message communicates ROS Time defined here:\n# https://design.ros2.org/articles/clock_and_time.html\n\n# The seconds component, valid over all int32 values.\nint32 sec\n\n# The nanoseconds component, valid in the range [0, 1e9), to be added to the seconds component. \n# e.g.\n# The time -1.7 seconds is represented as {sec: -2, nanosec: 3e8}\n# The time 1.7 seconds is represented as {sec: 1, nanosec: 7e8}\nuint32 nanosec\n================================================================================\nMSG: std_msgs/Header\n# Standard metadata for higher-level stamped data types.\n# This is generally used to communicate timestamped data\n# in a particular coordinate frame.\n\n# Two-integer timestamp that is expressed as seconds and nanoseconds.\nbuiltin_interfaces/Time stamp\n\n# Transform frame with which this data is associated.\nstring frame_id\n" + }, + "value": { + "header": { + "stamp": { + "sec": 1757400000, + "nanosec": 500000123 + }, + "frame_id": "map" + }, + "height": 1, + "width": 2, + "encoding": "rgb8", + "is_bigendian": 0, + "step": 6, + "data": [ + 0, + 1, + 2, + 253, + 254, + 255 + ] + }, + "payload_b64": "AAEAAMDLv2h7Zc0dBAAAAG1hcAABAAAAAgAAAAUAAAByZ2I4AAAAAAYAAAAGAAAAAAEC/f7/", + "big_endian_b64": "AAAAAGi/y8AdzWV7AAAABG1hcAAAAAABAAAAAgAAAAVyZ2I4AAAAAAAAAAYAAAAGAAEC/f7/" + }, + { + "name": "imu", + "encoding": "sensor_msgs/msg/Imu.cdr.v1", + "schema": { + "type": "sensor_msgs/msg/Imu", + "definition": "# This is a message to hold data from an IMU (Inertial Measurement Unit)\n#\n# Accelerations should be in m/s^2 (not in g's), and rotational velocity should be in rad/sec\n#\n# If the covariance of the measurement is known, it should be filled in (if all you know is the\n# variance of each measurement, e.g. from the datasheet, just put those along the diagonal)\n# A covariance matrix of all zeros will be interpreted as \"covariance unknown\", and to use the\n# data a covariance will have to be assumed or gotten from some other source\n#\n# If you have no estimate for one of the data elements (e.g. your IMU doesn't produce an\n# orientation estimate), please set element 0 of the associated covariance matrix to -1\n# If you are interpreting this message, please check for a value of -1 in the first element of each\n# covariance matrix, and disregard the associated estimate.\n\nstd_msgs/Header header\n\ngeometry_msgs/Quaternion orientation\nfloat64[9] orientation_covariance # Row major about x, y, z axes\n\ngeometry_msgs/Vector3 angular_velocity\nfloat64[9] angular_velocity_covariance # Row major about x, y, z axes\n\ngeometry_msgs/Vector3 linear_acceleration\nfloat64[9] linear_acceleration_covariance # Row major x, y z\n================================================================================\nMSG: builtin_interfaces/Time\n# This message communicates ROS Time defined here:\n# https://design.ros2.org/articles/clock_and_time.html\n\n# The seconds component, valid over all int32 values.\nint32 sec\n\n# The nanoseconds component, valid in the range [0, 1e9), to be added to the seconds component. \n# e.g.\n# The time -1.7 seconds is represented as {sec: -2, nanosec: 3e8}\n# The time 1.7 seconds is represented as {sec: 1, nanosec: 7e8}\nuint32 nanosec\n================================================================================\nMSG: geometry_msgs/Quaternion\n# This represents an orientation in free space in quaternion form.\n\nfloat64 x 0\nfloat64 y 0\nfloat64 z 0\nfloat64 w 1\n================================================================================\nMSG: geometry_msgs/Vector3\n# This represents a vector in free space.\n\n# This is semantically different than a point.\n# A vector is always anchored at the origin.\n# When a transform is applied to a vector, only the rotational component is applied.\n\nfloat64 x\nfloat64 y\nfloat64 z\n================================================================================\nMSG: std_msgs/Header\n# Standard metadata for higher-level stamped data types.\n# This is generally used to communicate timestamped data\n# in a particular coordinate frame.\n\n# Two-integer timestamp that is expressed as seconds and nanoseconds.\nbuiltin_interfaces/Time stamp\n\n# Transform frame with which this data is associated.\nstring frame_id\n" + }, + "value": { + "header": { + "stamp": { + "sec": 1757400000, + "nanosec": 500000123 + }, + "frame_id": "map" + }, + "orientation": { + "x": 0.0, + "y": 0.0, + "z": 0.0, + "w": 1.0 + }, + "orientation_covariance": [ + 0.0, + 1.0, + 2.0, + 3.0, + 4.0, + 5.0, + 6.0, + 7.0, + 8.0 + ], + "angular_velocity": { + "x": 0.0, + "y": 0.0, + "z": 0.0 + }, + "angular_velocity_covariance": [ + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0 + ], + "linear_acceleration": { + "x": 0.0, + "y": 0.0, + "z": 0.0 + }, + "linear_acceleration_covariance": [ + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0, + 0.0 + ] + }, + "payload_b64": "AAEAAMDLv2h7Zc0dBAAAAG1hcAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAADwPwAAAAAAAAAAAAAAAAAA8D8AAAAAAAAAQAAAAAAAAAhAAAAAAAAAEEAAAAAAAAAUQAAAAAAAABhAAAAAAAAAHEAAAAAAAAAgQAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA==", + "big_endian_b64": "AAAAAGi/y8AdzWV7AAAABG1hcAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA/8AAAAAAAAAAAAAAAAAAAP/AAAAAAAABAAAAAAAAAAEAIAAAAAAAAQBAAAAAAAABAFAAAAAAAAEAYAAAAAAAAQBwAAAAAAABAIAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA==" + }, + { + "name": "joints", + "encoding": "sensor_msgs/msg/JointState.cdr.v1", + "schema": { + "type": "sensor_msgs/msg/JointState", + "definition": "# This is a message that holds data to describe the state of a set of torque controlled joints.\n#\n# The state of each joint (revolute or prismatic) is defined by:\n# * the position of the joint (rad or m),\n# * the velocity of the joint (rad/s or m/s) and\n# * the effort that is applied in the joint (Nm or N).\n#\n# Each joint is uniquely identified by its name\n# The header specifies the time at which the joint states were recorded. All the joint states\n# in one message have to be recorded at the same time.\n#\n# This message consists of a multiple arrays, one for each part of the joint state.\n# The goal is to make each of the fields optional. When e.g. your joints have no\n# effort associated with them, you can leave the effort array empty.\n#\n# All arrays in this message should have the same size, or be empty.\n# This is the only way to uniquely associate the joint name with the correct\n# states.\n\nstd_msgs/Header header\n\nstring[] name\nfloat64[] position\nfloat64[] velocity\nfloat64[] effort\n================================================================================\nMSG: builtin_interfaces/Time\n# This message communicates ROS Time defined here:\n# https://design.ros2.org/articles/clock_and_time.html\n\n# The seconds component, valid over all int32 values.\nint32 sec\n\n# The nanoseconds component, valid in the range [0, 1e9), to be added to the seconds component. \n# e.g.\n# The time -1.7 seconds is represented as {sec: -2, nanosec: 3e8}\n# The time 1.7 seconds is represented as {sec: 1, nanosec: 7e8}\nuint32 nanosec\n================================================================================\nMSG: std_msgs/Header\n# Standard metadata for higher-level stamped data types.\n# This is generally used to communicate timestamped data\n# in a particular coordinate frame.\n\n# Two-integer timestamp that is expressed as seconds and nanoseconds.\nbuiltin_interfaces/Time stamp\n\n# Transform frame with which this data is associated.\nstring frame_id\n" + }, + "value": { + "header": { + "stamp": { + "sec": 1757400000, + "nanosec": 500000123 + }, + "frame_id": "map" + }, + "name": [ + "joint_a", + "joint_b" + ], + "position": [ + 1.5, + -2.5 + ], + "velocity": [], + "effort": [] + }, + "payload_b64": "AAEAAMDLv2h7Zc0dBAAAAG1hcAACAAAACAAAAGpvaW50X2EACAAAAGpvaW50X2IAAgAAAAAAAAAAAPg/AAAAAAAABMAAAAAAAAAAAA==", + "big_endian_b64": "AAAAAGi/y8AdzWV7AAAABG1hcAAAAAACAAAACGpvaW50X2EAAAAACGpvaW50X2IAAAAAAj/4AAAAAAAAwAQAAAAAAAAAAAAAAAAAAA==" + }, + { + "name": "int64", + "encoding": "std_msgs/msg/Int64.cdr.v1", + "schema": { + "type": "std_msgs/msg/Int64", + "definition": "# This was originally provided as an example message.\n# It is deprecated as of Foxy\n# It is recommended to create your own semantically meaningful message.\n# However if you would like to continue using this please use the equivalent in example_msgs.\n\nint64 data\n" + }, + "value": { + "data": "-9007199254740993" + }, + "payload_b64": "AAEAAP///////9//", + "big_endian_b64": "AAAAAP/f////////" + }, + { + "name": "empty", + "encoding": "std_msgs/msg/Empty.cdr.v1", + "schema": { + "type": "std_msgs/msg/Empty", + "definition": "\n" + }, + "value": {}, + "payload_b64": "AAEAAAA=", + "big_endian_b64": "AAAAAAA=" + }, + { + "name": "string", + "encoding": "std_msgs/msg/String.cdr.v1", + "schema": { + "type": "std_msgs/msg/String", + "definition": "# This was originally provided as an example message.\n# It is deprecated as of Foxy\n# It is recommended to create your own semantically meaningful message.\n# However if you would like to continue using this please use the equivalent in example_msgs.\n\nstring data\n" + }, + "value": { + "data": "robot 🦾 café" + }, + "payload_b64": "AAEAABEAAAByb2JvdCDwn6a+IGNhZsOpAA==", + "big_endian_b64": "AAAAAAAAABFyb2JvdCDwn6a+IGNhZsOpAA==" + } + ] +}