Skip to content

feat: --record flag, mcap recording by default - #3706

Closed
spomichter wants to merge 3 commits into
mainfrom
feat/record-by-default
Closed

feat: --record flag, mcap recording by default#3706
spomichter wants to merge 3 commits into
mainfrom
feat/record-by-default

Conversation

@spomichter

@spomichter spomichter commented Aug 26, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Summary

dimos --record run <blueprint> records every Recorder in the blueprint to recordings/<run-id>/memory.mcap, e.g. recordings/20260826-165100-unitree-go2/memory.mcap (same run id as logs/). --record-format sqlite writes memory.db. Without --record, Recorders are inert. Recording infra is one system (dimos.memory); this PR makes the path and format one convention too.

  • GlobalConfig.record / record_format / recording_dir. Recorder.start() returns unless --record (replay still disables). dimos run sets recording_dir to RECORDINGS_DIR/<run-id>; --recording-dir overrides. Blueprints whose purpose is recording (*-record, learning-collect-*, pcap_to_db, spot-record, xarm6 world belief) set record=True themselves.
  • McapStore write mode. One channel per stream, message_encoding="dimos-obs": 4-byte header length + JSON {pose, tags} + codec payload. Schema name = payload type path, schema data = codec id, so a recorded file reads back with no robot codec table. Existing read path (Go2 DDS captures) unchanged. Blobs/vectors/embeddings/on_existing=APPEND are sqlite-only.
  • RecorderConfig.db_path defaults to <recording_dir>/memory.<format>. Per-blueprint stamp helpers in go2 mid360/nav-3d/g1 removed; pcaps land in the same dir.
  • OdomRecorder (color_image/lidar/odom + tf, posed at latest odom) replaces Go2Memory; wired into unitree_go2_basic and unitree_g1_primitive_no_nav. DroneRecorder in drone_basic, PointlioRecorder (remapped) in go2_zenoh_basic. unitree-go2-memory is gone; it's unitree-go2 --record.
  • ReplayConnection opens by suffix via open_dataset, so --replay-db path/to/memory.mcap works.
  • mcap moves to core dependencies (was unitree-dds extra).

Per-robot recorders are interim: a Recorder records its declared In ports only. A record-everything module is specced separately (agents/engineering/record-flag/record-all-spec.md).

Not in this PR: dimos map pose-fill mcap support.

Test plan

  • dimos/memory/store/test_mcap.py: write -> read roundtrip (Image/PoseStamped/int, pose + tags), write-only guard
  • parser test for --record / --record-format
  • pytest dimos/memory dimos/robot dimos/core/coordination dimos/cli dimos/imitation/collection dimos/hardware/sensors/lidar: 954 passed
  • mypy on changed files
  • dimos --record --simulation run unitree-go2 / unitree-g1-sim, then dimos mem summary recordings/<run-id>/memory.mcap
  • dimos --replay --replay-db recordings/<run-id>/memory.mcap run unitree-go2

@spomichter
feat: --record flag, mcap recording by default, recorder in base blue…
87a1dfa 
…prints

- GlobalConfig.record / record_format (mcap|sqlite); Recorder.start() is a no-op
  unless --record (replay still disables it). Explicit *-record blueprints set
  record=True.
- McapStore gains write mode: one channel per stream, dimos-obs envelope
  (pose + tags header, codec payload), codec id in the schema so the file
  reads back without robot-specific codec tables.
- One RECORDING_DIR per process; RecorderConfig.db_path defaults to
  recordings/<stamp>/mem2.{mcap|db}. Per-blueprint stamp helpers removed.
- Go2Memory moves into unitree_go2_basic; DroneRecorder in drone_basic;
  PointlioRecorder in go2_zenoh_basic. unitree-go2-memory alias dropped.
- ReplayConnection opens by suffix (open_dataset), so mcap recordings replay.
- mcap moves to core dependencies.
@codecov

codecov Bot commented Aug 26, 2026 •
edited
Loading

Copy link
Copy Markdown

❌ 2 Tests Failed:

Tests completed Failed Passed Skipped
4679 2 4677 39
View the top 2 failed test(s) by shortest run time 
dimos.imitation.test_datacollection_e2e::test_collection_to_lerobot_roundtrip
Stack Traces | 0.001s run time 
tmp_path_factory = TempPathFactory(_given_basetemp=PosixPath('.../pytest-of-runner/pytest-0/popen-gw0'), _trace=<pluggy._tracing.TagTrac...6c7d820>, _basetemp=PosixPath('.../pytest-of-runner/pytest-0/popen-gw0'), _retention_count=3, _retention_policy='all')

    @pytest.fixture(scope="module")
    def recorded_session(
        tmp_path_factory: pytest.TempPathFactory,
    ) -> tuple[Path, dict[float, np.ndarray[Any, Any]]]:
        db_path = tmp_path_factory.mktemp("recorded-session") / "recording.db"
>       _record_session(db_path)

db_path    = PosixPath('.../pytest-of-runner/pytest-0/popen-gw0/recorded-session0/recording.db')
tmp_path_factory = TempPathFactory(_given_basetemp=PosixPath('.../pytest-of-runner/pytest-0/popen-gw0'), _trace=<pluggy._tracing.TagTrac...6c7d820>, _basetemp=PosixPath('.../pytest-of-runner/pytest-0/popen-gw0'), _retention_count=3, _retention_policy='all')

dimos/imitation/test_datacollection_e2e.py:226: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
dimos/imitation/test_datacollection_e2e.py:156: in _record_session
    publish(
        base       = 0.0
        counts     = {'color_image': 0, 'coordinator_joint_state': 0, 'status': 1}
        db_path    = PosixPath('.../pytest-of-runner/pytest-0/popen-gw0/recorded-session0/recording.db')
        discarded  = 0
        episodes   = [(100.0, 'pick', True, 0.0), (104.0, 'discard-me', False, 10.0), (108.0, 'place', True, 20.0)]
        name       = 'status'
        publish    = <function _record_session.<locals>.publish at 0x7fa06b369440>
        recorder   = <dimos.imitation.collection.recorder.CollectionRecorder object at 0x7f9e3cf3c6e0>
        saved      = 0
        start_ts   = 100.0
        success    = True
        task       = 'pick'
        transport  = <dimos.imitation.test_datacollection_e2e._DirectTransport object at 0x7f9e3cf3d250>
        transports = {'color_image': <dimos.imitation.test_datacollection_e2e._DirectTransport object at 0x7f9e3cf3d010>, 'coordinator_join...ject at 0x7f9e3cf3fa40>, 'status': <dimos.imitation.test_datacollection_e2e._DirectTransport object at 0x7f9e3cf3d250>}
dimos/imitation/test_datacollection_e2e.py:139: in publish
    wait_until(
        counts     = {'color_image': 0, 'coordinator_joint_state': 0, 'status': 1}
        message    = EpisodeStatus(ts=100.0, state='recording', episodes_saved=0, episodes_discarded=0, last_event='start', task_label='pick')
        name       = 'status'
        recorder   = <dimos.imitation.collection.recorder.CollectionRecorder object at 0x7f9e3cf3c6e0>
        transports = {'color_image': <dimos.imitation.test_datacollection_e2e._DirectTransport object at 0x7f9e3cf3d010>, 'coordinator_join...ject at 0x7f9e3cf3fa40>, 'status': <dimos.imitation.test_datacollection_e2e._DirectTransport object at 0x7f9e3cf3d250>}
.../utils/testing/waiting.py:32: in wait_until
    if predicate():
        deadline   = 751.380636952
        interval   = 0.005
        message    = 'status message 1 was not recorded'
        predicate  = <function _record_session.<locals>.publish.<locals>.<lambda> at 0x7fa06b369620>
        timeout    = 5.0
dimos/imitation/test_datacollection_e2e.py:140: in <lambda>
    lambda: recorder.store.stream(name).count() == counts[name],
        counts     = {'color_image': 0, 'coordinator_joint_state': 0, 'status': 1}
        name       = 'status'
        recorder   = <dimos.imitation.collection.recorder.CollectionRecorder object at 0x7f9e3cf3c6e0>
.../memory/store/base.py:197: in stream
    backend = self._create_backend(name, payload_type, **resolved)
        name       = 'status'
        overrides  = {}
        payload_type = None
        resolved   = {'eager_blobs': False, 'must_exist': False, 'page_size': 256, 'path': '.../pytest-of-runner/pytest-0/popen-gw0/recorded-session0/recording.db'}
        self       = <dimos.memory.store.sqlite.SqliteStore object at 0x7f9e3cf3c4a0>
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <dimos.memory.store.sqlite.SqliteStore object at 0x7f9e3cf3c4a0>
name = 'status', payload_type = None
config = {'eager_blobs': False, 'must_exist': False, 'page_size': 256, 'path': '.../pytest-of-runner/pytest-0/popen-gw0/recorded-session0/recording.db'}
stored = None

    def _create_backend(
        self, name: str, payload_type: type[Any] | None = None, **config: Any
    ) -> Backend[Any]:
        validate_identifier(name)
    
        stored = self._registry.get(name)
    
        if stored is not None:
            # Load path: validate type, assemble from stored config
            if payload_type is not None:
                actual_module = f"{payload_type.__module__}.{payload_type.__qualname__}"
                if actual_module != stored["payload_module"]:
                    raise ValueError(
                        f"Stream {name!r} was created with type {stored['payload_module']}, "
                        f"but opened with {actual_module}"
                    )
            return self._assemble_backend(name, stored)
    
        # Create path: inject conn-shared defaults, then delegate to base
        if payload_type is None:
>           raise TypeError(f"Stream {name!r} does not exist yet — payload_type is required")
E           TypeError: Stream 'status' does not exist yet — payload_type is required

__class__  = <class 'dimos.memory.store.sqlite.SqliteStore'>
config     = {'eager_blobs': False, 'must_exist': False, 'page_size': 256, 'path': '.../pytest-of-runner/pytest-0/popen-gw0/recorded-session0/recording.db'}
name       = 'status'
payload_type = None
self       = <dimos.memory.store.sqlite.SqliteStore object at 0x7f9e3cf3c4a0>
stored     = None

.../memory/store/sqlite.py:169: TypeError
dimos.imitation.test_datacollection_e2e::test_collection_to_hdf5_roundtrip
Stack Traces | 0.106s run time 
tmp_path_factory = TempPathFactory(_given_basetemp=PosixPath('.../pytest-of-runner/pytest-0/popen-gw0'), _trace=<pluggy._tracing.TagTrac...6c7d820>, _basetemp=PosixPath('.../pytest-of-runner/pytest-0/popen-gw0'), _retention_count=3, _retention_policy='all')

    @pytest.fixture(scope="module")
    def recorded_session(
        tmp_path_factory: pytest.TempPathFactory,
    ) -> tuple[Path, dict[float, np.ndarray[Any, Any]]]:
        db_path = tmp_path_factory.mktemp("recorded-session") / "recording.db"
>       _record_session(db_path)

db_path    = PosixPath('.../pytest-of-runner/pytest-0/popen-gw0/recorded-session0/recording.db')
tmp_path_factory = TempPathFactory(_given_basetemp=PosixPath('.../pytest-of-runner/pytest-0/popen-gw0'), _trace=<pluggy._tracing.TagTrac...6c7d820>, _basetemp=PosixPath('.../pytest-of-runner/pytest-0/popen-gw0'), _retention_count=3, _retention_policy='all')

dimos/imitation/test_datacollection_e2e.py:226: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
dimos/imitation/test_datacollection_e2e.py:156: in _record_session
    publish(
        base       = 0.0
        counts     = {'color_image': 0, 'coordinator_joint_state': 0, 'status': 1}
        db_path    = PosixPath('.../pytest-of-runner/pytest-0/popen-gw0/recorded-session0/recording.db')
        discarded  = 0
        episodes   = [(100.0, 'pick', True, 0.0), (104.0, 'discard-me', False, 10.0), (108.0, 'place', True, 20.0)]
        name       = 'status'
        publish    = <function _record_session.<locals>.publish at 0x7fa06b369440>
        recorder   = <dimos.imitation.collection.recorder.CollectionRecorder object at 0x7f9e3cf3c6e0>
        saved      = 0
        start_ts   = 100.0
        success    = True
        task       = 'pick'
        transport  = <dimos.imitation.test_datacollection_e2e._DirectTransport object at 0x7f9e3cf3d250>
        transports = {'color_image': <dimos.imitation.test_datacollection_e2e._DirectTransport object at 0x7f9e3cf3d010>, 'coordinator_join...ject at 0x7f9e3cf3fa40>, 'status': <dimos.imitation.test_datacollection_e2e._DirectTransport object at 0x7f9e3cf3d250>}
dimos/imitation/test_datacollection_e2e.py:139: in publish
    wait_until(
        counts     = {'color_image': 0, 'coordinator_joint_state': 0, 'status': 1}
        message    = EpisodeStatus(ts=100.0, state='recording', episodes_saved=0, episodes_discarded=0, last_event='start', task_label='pick')
        name       = 'status'
        recorder   = <dimos.imitation.collection.recorder.CollectionRecorder object at 0x7f9e3cf3c6e0>
        transports = {'color_image': <dimos.imitation.test_datacollection_e2e._DirectTransport object at 0x7f9e3cf3d010>, 'coordinator_join...ject at 0x7f9e3cf3fa40>, 'status': <dimos.imitation.test_datacollection_e2e._DirectTransport object at 0x7f9e3cf3d250>}
.../utils/testing/waiting.py:32: in wait_until
    if predicate():
        deadline   = 751.380636952
        interval   = 0.005
        message    = 'status message 1 was not recorded'
        predicate  = <function _record_session.<locals>.publish.<locals>.<lambda> at 0x7fa06b369620>
        timeout    = 5.0
dimos/imitation/test_datacollection_e2e.py:140: in <lambda>
    lambda: recorder.store.stream(name).count() == counts[name],
        counts     = {'color_image': 0, 'coordinator_joint_state': 0, 'status': 1}
        name       = 'status'
        recorder   = <dimos.imitation.collection.recorder.CollectionRecorder object at 0x7f9e3cf3c6e0>
.../memory/store/base.py:197: in stream
    backend = self._create_backend(name, payload_type, **resolved)
        name       = 'status'
        overrides  = {}
        payload_type = None
        resolved   = {'eager_blobs': False, 'must_exist': False, 'page_size': 256, 'path': '.../pytest-of-runner/pytest-0/popen-gw0/recorded-session0/recording.db'}
        self       = <dimos.memory.store.sqlite.SqliteStore object at 0x7f9e3cf3c4a0>
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <dimos.memory.store.sqlite.SqliteStore object at 0x7f9e3cf3c4a0>
name = 'status', payload_type = None
config = {'eager_blobs': False, 'must_exist': False, 'page_size': 256, 'path': '.../pytest-of-runner/pytest-0/popen-gw0/recorded-session0/recording.db'}
stored = None

    def _create_backend(
        self, name: str, payload_type: type[Any] | None = None, **config: Any
    ) -> Backend[Any]:
        validate_identifier(name)
    
        stored = self._registry.get(name)
    
        if stored is not None:
            # Load path: validate type, assemble from stored config
            if payload_type is not None:
                actual_module = f"{payload_type.__module__}.{payload_type.__qualname__}"
                if actual_module != stored["payload_module"]:
                    raise ValueError(
                        f"Stream {name!r} was created with type {stored['payload_module']}, "
                        f"but opened with {actual_module}"
                    )
            return self._assemble_backend(name, stored)
    
        # Create path: inject conn-shared defaults, then delegate to base
        if payload_type is None:
>           raise TypeError(f"Stream {name!r} does not exist yet — payload_type is required")
E           TypeError: Stream 'status' does not exist yet — payload_type is required

__class__  = <class 'dimos.memory.store.sqlite.SqliteStore'>
config     = {'eager_blobs': False, 'must_exist': False, 'page_size': 256, 'path': '.../pytest-of-runner/pytest-0/popen-gw0/recorded-session0/recording.db'}
name       = 'status'
payload_type = None
self       = <dimos.memory.store.sqlite.SqliteStore object at 0x7f9e3cf3c4a0>
stored     = None

.../memory/store/sqlite.py:169: TypeError

To view more test analytics, go to the Test Analytics Dashboard
📋 Got 3 mins? Take this short survey to help us improve Test Analytics.

@spomichter
recording file is memory.{mcap,db}
785cab0
@spomichter
recording dir = recordings/<run-id>; generic OdomRecorder in go2 + g1…
f30aca6 
… base blueprints

- GlobalConfig.recording_dir, set by `dimos run` to RECORDINGS_DIR/<run-id>
  (same id as the log dir); --recording-dir overrides.
- RecorderConfig.db_path defaults to None -> <recording_dir>/memory.<format>.
- Go2Memory -> OdomRecorder (dimos.memory.module), used by unitree_go2_basic
  and unitree_g1_primitive_no_nav (lidar remapped from pointcloud).
@spomichter spomichter mentioned this pull request Aug 27, 2026
Merged
@spomichter

spomichter commented Aug 27, 2026

Copy link
Copy Markdown
Contributor Author

Superseded by the transport-tap approach (record every published topic, no per-robot recorder modules); mcap writer moves to a follow-up. New PR: #3710

@spomichter spomichter closed this Aug 27, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

@leshy leshy Awaiting requested review from leshy leshy will be requested when the pull request is marked ready for review leshy is a code owner
@paul-nechifor paul-nechifor Awaiting requested review from paul-nechifor paul-nechifor will be requested when the pull request is marked ready for review paul-nechifor is a code owner
@mustafab0 mustafab0 Awaiting requested review from mustafab0 mustafab0 will be requested when the pull request is marked ready for review mustafab0 is a code owner
@Dreamsorcerer Dreamsorcerer Awaiting requested review from Dreamsorcerer Dreamsorcerer will be requested when the pull request is marked ready for review Dreamsorcerer is a code owner
@aclauer aclauer Awaiting requested review from aclauer aclauer will be requested when the pull request is marked ready for review aclauer is a code owner

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

1 participant

@spomichter