refactor(perception): migrate temporal memory and entity rendering to CDR

.github/workflows/message-codegen.yml

@@ -197,3 +197,8 @@ jobs:
           .native-review/bin/python -m pip install 'hydra-core==1.3.2'
           .native-review/bin/python -I -c 'import dimos; import pytest; import sys; raise SystemExit(pytest.main(sys.argv[1:]))' --noconftest --import-mode=importlib -o addopts='' -o 'markers=self_hosted: optional heavyweight integration' --strict-markers -m 'not self_hosted' -q "$PWD/dimos/perception/experimental/test_object_generated.py" "$PWD/dimos/perception/experimental/test_object_scene_registration_temporal.py" "$PWD/dimos/perception/experimental/test_generated_tracker.py" "$PWD/dimos/perception/experimental/test_spatial_memory.py" "$PWD/dimos/models/segmentation/test_generated_edge_tam_images.py" "$PWD/dimos/models/segmentation/test_yoloe.py"
           MYPYPATH="$PWD/build/message-source/python" .native-review/bin/python -m mypy --follow-imports=silent dimos/perception/experimental/object.py dimos/perception/experimental/objectDB.py dimos/perception/experimental/object_scene_registration.py dimos/perception/experimental/object_scene_registration_spec.py dimos/perception/experimental/object_tracker.py dimos/perception/experimental/object_tracker_2d.py dimos/perception/experimental/object_tracker_3d.py dimos/perception/experimental/spatial_perception.py dimos/models/segmentation/edge_tam.py dimos/models/segmentation/yoloe.py
+      - name: Verify generated temporal memory, PerceiveLoop and entity rendering
+        if: runner.os == 'Linux'
+        run: |
+          .native-review/bin/python -I -c 'import dimos; import pytest; import sys; raise SystemExit(pytest.main(sys.argv[1:]))' --noconftest --import-mode=importlib -o addopts='' -q "$PWD/dimos/perception/experimental/temporal_memory/test_temporal_memory_module.py" "$PWD/dimos/perception/experimental/test_perceive_loop_skill.py" "$PWD/dimos/visualization/rerun/test_cdr.py"
+          MYPYPATH="$PWD/build/message-source/python" .native-review/bin/python -m mypy --follow-imports=silent dimos/perception/experimental/perceive_loop_skill.py dimos/perception/experimental/temporal_memory/clip_filter.py dimos/perception/experimental/temporal_memory/entity_graph_db.py dimos/perception/experimental/temporal_memory/frame_window_accumulator.py dimos/perception/experimental/temporal_memory/temporal_memory.py dimos/perception/experimental/temporal_memory/temporal_utils/helpers.py dimos/perception/experimental/temporal_memory/window_analyzer.py dimos/visualization/rerun/cdr.py

dimos/models/embedding/clip.py

@@ -17,12 +17,12 @@
 from functools import cached_property
 from typing import overload
 
-from dimos.messages.sensor_msgs.msg import Image
 from PIL import Image as PILImage
 import torch
 import torch.nn.functional as functional
 from transformers import CLIPModel as HFCLIPModel, CLIPProcessor
 
+from dimos.messages.sensor_msgs.msg import Image
 from dimos.models.base import HuggingFaceModel
 from dimos.models.embedding.base import Embedding, EmbeddingModel, HuggingFaceEmbeddingModelConfig
 from dimos.msgs.image import image_to_rgb

dimos/models/embedding/dino.py

@@ -17,12 +17,12 @@
 from functools import cached_property
 from typing import Any, overload
 
-from dimos.messages.sensor_msgs.msg import Image
 from PIL import Image as PILImage
 import torch
 import torch.nn.functional as functional
 from transformers import AutoImageProcessor, AutoModel
 
+from dimos.messages.sensor_msgs.msg import Image
 from dimos.models.base import HuggingFaceModel
 from dimos.models.embedding.base import Embedding, EmbeddingModel, HuggingFaceEmbeddingModelConfig
 from dimos.msgs.image import image_to_rgb

dimos/models/embedding/siglip.py

@@ -17,12 +17,12 @@
 from functools import cached_property
 from typing import overload
 
-from dimos.messages.sensor_msgs.msg import Image
 from PIL import Image as PILImage
 import torch
 import torch.nn.functional as functional
 from transformers import SiglipModel as HFSiglipModel, SiglipProcessor
 
+from dimos.messages.sensor_msgs.msg import Image
 from dimos.models.base import HuggingFaceModel
 from dimos.models.embedding.base import Embedding, EmbeddingModel, HuggingFaceEmbeddingModelConfig
 from dimos.msgs.image import image_to_rgb

dimos/models/embedding/siglip2.py

@@ -17,12 +17,12 @@
 from functools import cached_property
 from typing import overload
 
-from dimos.messages.sensor_msgs.msg import Image
 from PIL import Image as PILImage
 import torch
 import torch.nn.functional as functional
 from transformers import SiglipModel as HFSiglipModel, SiglipProcessor
 
+from dimos.messages.sensor_msgs.msg import Image
 from dimos.models.base import HuggingFaceModel
 from dimos.models.embedding.base import Embedding, EmbeddingModel, HuggingFaceEmbeddingModelConfig
 from dimos.msgs.image import image_to_rgb

dimos/models/embedding/test_generated_images.py

@@ -29,20 +29,25 @@
 from dimos.msgs.image import image_from_array, image_view
 
 
-@pytest.mark.parametrize("model_type", [CLIPModel,DINOModel,SigLIPModel,SigLIP2Model])
-def test_generated_embeddings_keep_zero_source_time_and_convert_rgb(mocker,model_type):
-    processor,network=mocker.Mock(),mocker.Mock()
-    features=torch.tensor([[3.0,4.0]])
-    processor.return_value=BatchFeature(data={"pixel_values":torch.zeros(1,3,4,8)})
-    network.get_image_features.return_value=features
-    network.return_value=BaseModelOutputWithPooling(last_hidden_state=features[:,None,:],pooler_output=features)
-    mocker.patch.object(model_type,"_processor",new_callable=PropertyMock,return_value=processor)
-    mocker.patch.object(model_type,"_model",new_callable=PropertyMock,return_value=network)
-    image=image_from_array(np.full((4,8,3),[0,0,255],dtype=np.uint8),encoding="bgr8",
-        header=Header(frame_id="camera",stamp=Time(sec=0,nanosec=0)))
-    model=model_type(device="cpu")
-    result=model.embed(image)
+@pytest.mark.parametrize("model_type", [CLIPModel, DINOModel, SigLIPModel, SigLIP2Model])
+def test_generated_embeddings_keep_zero_source_time_and_convert_rgb(mocker, model_type):
+    processor, network = mocker.Mock(), mocker.Mock()
+    features = torch.tensor([[3.0, 4.0]])
+    processor.return_value = BatchFeature(data={"pixel_values": torch.zeros(1, 3, 4, 8)})
+    network.get_image_features.return_value = features
+    network.return_value = BaseModelOutputWithPooling(
+        last_hidden_state=features[:, None, :], pooler_output=features
+    )
+    mocker.patch.object(model_type, "_processor", new_callable=PropertyMock, return_value=processor)
+    mocker.patch.object(model_type, "_model", new_callable=PropertyMock, return_value=network)
+    image = image_from_array(
+        np.full((4, 8, 3), [0, 0, 255], dtype=np.uint8),
+        encoding="bgr8",
+        header=Header(frame_id="camera", stamp=Time(sec=0, nanosec=0)),
+    )
+    model = model_type(device="cpu")
+    result = model.embed(image)
     assert result.timestamp == 0.0
-    torch.testing.assert_close(result.vector,torch.tensor([0.6,0.8]))
-    assert processor.call_args.kwargs["images"][0].getpixel((0,0)) == (255,0,0)
-    np.testing.assert_array_equal(image_view(image)[0,0],[0,0,255])
+    torch.testing.assert_close(result.vector, torch.tensor([0.6, 0.8]))
+    assert processor.call_args.kwargs["images"][0].getpixel((0, 0)) == (255, 0, 0)
+    np.testing.assert_array_equal(image_view(image)[0, 0], [0, 0, 255])

dimos/models/embedding/treid.py

@@ -19,11 +19,11 @@
 
 from functools import cached_property
 
-from dimos.messages.sensor_msgs.msg import Image
 import torch
 import torch.nn.functional as functional
 from torchreid import utils as torchreid_utils
 
+from dimos.messages.sensor_msgs.msg import Image
 from dimos.models.base import LocalModel
 from dimos.models.embedding.base import Embedding, EmbeddingModel, EmbeddingModelConfig
 from dimos.msgs.image import image_to_bgr

dimos/perception/experimental/perceive_loop_skill.py

@@ -14,19 +14,27 @@
 
 from __future__ import annotations
 
+import base64
 from datetime import datetime, timezone
 import json
 import os
 from threading import RLock
 from typing import TYPE_CHECKING, Any
 
+import reactivex as rx
+from reactivex import Observable, operators as ops
+from reactivex.abc import SchedulerBase
+
 from dimos.agents.agent_spec import AgentSpec
 from dimos.agents.annotation import skill
 from dimos.core.core import rpc
 from dimos.core.module import Module
 from dimos.core.stream import In
+from dimos.messages.sensor_msgs.msg import Image
 from dimos.models.vl.create import create
-from dimos.msgs.sensor_msgs.Image import Image, sharpness_window
+from dimos.msgs.image import image_sharpness, image_to_bgr, image_to_jpeg
+from dimos.msgs.time import to_seconds
+from dimos.types.timestamped import TimestampedBufferCollection, TimestampedData
 from dimos.utils.logging_config import setup_logger
 from dimos.utils.reactive import backpressure
 from dimos.utils.threadpool import run_in_thread
@@ -41,6 +49,25 @@
 logger = setup_logger()
 
 
+def _sharpness_window(
+    frequency: float, source: Observable[Image], *, scheduler: SchedulerBase | None = None
+) -> Observable[Image]:
+    """Keep the existing source-time rolling selection and repeated timer emissions."""
+    if frequency <= 0:
+        raise ValueError("frequency must be positive")
+    window: TimestampedBufferCollection[TimestampedData[Image]] = TimestampedBufferCollection(
+        1 / frequency
+    )
+
+    def remember(image: Image) -> None:
+        window.add(TimestampedData(image, to_seconds(image.header.stamp)))
+
+    return rx.interval(1 / frequency, scheduler=scheduler).pipe(
+        ops.with_latest_from(source.pipe(ops.do_action(remember))),
+        ops.map(lambda _: max(window, key=lambda item: image_sharpness(item.value)).value),
+    )
+
+
 class PerceiveLoopSkill(Module):
     color_image: In[Image]
 
@@ -114,7 +141,7 @@ def look_out_for(
 
             model = self._vl_model.result()
             sharpest = backpressure(
-                sharpness_window(1.0 / self._period, self.color_image.pure_observable())
+                _sharpness_window(1.0 / self._period, self.color_image.pure_observable())
             )
             model.start()
             self._model_started = True
@@ -181,7 +208,7 @@ def _on_image(self, image: Image) -> None:
         continuation_context: dict[str, Any] = {
             "bbox": list(best.bbox),
             "label": best.name,
-            "image": image.to_base64(quality=70),
+            "image": base64.b64encode(image_to_jpeg(image, quality=70)).decode("ascii"),
         }
         logger.info(
             "Lookout matched, dispatching continuation",
@@ -208,7 +235,7 @@ def _write_debug_image(image: Image, detections: ImageDetections2D[Detection2DBB
     import cv2
 
     try:
-        debug_img = image.to_opencv().copy()
+        debug_img = image_to_bgr(image).copy()
         for det in detections.detections:
             x1, y1, x2, y2 = (int(v) for v in det.bbox)
             cv2.rectangle(debug_img, (x1, y1), (x2, y2), (0, 255, 0), 2)

dimos/perception/experimental/temporal_memory/clip_filter.py

@@ -18,7 +18,8 @@
 
 import numpy as np
 
-from dimos.msgs.sensor_msgs.Image import Image
+from dimos.messages.sensor_msgs.msg import Image
+from dimos.msgs.image import image_to_rgb
 from dimos.utils.logging_config import setup_logger
 
 logger = setup_logger()
@@ -36,9 +37,7 @@
 
 def _get_image_data(image: Image) -> np.ndarray[Any, Any]:
     """Extract numpy array from Image."""
-    if not hasattr(image, "data"):
-        raise AttributeError(f"Image missing .data attribute: {type(image)}")
-    return image.data
+    return image_to_rgb(image)
 
 
 if CLIP_AVAILABLE:

dimos/perception/experimental/temporal_memory/entity_graph_db.py

@@ -37,8 +37,8 @@
 from dimos.utils.logging_config import setup_logger
 
 if TYPE_CHECKING:
+    from dimos.messages.sensor_msgs.msg import Image
     from dimos.models.vl.base import VlModel
-    from dimos.msgs.sensor_msgs.Image import Image
 
 logger = setup_logger()
 

dimos/perception/experimental/temporal_memory/frame_window_accumulator.py

@@ -24,8 +24,10 @@
 import threading
 from typing import TYPE_CHECKING
 
+from dimos.msgs.time import to_seconds
+
 if TYPE_CHECKING:
-    from dimos.msgs.sensor_msgs.Image import Image
+    from dimos.messages.sensor_msgs.msg import Image
 
 
 @dataclass
@@ -87,10 +89,7 @@ def add_frame(self, image: Image, wall_time: float) -> None:
         with self._lock:
             if self._video_start_wall_time is None:
                 return
-            if image.ts is not None:
-                timestamp_s = image.ts - self._video_start_wall_time
-            else:
-                timestamp_s = wall_time - self._video_start_wall_time
+            timestamp_s = to_seconds(image.header.stamp) - self._video_start_wall_time
             frame = Frame(
                 frame_index=self._frame_count,
                 timestamp_s=timestamp_s,

dimos/perception/experimental/temporal_memory/temporal_memory.py

@@ -37,10 +37,12 @@
 from dimos.core.core import rpc
 from dimos.core.module import Module, ModuleConfig
 from dimos.core.stream import In, Out
+from dimos.messages.dimos_msgs.msg import EntityMarker, EntityMarkers
+from dimos.messages.geometry_msgs.msg import Point, PoseStamped
+from dimos.messages.sensor_msgs.msg import Image
 from dimos.models.vl.base import VlModel
-from dimos.msgs.geometry_msgs.PoseStamped import PoseStamped
-from dimos.msgs.sensor_msgs.Image import Image, sharpness_barrier
-from dimos.msgs.visualization_msgs.EntityMarkers import EntityMarkers, Marker
+from dimos.msgs.image import image_sharpness
+from dimos.msgs.time import header_now
 from dimos.perception.experimental.temporal_memory.clip_filter import (
     CLIP_AVAILABLE,
     adaptive_keyframes,
@@ -57,6 +59,7 @@
 from dimos.perception.experimental.temporal_memory.temporal_utils.helpers import is_scene_stale
 from dimos.perception.experimental.temporal_memory.window_analyzer import WindowAnalyzer
 from dimos.utils.logging_config import get_run_log_dir, setup_logger
+from dimos.utils.reactive import quality_barrier
 
 try:
     from dimos.perception.experimental.temporal_memory.clip_filter import CLIPFrameFilter
@@ -255,7 +258,7 @@ def _publish_entity_markers(self) -> None:
             if not all_entities:
                 return
 
-            markers: list[Marker] = []
+            markers: list[EntityMarker] = []
             for e in all_entities:
                 meta = e.get("metadata") or {}
                 x = meta.get("world_x")
@@ -264,18 +267,18 @@ def _publish_entity_markers(self) -> None:
                 if x is None or y is None:
                     continue
                 markers.append(
-                    Marker(
+                    EntityMarker(
                         entity_id=e["entity_id"],
                         label=(e.get("descriptor") or "")[:40],
                         entity_type=e.get("entity_type", "object"),
-                        x=x,
-                        y=y,
-                        z=(z or 0.0) + 0.3,  # Offset up so labels float above ground
+                        position=Point(x=x, y=y, z=(z or 0.0) + 0.3),  # Offset labels above ground
                     )
                 )
 
             if markers:
-                self.entity_markers.publish(EntityMarkers(markers=markers))
+                self.entity_markers.publish(
+                    EntityMarkers(header=header_now("world"), markers=markers)
+                )
                 logger.info(f"[temporal-memory] published {len(markers)} entity markers to Rerun")
         except Exception as e:
             logger.debug(f"entity marker publish error: {e}")
@@ -302,17 +305,19 @@ def _on_frame(img: Image) -> None:
                 )
 
         self.register_disposable(
-            frame_subject.pipe(sharpness_barrier(self.config.fps)).subscribe(_on_frame)
+            frame_subject.pipe(quality_barrier(image_sharpness, self.config.fps)).subscribe(
+                _on_frame
+            )
         )
         unsub_image = self.color_image.subscribe(frame_subject.on_next)
         self.register_disposable(Disposable(unsub_image))
 
         # Odometry tracking for entity world positioning (optional —
         # module works without it, entities just won't have world positions)
         def _on_odom(msg: PoseStamped) -> None:
-            self._robot_x = msg.position.x
-            self._robot_y = msg.position.y
-            self._robot_z = msg.position.z
+            self._robot_x = msg.pose.position.x
+            self._robot_y = msg.pose.position.y
+            self._robot_z = msg.pose.position.z
             self._odom_count += 1
 
         if self.odom.transport is not None:

dimos/perception/experimental/temporal_memory/temporal_utils/helpers.py

@@ -18,6 +18,8 @@
 
 import numpy as np
 
+from dimos.msgs.image import image_to_rgb
+
 if TYPE_CHECKING:
     from dimos.perception.experimental.temporal_memory.frame_window_accumulator import Frame
 
@@ -68,7 +70,9 @@ def is_scene_stale(frames: list["Frame"], stale_threshold: float = 5.0) -> bool:
     last_img = frames[-1].image
     if first_img is None or last_img is None:
         return False
-    if not hasattr(first_img, "data") or not hasattr(last_img, "data"):
+    first = image_to_rgb(first_img)
+    last = image_to_rgb(last_img)
+    if first.shape != last.shape:
         return False
-    diff = np.abs(first_img.data.astype(float) - last_img.data.astype(float))
+    diff = np.abs(first.astype(float) - last.astype(float))
     return bool(diff.mean() < stale_threshold)