feat: --record flag, mcap recording by default

README.md

@@ -191,7 +191,7 @@ dimos run unitree-go2
 | Run command | What it does |
 |-------------|-------------|
 | `dimos --replay run unitree-go2` | Quadruped navigation replay — SLAM, costmap, A* planning |
-| `dimos --replay --replay-db go2_bigoffice run unitree-go2-memory` | Quadruped temporal memory replay |
+| `dimos --replay --replay-db go2_bigoffice run unitree-go2` | Quadruped temporal memory replay |
 | `dimos --simulation run unitree-go2-agentic` | Quadruped agentic + MCP server in simulation |
 | `dimos --simulation run unitree-g1-sim` | Humanoid in MuJoCo simulation |
 | `dimos --replay run drone-basic` | Drone video + telemetry replay |

dimos/cli/commands/lifecycle.py

@@ -27,7 +27,7 @@
 
 import typer
 
-from dimos.constants import CONFIG_DIR, LOG_DIR
+from dimos.constants import CONFIG_DIR, LOG_DIR, RECORDINGS_DIR
 from dimos.core.daemon import (
     fork_daemon,
     install_signal_handlers,
@@ -149,6 +149,9 @@ def run(
     except BlueprintConfigError as error:
         typer.echo(f"Error: {error}", err=True)
         raise typer.Exit(2) from error
+    blueprint_name = "-".join(blueprint_names)
+    run_id = generate_run_id(blueprint_name)
+    preparsed_global_config.setdefault("recording_dir", str(RECORDINGS_DIR / run_id))
     # Some blueprint modules select their composition at import time, so all
     # global sources must be visible before resolving the requested names.
     global_config.update(**preparsed_global_config)
@@ -195,8 +198,6 @@ def run(
     if stale:
         logger.info(f"Cleaned {stale} stale run entries")
 
-    blueprint_name = "-".join(blueprint_names)
-    run_id = generate_run_id(blueprint_name)
     log_dir = LOG_DIR / run_id
 
     # Tag every descendant with the run id so the watchdog and stale-run

dimos/core/coordination/blueprint_config/test_parser.py

@@ -152,12 +152,17 @@ class CollisionModule(Module):
             "module-value",
             "--replay=false",
             "--no-obstacle-avoidance",
+            "--record",
+            "--record-format",
+            "sqlite",
         ],
         environ={},
     )
 
     assert parsed.global_config_values()["robot_ip"] == "192.0.2.10"
     assert parsed.global_config_values()["replay"] is False
+    assert parsed.global_config_values()["record"] is True
+    assert parsed.global_config_values()["record_format"] == "sqlite"
     assert parsed.global_config_values()["obstacle_avoidance"] is False
     assert parsed.module_kwargs("collisionmodule") == {"robot_ip": "module-value"}
 

dimos/core/global_config.py

@@ -60,6 +60,11 @@ class GlobalConfig(BaseSettings):
     simulation: str = ""
     replay: bool = False
     replay_db: str = "go2_short"
+    # --record: Recorders in the blueprint write under recording_dir (off unless set).
+    record: bool = False
+    record_format: Literal["mcap", "sqlite"] = "mcap"
+    # `dimos run` sets this to RECORDINGS_DIR/<run-id> (same id as the log dir).
+    recording_dir: str = ""
     new_memory: bool = False
     # How every zenoh session this process opens joins the network.
     zenoh_mode: ZenohProcessMode = "peer"

dimos/experimental/robot/bosdyn/spot/blueprints/spot_record.py

@@ -33,4 +33,4 @@
 from dimos.experimental.robot.bosdyn.spot.blueprints.spot import spot
 from dimos.experimental.robot.bosdyn.spot.recorder import SpotRecorder
 
-spot_record = autoconnect(spot, SpotRecorder.blueprint())
+spot_record = autoconnect(spot, SpotRecorder.blueprint()).global_config(record=True)

dimos/experimental/world_belief/xarm6_blueprint.py

@@ -134,4 +134,4 @@ def _rerun_blueprint() -> rrb.Blueprint:
         hardware=[_hw],
         tasks=[trajectory_task(_hw)],
     ),
-).global_config(n_workers=8)
+).global_config(n_workers=8, record=True)

dimos/hardware/sensors/lidar/fastlio2/tools/pcap_to_db.py

@@ -27,7 +27,7 @@
 
     # add to existing .db (a missing --db is fetched via get_data before falling
     # back to building from scratch)
-    DB="mem2.db"
+    DB="memory.db"
     python -m dimos.hardware.sensors.lidar.fastlio2.tools.pcap_to_db --db "$DB"  --pcap "$PCAP_PATH"
 
     # A quick-look <db>.rrd (aggregated world lidar + pose path) is written next
@@ -263,7 +263,7 @@ def _build_blueprint(
                 (FastLio2Recorder, "fastlio_lidar", "lidar"),
             ]
         )
-        .global_config(n_workers=4, robot_model="mid360_fastlio_pcap_to_db")
+        .global_config(n_workers=4, robot_model="mid360_fastlio_pcap_to_db", record=True)
     )
 
 

dimos/hardware/sensors/lidar/pointlio/scripts/pcap_to_db.py

@@ -23,7 +23,7 @@
 
     # add to existing .db (a missing --db is fetched via get_data before falling
     # back to building from scratch; a missing --pcap is likewise fetched)
-    DB="mem2.db"
+    DB="memory.db"
     python -m dimos.hardware.sensors.lidar.pointlio.scripts.pcap_to_db --db "$DB"  --pcap "$PCAP_PATH"
 
     # A quick-look <db>.rrd (aggregated world lidar + pose path) is written next
@@ -342,7 +342,7 @@ def _build_blueprint(
                 (PointlioRecorder, _LIDAR_STREAM, "lidar"),
             ]
         )
-        .global_config(n_workers=4, robot_model="mid360_pointlio_pcap_to_db")
+        .global_config(n_workers=4, robot_model="mid360_pointlio_pcap_to_db", record=True)
     )
 
 

dimos/imitation/collection/blueprint.py

@@ -63,7 +63,7 @@ def _camera_if_real() -> tuple[Blueprint, ...]:
     EpisodeMonitorModule.blueprint(),  # default button_map: toggle=B, discard=Y
     teleop_quest_xarm7,
     *_camera_if_real(),
-)
+).global_config(record=True)
 
 
 learning_collect_quest_piper = autoconnect(
@@ -75,4 +75,4 @@ def _camera_if_real() -> tuple[Blueprint, ...]:
     EpisodeMonitorModule.blueprint(),  # default button_map: toggle=B, discard=Y
     teleop_quest_piper,
     *_camera_if_real(),
-)
+).global_config(record=True)

dimos/memory/module.py

@@ -31,17 +31,22 @@
 from dimos.agents.annotation import skill
 from dimos.constants import DIMOS_PROJECT_ROOT, RECORDINGS_DIR
 from dimos.core.core import rpc
+from dimos.core.global_config import global_config
 from dimos.core.module import Module, ModuleConfig
 from dimos.core.stream import In
 from dimos.memory.embed import EmbedImages
+from dimos.memory.store.base import Store
+from dimos.memory.store.mcap import McapStore
 from dimos.memory.store.null import NullStore
 from dimos.memory.store.sqlite import SqliteStore
 from dimos.memory.stream import Stream
 from dimos.memory.transform import QualityWindow
 from dimos.memory.type.observation import EmbeddedObservation, Observation
 from dimos.models.embedding.base import EmbeddingModel
+from dimos.msgs.geometry_msgs.Pose import Pose
 from dimos.msgs.geometry_msgs.PoseStamped import PoseStamped
 from dimos.msgs.sensor_msgs.Image import Image
+from dimos.msgs.sensor_msgs.PointCloud2 import PointCloud2
 from dimos.msgs.tf2_msgs.TFMessage import TFMessage
 from dimos.utils.data import backup_file
 from dimos.utils.logging_config import setup_logger
@@ -50,7 +55,6 @@
     from reactivex.abc import DisposableBase
 
     from dimos.core.stream import Out
-    from dimos.msgs.geometry_msgs.Pose import Pose
 
 logger = setup_logger()
 
@@ -72,6 +76,15 @@ def default_recording_dir() -> Path:
     return RECORDINGS_DIR / stamp
 
 
+def recording_dir() -> Path:
+    """Where this run records (memory.*, pcaps, ...): ``--recording-dir``, else a fresh stamp."""
+    return (
+        Path(global_config.recording_dir)
+        if global_config.recording_dir
+        else default_recording_dir()
+    )
+
+
 def stream_to_port(stream: Stream[T], out: Out[T]) -> DisposableBase:
     """Forward each observation's ``data`` from *stream* to a Module ``Out`` port.
 
@@ -177,11 +190,13 @@ def stop(self) -> None:
 
 
 class MemoryModuleConfig(ModuleConfig):
-    db_path: str | Path = "recording.db"
+    db_path: str | Path | None = "recording.db"
 
     @field_validator("db_path", mode="before")
     @classmethod
-    def _resolve_path(cls, v: str | Path) -> Path:
+    def _resolve_path(cls, v: str | Path | None) -> Path | None:
+        if v is None:
+            return None
         p = Path(os.fspath(v))
         if not p.is_absolute():
             p = DIMOS_PROJECT_ROOT / p
@@ -196,17 +211,34 @@ class MemoryModule(Module):
     """
 
     config: MemoryModuleConfig
-    _store: SqliteStore | None = None
+    _store: Store | None = None
 
     @property
-    def store(self) -> SqliteStore:
+    def db_path(self) -> Path:
+        """``config.db_path``, defaulting to ``<recording_dir>/memory.<record-format>``."""
+        g = self.config.g
+        if self.config.db_path is None:
+            base = Path(g.recording_dir) if g.recording_dir else default_recording_dir()
+            p = base / "memory"
+        else:
+            p = Path(self.config.db_path)
+        if p.suffix:
+            return p
+        return p.with_suffix(".mcap" if g.record_format == "mcap" else ".db")
+
+    @property
+    def store(self) -> Store:
         if self._store is not None:
             return self._store
 
-        Path(self.config.db_path).parent.mkdir(parents=True, exist_ok=True)
-        self._store = self.register_disposable(
-            SqliteStore(path=str(self.config.db_path)),
-        )
+        path = self.db_path
+        path.parent.mkdir(parents=True, exist_ok=True)
+        store: Store
+        if path.suffix == ".mcap":
+            store = McapStore(path=str(path), mode="w")
+        else:
+            store = SqliteStore(path=str(path))
+        self._store = self.register_disposable(store)
         self._store.start()
         return self._store
 
@@ -281,7 +313,7 @@ class RecorderConfig(MemoryModuleConfig):
     root_frame: str = "world"
     default_frame_id: str = "base_link"
     tf_tolerance: float = 0.5
-    db_path: str | Path = "recording.db"
+    db_path: str | Path | None = None  # default: recording_dir()/memory.<record-format>
     # Also record the live tf stream (under "tf") alongside the In ports.
     record_tf: bool = True
     # Rename recorded streams: {port_name: db_stream_name}. Conceptually this is
@@ -347,9 +379,10 @@ def start(self) -> None:
         super().start()
 
         if self.config.g.replay:
-            logger.info(
-                "Replay mode active — Recorder disabled, leaving %s untouched", self.config.db_path
-            )
+            logger.info("Replay mode active — Recorder disabled")
+            return
+        if not self.config.g.record:
+            logger.info("Recording off — pass --record to write %s", self.db_path)
             return
 
         self._pose_setters = self._collect_pose_setters()
@@ -358,7 +391,7 @@ def start(self) -> None:
         # shouldn't need to know about files (SqliteStore specific), and
         # .live() subs need to know how to re-sub in case of a restart of
         # this module in a deployed blueprint.
-        db_path = Path(self.config.db_path)
+        db_path = self.db_path
         if db_path.exists():
             if self.config.on_existing is OnExisting.APPEND:
                 pass  # keep the db; _prepare_streams handles any per-stream replacement
@@ -484,3 +517,22 @@ def on_tf(msg: TFMessage) -> None:
                 pass
 
         self.register_disposable(Disposable(self.tf.subscribe(on_tf)))
+
+
+class OdomRecorder(Recorder):
+    """Records ``color_image``, ``lidar``, ``odom`` (+ tf), posing each frame at the latest odom."""
+
+    color_image: In[Image]
+    lidar: In[PointCloud2]
+    odom: In[PoseStamped]
+
+    _last_odom: Pose | None = None
+
+    @pose_setter_for("odom")
+    async def _odom_pose(self, msg: PoseStamped) -> Pose | None:
+        self._last_odom = msg
+        return self._last_odom
+
+    @pose_setter_for("lidar")
+    async def _lidar_pose(self, msg: PointCloud2) -> Pose | None:
+        return self._last_odom