Unified TimeSeriesStore with pluggable backends, global rewrite of timed event storage

.gitignore

@@ -66,4 +66,5 @@ yolo11n.pt
 *mobileclip*
 /results
 
+CLAUDE.MD
 /assets/teleop_certs/

dimos/core/stream.py

@@ -69,7 +69,7 @@ def _subscribe(observer, scheduler=None):  # type: ignore[no-untyped-def]
 
     # default return is backpressured because most
     # use cases will want this by default
-    def observable(self):  # type: ignore[no-untyped-def]
+    def observable(self) -> Observable[T]:
         return backpressure(self.pure_observable())
 
 

dimos/core/testing.py

@@ -79,6 +79,5 @@ def odomloop(self) -> None:
                 self.odometry.publish(odom)
 
                 lidarmsg = next(lidariter)
-                lidarmsg.pubtime = time.perf_counter()  # type: ignore[union-attr]
                 self.lidar.publish(lidarmsg)
                 time.sleep(0.1)

dimos/manipulation/planning/utils/mesh_utils.py

@@ -133,7 +133,7 @@ def _process_xacro(
 ) -> str:
     """Process xacro file to URDF."""
     try:
-        import xacro  # type: ignore[import-not-found]
+        import xacro  # type: ignore[import-not-found,import-untyped]
     except ImportError:
         raise ImportError(
             "xacro is required for processing .xacro files. Install with: pip install xacro"

dimos/mapping/pointclouds/test_occupancy_speed.py

@@ -20,19 +20,18 @@
 from dimos.mapping.pointclouds.occupancy import OCCUPANCY_ALGOS
 from dimos.mapping.voxels import VoxelGridMapper
 from dimos.utils.cli.plot import bar
-from dimos.utils.data import _get_data_dir, get_data
+from dimos.utils.data import get_data, get_data_dir
 from dimos.utils.testing import TimedSensorReplay
 
 
 @pytest.mark.tool
 def test_build_map():
     mapper = VoxelGridMapper(publish_interval=-1)
 
-    for ts, frame in TimedSensorReplay("unitree_go2_bigoffice/lidar").iterate_duration():
-        print(ts, frame)
+    for _ts, frame in TimedSensorReplay("unitree_go2_bigoffice/lidar").iterate():
         mapper.add_frame(frame)
 
-    pickle_file = _get_data_dir() / "unitree_go2_bigoffice_map.pickle"
+    pickle_file = get_data_dir() / "unitree_go2_bigoffice_map.pickle"
     global_pcd = mapper.get_global_pointcloud2()
 
     with open(pickle_file, "wb") as f:

dimos/memory/embedding.py

@@ -0,0 +1,104 @@
+# 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.
+
+from collections.abc import Callable
+from dataclasses import dataclass, field
+from typing import cast
+
+import reactivex as rx
+from reactivex import operators as ops
+from reactivex.observable import Observable
+
+from dimos.core import In, Module, ModuleConfig, rpc
+from dimos.models.embedding.base import Embedding, EmbeddingModel
+from dimos.models.embedding.clip import CLIPModel
+from dimos.msgs.geometry_msgs import PoseStamped
+from dimos.msgs.nav_msgs import OccupancyGrid
+from dimos.msgs.sensor_msgs import Image
+from dimos.msgs.sensor_msgs.Image import Image, sharpness_barrier
+from dimos.utils.reactive import getter_hot
+
+
+@dataclass
+class Config(ModuleConfig):
+    embedding_model: EmbeddingModel = field(default_factory=CLIPModel)
+
+
+@dataclass
+class SpatialEntry:
+    image: Image
+    pose: PoseStamped
+
+
+@dataclass
+class SpatialEmbedding(SpatialEntry):
+    embedding: Embedding
+
+
+class EmbeddingMemory(Module[Config]):
+    default_config = Config
+    config: Config
+    color_image: In[Image]
+    global_costmap: In[OccupancyGrid]
+
+    _costmap_getter: Callable[[], OccupancyGrid] | None = None
+
+    def get_costmap(self) -> OccupancyGrid:
+        if self._costmap_getter is None:
+            self._costmap_getter = getter_hot(self.global_costmap.pure_observable())
+            self._disposables.add(self._costmap_getter)
+        return self._costmap_getter()
+
+    @rpc
+    def query_costmap(self, text: str) -> OccupancyGrid:
+        costmap = self.get_costmap()
+        # overlay costmap with embedding heat
+        return costmap
+
+    @rpc
+    def start(self) -> None:
+        # would be cool if this sharpness_barrier was somehow self-calibrating
+        #
+        # we need a Governor system, sharpness_barrier frequency shouldn't
+        # be a fixed float but an observable that adjusts based on downstream load
+        #
+        # (also voxel size for mapper for example would benefit from this)
+        self.color_image.pure_observable().pipe(
+            sharpness_barrier(0.5),
+            ops.flat_map(self._try_create_spatial_entry),
+            ops.map(self._embed_spatial_entry),
+            ops.map(self._store_spatial_entry),
+        ).subscribe(print)
+
+    def _try_create_spatial_entry(self, img: Image) -> Observable[SpatialEntry]:
+        pose = self.tf.get_pose("world", "base_link")
+        if not pose:
+            return rx.empty()
+        return rx.of(SpatialEntry(image=img, pose=pose))
+
+    def _embed_spatial_entry(self, spatial_entry: SpatialEntry) -> SpatialEmbedding:
+        embedding = cast("Embedding", self.config.embedding_model.embed(spatial_entry.image))
+        return SpatialEmbedding(
+            image=spatial_entry.image,
+            pose=spatial_entry.pose,
+            embedding=embedding,
+        )
+
+    def _store_spatial_entry(self, spatial_embedding: SpatialEmbedding) -> SpatialEmbedding:
+        return spatial_embedding
+
+    def query_text(self, query: str) -> list[SpatialEmbedding]:
+        self.config.embedding_model.embed_text(query)
+        results: list[SpatialEmbedding] = []
+        return results

dimos/memory/test_embedding.py

@@ -0,0 +1,53 @@
+# 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.
+
+import pytest
+
+from dimos.memory.embedding import EmbeddingMemory, SpatialEntry
+from dimos.msgs.geometry_msgs import PoseStamped
+from dimos.utils.data import get_data
+from dimos.utils.testing import TimedSensorReplay
+
+dir_name = "unitree_go2_bigoffice"
+
+
+@pytest.mark.skip
+def test_embed_frame() -> None:
+    """Test embedding a single frame."""
+    # Load a frame from recorded data
+    video = TimedSensorReplay(get_data(dir_name) / "video")
+    frame = video.find_closest_seek(10)
+
+    # Create memory and embed
+    memory = EmbeddingMemory()
+
+    try:
+        # Create a spatial entry with dummy pose (no TF needed for this test)
+        dummy_pose = PoseStamped(
+            position=[0, 0, 0],
+            orientation=[0, 0, 0, 1],  # identity quaternion
+        )
+        spatial_entry = SpatialEntry(image=frame, pose=dummy_pose)
+
+        # Embed the frame
+        result = memory._embed_spatial_entry(spatial_entry)
+
+        # Verify
+        assert result is not None
+        assert result.embedding is not None
+        assert result.embedding.vector is not None
+        print(f"Embedding shape: {result.embedding.vector.shape}")
+        print(f"Embedding vector (first 5): {result.embedding.vector[:5]}")
+    finally:
+        memory.stop()

dimos/memory/timeseries/__init__.py

@@ -0,0 +1,41 @@
+# 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.
+"""Time series storage and replay."""
+
+from dimos.memory.timeseries.base import TimeSeriesStore
+from dimos.memory.timeseries.inmemory import InMemoryStore
+from dimos.memory.timeseries.pickledir import PickleDirStore
+from dimos.memory.timeseries.sqlite import SqliteStore
+
+
+def __getattr__(name: str):  # type: ignore[no-untyped-def]
+    if name == "PostgresStore":
+        from dimos.memory.timeseries.postgres import PostgresStore
+
+        return PostgresStore
+    if name == "reset_db":
+        from dimos.memory.timeseries.postgres import reset_db
+
+        return reset_db
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+
+
+__all__ = [
+    "InMemoryStore",
+    "PickleDirStore",
+    "PostgresStore",
+    "SqliteStore",
+    "TimeSeriesStore",
+    "reset_db",
+]