Merge pull request #917 from dimensionalOS/feat/rerun-latency-panels

Feat/rerun latency panels

Former-commit-id: a3a03b3 [formerly 72935f8]
Former-commit-id: 41a54dc

Author: Nabla7

dimos/core/blueprints.py

@@ -23,6 +23,9 @@
 from types import MappingProxyType
 from typing import Any, Literal, get_args, get_origin, get_type_hints
 
+import rerun as rr
+import rerun.blueprint as rrb
+
 from dimos.core.global_config import GlobalConfig
 from dimos.core.module import Module
 from dimos.core.module_coordinator import ModuleCoordinator
@@ -280,6 +283,50 @@ def _connect_rpc_methods(self, module_coordinator: ModuleCoordinator) -> None:
                     requested_method_name, rpc_methods_dot[requested_method_name]
                 )
 
+    def _init_rerun_blueprint(self, module_coordinator: ModuleCoordinator) -> None:
+        """Compose and send Rerun blueprint from module contributions.
+
+        Collects rerun_views() from all modules and composes them into a unified layout.
+        """
+        # Collect view contributions from all modules
+        side_panels = []
+        for blueprint in self.blueprints:
+            if hasattr(blueprint.module, "rerun_views"):
+                views = blueprint.module.rerun_views()
+                if views:
+                    side_panels.extend(views)
+
+        # Always include latency panel if we have any panels
+        if side_panels:
+            side_panels.append(
+                rrb.TimeSeriesView(
+                    name="Latency (ms)",
+                    origin="/metrics",
+                    contents=[
+                        "+ /metrics/voxel_map/latency_ms",
+                        "+ /metrics/costmap/latency_ms",
+                    ],
+                )
+            )
+
+        # Compose final layout
+        if side_panels:
+            composed_blueprint = rrb.Blueprint(
+                rrb.Horizontal(
+                    rrb.Spatial3DView(
+                        name="3D View",
+                        origin="world",
+                        background=[0, 0, 0],
+                    ),
+                    rrb.Vertical(*side_panels, row_shares=[2] + [1] * (len(side_panels) - 1)),
+                    column_shares=[3, 1],
+                ),
+                rrb.TimePanel(state="collapsed"),
+                rrb.SelectionPanel(state="collapsed"),
+                rrb.BlueprintPanel(state="collapsed"),
+            )
+            rr.send_blueprint(composed_blueprint)
+
     def build(
         self,
         global_config: GlobalConfig | None = None,
@@ -294,6 +341,17 @@ def build(
         self._check_requirements()
         self._verify_no_name_conflicts()
 
+        # Initialize Rerun server before deploying modules (if backend is Rerun)
+        if global_config.rerun_enabled and global_config.viewer_backend.startswith("rerun"):
+            try:
+                from dimos.dashboard.rerun_init import init_rerun_server
+
+                server_addr = init_rerun_server(viewer_mode=global_config.viewer_backend)
+                global_config = global_config.model_copy(update={"rerun_server_addr": server_addr})
+                logger.info("Rerun server initialized", addr=server_addr)
+            except Exception as e:
+                logger.warning(f"Failed to initialize Rerun server: {e}")
+
         module_coordinator = ModuleCoordinator(global_config=global_config)
         module_coordinator.start()
 
@@ -303,6 +361,10 @@ def build(
 
         module_coordinator.start_all_modules()
 
+        # Compose and send Rerun blueprint from module contributions
+        if global_config.viewer_backend.startswith("rerun"):
+            self._init_rerun_blueprint(module_coordinator)
+
         return module_coordinator
 
 

dimos/core/global_config.py

@@ -14,12 +14,15 @@
 
 from functools import cached_property
 import re
+from typing import Literal, TypeAlias
 
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 from dimos.mapping.occupancy.path_map import NavigationStrategy
 from dimos.navigation.global_planner.types import AStarAlgorithm
 
+ViewerBackend: TypeAlias = Literal["rerun-web", "rerun-native", "foxglove"]
+
 
 def _get_all_numbers(s: str) -> list[float]:
     return [float(x) for x in re.findall(r"-?\d+\.?\d*", s)]
@@ -29,6 +32,9 @@ class GlobalConfig(BaseSettings):
     robot_ip: str | None = None
     simulation: bool = False
     replay: bool = False
+    rerun_enabled: bool = True
+    rerun_server_addr: str | None = None
+    viewer_backend: ViewerBackend = "rerun-native"
     n_dask_workers: int = 2
     memory_limit: str = "auto"
     mujoco_camera_position: str | None = None

dimos/core/stream.py

@@ -26,6 +26,7 @@
 import reactivex as rx
 from reactivex import operators as ops
 from reactivex.disposable import Disposable
+import rerun as rr
 
 import dimos.core.colors as colors
 from dimos.core.resource import Resource
@@ -137,20 +138,21 @@ def __str__(self) -> str:
         )
 
 
-class Out(Stream[T]):
+class Out(Stream[T], ObservableMixin[T]):
     _transport: Transport  # type: ignore[type-arg]
 
     def __init__(self, *argv, **kwargs) -> None:  # type: ignore[no-untyped-def]
         super().__init__(*argv, **kwargs)
+        self._rerun_config: dict | None = None  # type: ignore[type-arg]
+        self._rerun_last_log: float = 0.0
 
     @property
     def transport(self) -> Transport[T]:
         return self._transport
 
     @transport.setter
     def transport(self, value: Transport[T]) -> None:
-        # just for type checking
-        ...
+        self._transport = value
 
     @property
     def state(self) -> State:
@@ -173,8 +175,76 @@ def publish(self, msg) -> None:  # type: ignore[no-untyped-def]
         if not hasattr(self, "_transport") or self._transport is None:
             logger.warning(f"Trying to publish on Out {self} without a transport")
             return
+
+        # Log to Rerun directly if configured
+        if self._rerun_config is not None:
+            self._log_to_rerun(msg)
+
         self._transport.broadcast(self, msg)
 
+    def subscribe(self, cb) -> Callable[[], None]:  # type: ignore[no-untyped-def]
+        """Subscribe to this output stream.
+
+        Args:
+            cb: Callback function to receive messages
+
+        Returns:
+            Unsubscribe function
+        """
+        return self.transport.subscribe(cb, self)  # type: ignore[arg-type, func-returns-value, no-any-return]
+
+    def autolog_to_rerun(
+        self,
+        entity_path: str,
+        rate_limit: float | None = None,
+        **rerun_kwargs: Any,
+    ) -> None:
+        """Configure this output to auto-log to Rerun (fire-and-forget).
+
+        Call once in start() - messages auto-logged when published.
+
+        Args:
+            entity_path: Rerun entity path (e.g., "world/map")
+            rate_limit: Max Hz to log (None = unlimited)
+            **rerun_kwargs: Passed to msg.to_rerun() for rendering config
+                           (e.g., radii=0.02, colormap="turbo", colors=[255,0,0])
+
+        Example:
+            def start(self):
+                super().start()
+                # Just declare it - fire and forget!
+                self.global_map.autolog_to_rerun("world/map", rate_limit=5.0, radii=0.02)
+        """
+        self._rerun_config = {
+            "entity_path": entity_path,
+            "rate_limit": rate_limit,
+            "rerun_kwargs": rerun_kwargs,
+        }
+        self._rerun_last_log = 0.0
+
+    def _log_to_rerun(self, msg: T) -> None:
+        """Log message to Rerun with rate limiting."""
+        if not hasattr(msg, "to_rerun"):
+            return
+
+        if self._rerun_config is None:
+            return
+
+        import time
+
+        config = self._rerun_config
+
+        # Rate limiting
+        if config["rate_limit"] is not None:
+            now = time.monotonic()
+            min_interval = 1.0 / config["rate_limit"]
+            if now - self._rerun_last_log < min_interval:
+                return  # Skip - too soon
+            self._rerun_last_log = now
+
+        rerun_data = msg.to_rerun(**config["rerun_kwargs"])
+        rr.log(config["entity_path"], rerun_data)
+
 
 class RemoteStream(Stream[T]):
     @property

dimos/dashboard/__init__.py

@@ -0,0 +1,34 @@
+# Copyright 2025 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.
+
+"""Dashboard module for visualization and monitoring.
+
+Rerun Initialization:
+    Main process (e.g., blueprints.build) starts Rerun server automatically.
+    Worker modules connect to the server via connect_rerun().
+
+Usage in modules:
+    import rerun as rr
+    from dimos.dashboard.rerun_init import connect_rerun
+
+    class MyModule(Module):
+        def start(self):
+            super().start()
+            connect_rerun()  # Connect to Rerun server
+            rr.log("my/entity", my_data.to_rerun())
+"""
+
+from dimos.dashboard.rerun_init import connect_rerun, init_rerun_server, shutdown_rerun
+
+__all__ = ["connect_rerun", "init_rerun_server", "shutdown_rerun"]