refactor(perception): migrate callers to generated CDR messages

dimos/experimental/memory/rust/Cargo.toml

@@ -9,7 +9,7 @@ license = "Apache-2.0"
 anyhow = "1"
 crossbeam-channel = "0.5"
 crossbeam-utils = "0.8"
-dimos-module = { git = "https://github.com/dimensionalOS/dimos", rev = "6663bcb6e5e18f43a6cb9cf77ccbec613e310443", version = "=0.1.0" }
+dimos-module = { git = "https://github.com/dimensionalOS/dimos", rev = "fc6a4e58da89bb073a38dce5ac63728c6559d3a6", version = "=0.1.0" }
 lcm-msgs = { git = "https://github.com/dimensionalOS/dimos-lcm.git", branch = "rust-codegen" }
 lz4_flex = "0.14.0"
 mcap = { version = "0.25.0", default-features = false, features = ["zstd"] }

dimos/experimental/memory/rust/crate-hashes.json

@@ -1,6 +1,7 @@
 {
-  "git+https://github.com/dimensionalOS/dimos-lcm.git?branch=rust-codegen#dimos-lcm@0.1.0": "sha256-GGkx4Mn6NYP6KZecmoRLKGWIih/+y8OgNn12DeXX6n8=",
   "git+https://github.com/dimensionalOS/dimos-lcm.git?branch=rust-codegen#lcm-msgs@0.1.0": "sha256-GGkx4Mn6NYP6KZecmoRLKGWIih/+y8OgNn12DeXX6n8=",
-  "git+https://github.com/dimensionalOS/dimos?rev=6663bcb6e5e18f43a6cb9cf77ccbec613e310443#dimos-module-macros@0.1.0": "sha256-6ikqdS3sroOJGI4WGuSyFx2yZjtJ+4Qgt+IGlkFl+qA=",
-  "git+https://github.com/dimensionalOS/dimos?rev=6663bcb6e5e18f43a6cb9cf77ccbec613e310443#dimos-module@0.1.0": "sha256-6ikqdS3sroOJGI4WGuSyFx2yZjtJ+4Qgt+IGlkFl+qA="
+  "git+https://github.com/dimensionalOS/dimos-lcm.git?rev=04d78e8622500244123ba9cefa4c51b4cb454549#dimos-lcm@0.1.0": "sha256-GGkx4Mn6NYP6KZecmoRLKGWIih/+y8OgNn12DeXX6n8=",
+  "git+https://github.com/dimensionalOS/dimos-lcm.git?rev=04d78e8622500244123ba9cefa4c51b4cb454549#lcm-msgs@0.1.0": "sha256-GGkx4Mn6NYP6KZecmoRLKGWIih/+y8OgNn12DeXX6n8=",
+  "git+https://github.com/dimensionalOS/dimos?rev=fc6a4e58da89bb073a38dce5ac63728c6559d3a6#dimos-module-macros@0.1.0": "sha256-ECU2xYh/j+a3XwyhO3UR/GPruo5jNzM3n/9S5BIvVrA=",
+  "git+https://github.com/dimensionalOS/dimos?rev=fc6a4e58da89bb073a38dce5ac63728c6559d3a6#dimos-module@0.1.0": "sha256-ECU2xYh/j+a3XwyhO3UR/GPruo5jNzM3n/9S5BIvVrA="
 }

dimos/hardware/sensors/camera/gstreamer/gstreamer_camera.py

@@ -26,7 +26,10 @@
 from dimos.core.core import rpc
 from dimos.core.module import Module, ModuleConfig
 from dimos.core.stream import Out
-from dimos.msgs.sensor_msgs.Image import Image, ImageFormat
+from dimos.message_utils.image import image_from_array
+from dimos.message_utils.time import duration_from_seconds, time_from_nanoseconds, to_nanoseconds
+from dimos.messages.sensor_msgs.msg import Image
+from dimos.messages.std_msgs.msg import Header
 from dimos.utils.logging_config import setup_logger
 
 # Add system path for gi module if needed
@@ -37,7 +40,8 @@
 
 gi.require_version("Gst", "1.0")
 gi.require_version("GstApp", "1.0")
-from gi.repository import GLib, Gst  # type: ignore[import-not-found,import-untyped]
+gi.require_version("GstVideo", "1.0")
+from gi.repository import GLib, Gst, GstVideo  # type: ignore[import-not-found,import-untyped]
 
 logger = setup_logger(level=logging.INFO)
 
@@ -101,6 +105,7 @@ def stop(self) -> None:
         self._cleanup_reconnect_timer()
 
         if not self.running:
+            super().stop()
             return
 
         self.running = False
@@ -251,12 +256,15 @@ def _on_new_sample(self, appsink):  # type: ignore[no-untyped-def]
         if buffer.pts != Gst.CLOCK_TIME_NONE:
             # Convert nanoseconds to seconds and add offset
             # This is the absolute time from when the frame was captured
-            timestamp = (buffer.pts / 1e9) + self.config.timestamp_offset
+            stamp_ns = int(buffer.pts) + to_nanoseconds(
+                duration_from_seconds(self.config.timestamp_offset)
+            )
+            timestamp = stamp_ns / 1e9
 
             # Skip frames with invalid timestamps (before year 2000)
             # This filters out initial gray frames with relative timestamps
             year_2000_timestamp = 946684800.0  # January 1, 2000 00:00:00 UTC
-            if timestamp < year_2000_timestamp:
+            if stamp_ns < int(year_2000_timestamp) * 1_000_000_000:
                 logger.debug(f"Skipping frame with invalid timestamp: {timestamp:.6f}")
                 return Gst.FlowReturn.OK
 
@@ -272,18 +280,22 @@ def _on_new_sample(self, appsink):  # type: ignore[no-untyped-def]
         try:
             # Convert buffer data to numpy array
             # The videoconvert element outputs BGR format
-            data = np.frombuffer(map_info.data, dtype=np.uint8)
-
-            # Reshape to image dimensions
-            # For BGR format, we have 3 channels
-            image_array = data.reshape((height, width, 3))
+            info = GstVideo.VideoInfo.new_from_caps(caps)
+            if info.finfo.name != "BGR":
+                raise ValueError("The camera pipeline must supply BGR pixels")
+            image_array = np.ndarray(
+                (height, width, 3),
+                dtype=np.uint8,
+                buffer=map_info.data,
+                offset=info.offset[0],
+                strides=(info.stride[0], 3, 1),
+            )
 
             # Create an Image message with the absolute timestamp
-            image_msg = Image(
-                data=image_array.copy(),  # Make a copy to ensure data persistence
-                format=ImageFormat.BGR,
-                frame_id=self.frame_id,
-                ts=timestamp,
+            image_msg = image_from_array(
+                image_array,
+                encoding="bgr8",
+                header=Header(frame_id=self.frame_id, stamp=time_from_nanoseconds(stamp_ns)),
             )
 
             # Publish the image

dimos/hardware/sensors/camera/gstreamer/test_generated_frames.py

@@ -0,0 +1,105 @@
+# 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.
+
+"""Actual in-memory GStreamer buffers/pipelines; no camera, TCP stream or fake SDK."""
+
+import gc
+import threading
+from unittest.mock import Mock
+
+from dimos_cdrgen.registry import decode, encode
+import numpy as np
+import pytest
+
+from dimos.message_utils.image import image_view
+from dimos.message_utils.time import to_nanoseconds
+from dimos.messages.sensor_msgs.msg import Image
+from dimos.protocol.rpc.zenohrpc import ZenohRPC
+
+gi = pytest.importorskip("gi", reason="GStreamer acceptance requires the optional GI binding")
+gi.require_version("Gst", "1.0")
+from gi.repository import Gst
+
+from dimos.hardware.sensors.camera.gstreamer.gstreamer_camera import GstreamerCameraModule
+
+
+@pytest.fixture
+def camera(mocker):
+    mocker.patch.object(ZenohRPC, "start")
+    mocker.patch.object(ZenohRPC, "serve_module_rpc")
+    instance = GstreamerCameraModule(frame_id="camera_test")
+    yield instance
+    instance.stop()
+
+
+@pytest.mark.parametrize("offset,delta_ns", [(0.0, 0), (1e-9, 1), (-1e-9, -1)])
+def test_actual_pipeline_copies_padded_bgr_and_exact_stamp(
+    camera, mocker, offset, delta_ns
+) -> None:
+    camera.config.timestamp_offset = offset
+    camera.running = True
+    pipeline = Gst.parse_launch(
+        "appsrc name=source format=time ! video/x-raw,format=BGR,width=3,height=2,framerate=30/1 "
+        "! appsink name=sink emit-signals=true sync=false"
+    )
+    sink, source = pipeline.get_by_name("sink"), pipeline.get_by_name("source")
+    sink.connect("new-sample", camera._on_new_sample)
+    received = threading.Event()
+    published = []
+
+    def publish(value):
+        published.append(value)
+        received.set()
+
+    mocker.patch.object(camera.video, "publish", side_effect=publish)
+    pixels = np.arange(18, dtype=np.uint8).reshape(2, 3, 3)
+    raw = bytearray([99] * 24)
+    np.ndarray((2, 3, 3), dtype=np.uint8, buffer=raw, strides=(12, 3, 1))[:] = pixels
+    buffer = Gst.Buffer.new_allocate(None, len(raw), None)
+    buffer.fill(0, raw)
+    buffer.pts = 1700000000123456789
+    try:
+        pipeline.set_state(Gst.State.PLAYING)
+        assert source.emit("push-buffer", buffer) == Gst.FlowReturn.OK
+        source.emit("end-of-stream")
+        assert received.wait(timeout=5)
+    finally:
+        pipeline.set_state(Gst.State.NULL)
+    del buffer, raw, pipeline
+    gc.collect()
+    assert len(published) == 1
+    assert type(published[0]) is Image
+    message = decode(encode(published[0]), Image)
+    assert message.encoding == "bgr8" and message.step == 9
+    assert to_nanoseconds(message.header.stamp) == 1700000000123456789 + delta_ns
+    assert message.header.frame_id == "camera_test"
+    np.testing.assert_array_equal(image_view(message), pixels)
+
+
+def test_invalid_source_stamp_keeps_existing_skip_behavior(camera, mocker) -> None:
+    buffer = Gst.Buffer.new_allocate(None, 24, None)
+    buffer.pts = 0
+    caps = Gst.Caps.from_string("video/x-raw,format=BGR,width=3,height=2,framerate=30/1")
+    sample = Gst.Sample.new(buffer, caps, None, None)
+    sink = Mock()
+    sink.emit.return_value = sample
+    publish = mocker.patch.object(camera.video, "publish")
+    assert camera._on_new_sample(sink) == Gst.FlowReturn.OK
+    publish.assert_not_called()
+
+
+def test_stop_closes_module_loop_when_pipeline_was_never_started(camera):
+    loop_thread = camera._loop_thread
+    camera.stop()
+    assert not loop_thread.is_alive()

dimos/hardware/sensors/camera/module.py

@@ -13,9 +13,10 @@
 # limitations under the License.
 
 from collections.abc import Callable
+from copy import deepcopy
 import time
 
-from pydantic import Field
+from pydantic import Field, InstanceOf
 import reactivex as rx
 
 from dimos.core.coordination.blueprints import autoconnect
@@ -25,28 +26,32 @@
 from dimos.core.stream import Out
 from dimos.hardware.sensors.camera.spec import CameraHardware
 from dimos.hardware.sensors.camera.webcam import Webcam
-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, sharpness_barrier
-from dimos.msgs.tf2_msgs.TFMessage import TFMessage
+from dimos.message_utils.image import image_sharpness
+from dimos.message_utils.time import time_from_nanoseconds
+from dimos.messages.builtin_interfaces.msg import Time
+from dimos.messages.geometry_msgs.msg import Quaternion, Transform, TransformStamped, Vector3
+from dimos.messages.sensor_msgs.msg import CameraInfo, Image
+from dimos.messages.std_msgs.msg import Header
+from dimos.messages.tf2_msgs.msg import TFMessage
 from dimos.spec import perception
+from dimos.utils.reactive import quality_barrier
 from dimos.visualization.vis_module import vis_module
 
 
-def default_transform() -> Transform:
-    return Transform(
-        translation=Vector3(0.0, 0.0, 0.0),
-        rotation=Quaternion(0.0, 0.0, 0.0, 1.0),
-        frame_id="base_link",
+def default_transform() -> TransformStamped:
+    return TransformStamped(
+        header=Header(frame_id="base_link", stamp=Time(sec=0, nanosec=0)),
         child_frame_id="camera_link",
+        transform=Transform(
+            rotation=Quaternion(w=1.0, x=0.0, y=0.0, z=0.0),
+            translation=Vector3(x=0.0, y=0.0, z=0.0),
+        ),
     )
 
 
 class CameraModuleConfig(ModuleConfig):
     frame_id: str = "camera_link"
-    transform: Transform | None = Field(default_factory=default_transform)
+    transform: InstanceOf[TransformStamped] | None = Field(default_factory=default_transform)
     hardware: Callable[[], CameraHardware] | CameraHardware = Webcam
     frequency: float = 0.0  # Hz, 0 means no limit
 
@@ -71,7 +76,7 @@ def start(self) -> None:
         stream = self.hardware.image_stream()
 
         if self.config.frequency > 0:
-            stream = stream.pipe(sharpness_barrier(self.config.frequency))
+            stream = stream.pipe(quality_barrier(image_sharpness, self.config.frequency))
 
         self.register_disposable(
             stream.subscribe(self.color_image.publish),
@@ -82,24 +87,24 @@ def start(self) -> None:
         )
 
     def publish_metadata(self) -> None:
-        camera_info = self.hardware.camera_info.with_ts(time.time())
+        camera_info = deepcopy(self.hardware.camera_info)
+        camera_info.header.stamp = time_from_nanoseconds(time.time_ns())
         self.camera_info.publish(camera_info)
 
         if not self.config.transform:
             return
 
-        camera_link = self.config.transform
-        camera_link.ts = camera_info.ts
-
-        camera_optical = Transform(
-            translation=Vector3(0.0, 0.0, 0.0),
-            rotation=Quaternion(-0.5, 0.5, -0.5, 0.5),
-            frame_id="camera_link",
-            child_frame_id="camera_optical",
-            ts=camera_link.ts,
+        camera_link = deepcopy(self.config.transform)
+        camera_link.header.stamp = camera_info.header.stamp
+        camera_optical = TransformStamped(
+            header=Header(frame_id=camera_link.child_frame_id, stamp=camera_info.header.stamp),
+            child_frame_id=camera_info.header.frame_id,
+            transform=Transform(
+                rotation=Quaternion(x=-0.5, y=0.5, z=-0.5, w=0.5),
+                translation=Vector3(x=0.0, y=0.0, z=0.0),
+            ),
         )
-
-        self.tf.publish(TFMessage(camera_link, camera_optical))
+        self.tf.publish(TFMessage(transforms=[camera_link, camera_optical]))
 
     @rpc
     def stop(self) -> None: