RerunBridge module and CLI (#1154)

* feat: add RerunBridge module and CLI tool

- Add RerunBridgeModule that subscribes to all LCM messages and logs
  those with to_rerun() methods to Rerun viewer
- Add SubscribeAllCapable protocol to pubsub spec for type-safe
  subscribe_all support
- Add rerun-bridge CLI tool with --viewer-mode native|web|none
- Add "none" option to ViewerBackend for headless operation
- Add unitree_go2_bridge blueprint variant

* rerun bridge blueprint

* adding stop method to bridge

* working on rerun conversion to bridge

* small cleanup

* supress treid warning

* occupancygrid meshing rewrite

* bridge inline imports

* implemented to_observable for dimos standard callbacks

* settled on the temporary bridge api

* restructure of pubsub patterns, good bridge match api

* planner uses occupancygrid now, bridge improvements

* image pickle fix

* paul good vis

* renamed visual transforms to converters

* small cleamnup, pulled image from dev

* looks nice

* camerainfo knows how to render itself in the space

* typing for to_rerun calls

* utils type fix

* better transport comment

* base blueprints defined

* small restructure

* old rerun kicked out

* typing cleanup

* blueprints update

* type fixes, manip type ignores

* occupancygrid is never an image

* tf cleanup

* bridge comments, box for go2

* small comments

* turn off local rerun

* fix mypy type errors in CameraInfo and drake_world

* global config singleton, web view works

* deleted base_blueprints

* type fixes

* rerun bridge typer CLI

* Rename globalconfig singleton to global_config

Use the singleton as the default kwarg value directly, eliminating
the `global_config or globalconfig` fallback pattern.

* Replace @cached_property with @Property in GlobalConfig

Removes manual cache invalidation in update() since properties
are now recomputed on access. Prevents stale cache bugs when
new derived properties are added.

* Rename debug_navigation to navigation_costmap

Keep DEBUG_NAVIGATION env var gate as-is.

* Fix broken rerun_init import, rename debug_navigation to navigation_costmap

- Move RERUN_GRPC_PORT/RERUN_WEB_PORT constants to bridge.py,
  fixing import of deleted dimos.dashboard.rerun_init
- Rename debug_navigation to navigation_costmap throughout the
  navigation stack and visual overrides
- Remove redundant __init__ in RerunBridgeModule

* test_blueprints fix

* removed unused rerun config keys

* Remove broken rerun_init import from teleop visualization

RerunBridge handles initialization; teleop modules just need to log.

Author: leshy

dimos/core/blueprints.py

@@ -23,10 +23,7 @@
 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.global_config import GlobalConfig, global_config
 from dimos.core.module import Module, is_module_type
 from dimos.core.module_coordinator import ModuleCoordinator
 from dimos.core.stream import In, Out
@@ -262,8 +259,8 @@ def _deploy_all_modules(
         for blueprint in self.blueprints:
             kwargs = {**blueprint.kwargs}
             sig = inspect.signature(blueprint.module.__init__)
-            if "global_config" in sig.parameters:
-                kwargs["global_config"] = global_config
+            if "cfg" in sig.parameters:
+                kwargs["cfg"] = global_config
             module_specs.append((blueprint.module, blueprint.args, kwargs))
 
         module_coordinator.deploy_parallel(module_specs)
@@ -456,78 +453,18 @@ 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 _start_rerun(self, global_config: GlobalConfig) -> None:
-        # 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.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}")
-
     def build(
         self,
-        global_config: GlobalConfig | None = None,
         cli_config_overrides: Mapping[str, Any] | None = None,
     ) -> ModuleCoordinator:
-        if global_config is None:
-            global_config = GlobalConfig()
-        global_config = global_config.model_copy(update=dict(self.global_config_overrides))
+        global_config.update(**dict(self.global_config_overrides))
         if cli_config_overrides:
-            global_config = global_config.model_copy(update=dict(cli_config_overrides))
+            global_config.update(**dict(cli_config_overrides))
 
         self._check_requirements()
         self._verify_no_name_conflicts()
-        self._start_rerun(global_config)
 
-        module_coordinator = ModuleCoordinator(global_config=global_config)
+        module_coordinator = ModuleCoordinator(cfg=global_config)
         module_coordinator.start()
 
         # all module constructors are called here (each of them setup their own)
@@ -538,10 +475,6 @@ 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/docker_runner.py

@@ -28,9 +28,9 @@
 from dimos.core.docker_build import build_image, image_exists
 from dimos.core.module import Module, ModuleConfig
 from dimos.core.rpc_client import RpcCall
-from dimos.dashboard.rerun_init import RERUN_GRPC_PORT, RERUN_WEB_PORT
 from dimos.protocol.rpc import LCMRPC
 from dimos.utils.logging_config import setup_logger
+from dimos.visualization.rerun.bridge import RERUN_GRPC_PORT, RERUN_WEB_PORT
 
 if TYPE_CHECKING:
     from collections.abc import Callable

dimos/core/global_config.py

@@ -12,15 +12,14 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-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
 
-ViewerBackend: TypeAlias = Literal["rerun-web", "rerun-native", "foxglove"]
+ViewerBackend: TypeAlias = Literal["rerun", "rerun-web", "foxglove", "none"]
 
 
 def _get_all_numbers(s: str) -> list[float]:
@@ -31,9 +30,7 @@ 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"
+    viewer_backend: ViewerBackend = "rerun"
     n_dask_workers: int = 2
     memory_limit: str = "auto"
     mujoco_camera_position: str | None = None
@@ -54,24 +51,33 @@ class GlobalConfig(BaseSettings):
         env_file=".env",
         env_file_encoding="utf-8",
         extra="ignore",
-        frozen=True,
     )
 
-    @cached_property
+    def update(self, **kwargs: object) -> None:
+        """Update config fields in place."""
+        for key, value in kwargs.items():
+            if not hasattr(self, key):
+                raise AttributeError(f"GlobalConfig has no field '{key}'")
+            setattr(self, key, value)
+
+    @property
     def unitree_connection_type(self) -> str:
         if self.replay:
             return "replay"
         if self.simulation:
             return "mujoco"
         return "webrtc"
 
-    @cached_property
+    @property
     def mujoco_start_pos_float(self) -> tuple[float, float]:
         x, y = _get_all_numbers(self.mujoco_start_pos)
         return (x, y)
 
-    @cached_property
+    @property
     def mujoco_camera_position_float(self) -> tuple[float, ...]:
         if self.mujoco_camera_position is None:
             return (-0.906, 0.008, 1.101, 4.931, 89.749, -46.378)
         return tuple(_get_all_numbers(self.mujoco_camera_position))
+
+
+global_config = GlobalConfig()

dimos/core/module_coordinator.py

@@ -18,7 +18,7 @@
 
 from dimos import core
 from dimos.core import DimosCluster
-from dimos.core.global_config import GlobalConfig
+from dimos.core.global_config import GlobalConfig, global_config
 from dimos.core.module import Module, ModuleT
 from dimos.core.resource import Resource
 from dimos.core.worker_manager import WorkerManager
@@ -37,9 +37,8 @@ class ModuleCoordinator(Resource):  # type: ignore[misc]
     def __init__(
         self,
         n: int | None = None,
-        global_config: GlobalConfig | None = None,
+        cfg: GlobalConfig = global_config,
     ) -> None:
-        cfg = global_config or GlobalConfig()
         self._n = n if n is not None else cfg.n_dask_workers
         self._memory_limit = cfg.memory_limit
         self._global_config = cfg

dimos/core/test_blueprints.py

@@ -28,7 +28,6 @@
     autoconnect,
 )
 from dimos.core.core import rpc
-from dimos.core.global_config import GlobalConfig
 from dimos.core.module import Module
 from dimos.core.module_coordinator import ModuleCoordinator
 from dimos.core.rpc_client import RpcCall
@@ -40,7 +39,7 @@
 
 # Disable Rerun for tests (prevents viewer spawn and gRPC flush errors)
 _BUILD_WITHOUT_RERUN = {
-    "global_config": GlobalConfig(rerun_enabled=False, viewer_backend="foxglove"),
+    "cli_config_overrides": {"viewer_backend": "none"},
 }
 
 

dimos/core/transport.py

@@ -37,6 +37,26 @@
 
 T = TypeVar("T")  # type: ignore[misc]
 
+# TODO
+# Transports need to be rewritten and simplified,
+#
+# there is no need for them to get a reference to "a stream" on publish/subscribe calls
+# this is a legacy from dask transports.
+#
+# new transport should literally have 2 functions (next to start/stop)
+# "send(msg)" and "receive(callback)" and that's all
+#
+# we can also consider pubsubs conforming directly to Transport specs
+# and removing PubSubTransport glue entirely
+#
+# Why not ONLY pubsubs without Transport abstraction?
+#
+# General idea for transports (and why they exist at all)
+# is that they can be * anything * like
+#
+# a web camera rtsp stream for Image, audio stream from mic, etc
+# http binary streams, tcp connections etc
+
 
 class PubSubTransport(Transport[T]):
     topic: Any