mem summary cleanup, go2 replay cleanup, progress bars

dimos/mapping/utils/cli/dataset_validation.md

@@ -1,16 +1,12 @@
 Dataset Validation
 
 ```sh
-dimos map summary recording_go2_mid360_2026-05-29_4-45pm-PST.db
-
-Stream("color_image"): 11141 items, 2026-05-29 23:32:57 — 2026-05-29 23:45:57 (780.1s)
-Stream("fastlio_lidar"): 7240 items, 2026-05-29 23:32:56 — 2026-05-29 23:45:57 (781.7s)
-Stream("fastlio_odometry"): 18737 items, 2026-05-29 23:32:56 — 2026-05-29 23:45:57 (781.8s)
-Stream("lidar"): 6025 items, 2026-05-29 23:32:55 — 2026-05-29 23:45:57 (782.3s)
-Stream("odom"): 14630 items, 2026-05-29 23:32:55 — 2026-05-29 23:45:57 (782.3s)
+dimos mem summary recording_go2_mid360_2026-05-29_4-45pm-PST.db
 ```
 
-Shows which streams are in the database. You can replay messages in rerun:
+Prints a table of the streams in the database: per-stream item count, average
+Hz, start time (UTC), duration, and stored size, plus a total row. You can
+replay messages in rerun:
 
 ```sh
 dimos map replay recording_go2_mid360_2026-05-29_4-45pm-PST.db --duration 60

dimos/mapping/utils/cli/map.py

@@ -526,50 +526,53 @@ def _position(obs: Observation[Any]) -> tuple[float, float, float] | None:
     graph: PoseGraph | None = None
     if pgo:
         print("running PGO twopass map...")
-        prog = progress(total, "pgo pass 1 (optimizing)")
-        graph = lidar.tap(prog).transform(PGO()).last().data
+        with progress(total, "pgo pass 1 (optimizing)") as bar:
+            graph = lidar.tap(bar).transform(PGO()).last().data
 
         pgo_path = [
             (kf.optimized.translation.x, kf.optimized.translation.y, kf.optimized.translation.z)
             for kf in graph.keyframes
         ]
 
-        pgo_map = _accumulate(
-            kept,
-            voxel=voxel,
-            block_count=block_count,
-            device=device,
-            graph=graph,
-            register=register,
-            carve_columns=carve,
-            progress_cb=progress(n_kept, "pgo pass 2 (rebuilding)"),
-        )
+        with progress(n_kept, "pgo pass 2 (rebuilding)") as bar:
+            pgo_map = _accumulate(
+                kept,
+                voxel=voxel,
+                block_count=block_count,
+                device=device,
+                graph=graph,
+                register=register,
+                carve_columns=carve,
+                progress_cb=bar,
+            )
 
     full_pgo_map = None
     if full_pgo:
         assert graph is not None
-        full_pgo_map = _accumulate(
-            lidar,
+        with progress(total, "full pgo (rebuilding)") as bar:
+            full_pgo_map = _accumulate(
+                lidar,
+                voxel=voxel,
+                block_count=block_count,
+                device=device,
+                graph=graph,
+                register=register,
+                carve_columns=carve,
+                progress_cb=bar,
+            )
+
+    # Raw map: same dedup'd frames, no PGO correction.
+    with progress(n_kept, "reconstructing global map") as bar:
+        global_map = _accumulate(
+            kept,
             voxel=voxel,
             block_count=block_count,
             device=device,
-            graph=graph,
             register=register,
             carve_columns=carve,
-            progress_cb=progress(total, "full pgo (rebuilding)"),
+            progress_cb=bar,
         )
 
-    # Raw map: same dedup'd frames, no PGO correction.
-    global_map = _accumulate(
-        kept,
-        voxel=voxel,
-        block_count=block_count,
-        device=device,
-        register=register,
-        carve_columns=carve,
-        progress_cb=progress(n_kept, "reconstructing global map"),
-    )
-
     marker_dets: list[Observation[Any]] = []
     if markers:
         # Image observations in dimos recordings are stamped with
@@ -596,17 +599,18 @@ def _position(obs: Observation[Any]) -> tuple[float, float, float] | None:
         # Keep the sharpest frame per --marker-quality-window window, then
         # drop frames where the robot was moving (linear + rotational) faster
         # than the limits. Defaults match replay_marker.py so positions agree.
-        pipeline: Stream[Image] = color_image.tap(
-            progress(n_images, "detecting markers")
-        ).transform(QualityWindow(lambda img: img.sharpness, window=marker_quality_window))
-        if marker_max_speed > 0:
-            pipeline = pipeline.transform(
-                SpeedLimit(
-                    max_mps=marker_max_speed,
-                    max_dps=marker_max_rot_rate if marker_max_rot_rate > 0 else None,
-                )
+        with progress(n_images, "detecting markers") as bar:
+            pipeline: Stream[Image] = color_image.tap(bar).transform(
+                QualityWindow(lambda img: img.sharpness, window=marker_quality_window)
             )
-        all_dets = pipeline.transform(xf).to_list()
+            if marker_max_speed > 0:
+                pipeline = pipeline.transform(
+                    SpeedLimit(
+                        max_mps=marker_max_speed,
+                        max_dps=marker_max_rot_rate if marker_max_rot_rate > 0 else None,
+                    )
+                )
+            all_dets = pipeline.transform(xf).to_list()
         if marker_smoothing > 0:
             # Keep only the latest emission per track_id — that's the most
             # averaged pose, drawn once per tracked marker session.

dimos/mapping/utils/cli/rename.py

@@ -31,7 +31,7 @@
 
 import typer
 
-from dimos.memory2.codecs.base import _resolve_payload_type
+from dimos.memory2.codecs.base import resolve_payload_type
 from dimos.memory2.store.sqlite import SqliteStore
 from dimos.memory2.stream import Stream
 from dimos.memory2.type.observation import Observation
@@ -77,7 +77,7 @@ def _stream_payload_types(db_path: Path) -> dict[str, type]:
         rows = conn.execute("SELECT name, config FROM _streams").fetchall()
     finally:
         conn.close()
-    return {name: _resolve_payload_type(json.loads(cfg)["payload_module"]) for name, cfg in rows}
+    return {name: resolve_payload_type(json.loads(cfg)["payload_module"]) for name, cfg in rows}
 
 
 def _parse_renames(pairs: list[str]) -> dict[str, str]:

dimos/mapping/utils/cli/replay.py

@@ -31,57 +31,25 @@
 
 from __future__ import annotations
 
-from collections.abc import Callable
 from pathlib import Path
 import subprocess
-import time
 from typing import TYPE_CHECKING, Any
 
 import rerun as rr
 import typer
 
+from dimos.memory2.utils.progress import progress
+
 # Heavy dimos imports (mapping/memory2 → torch, scipy, open3d) are deferred into
 # main() so that `dimos map --help` stays fast. See test_cli_startup.py and the
 # same pattern in dimos/mapping/utils/cli/map.py.
 if TYPE_CHECKING:
     from dimos.memory2.stream import Stream
-    from dimos.memory2.type.observation import Observation
     from dimos.msgs.sensor_msgs.PointCloud2 import PointCloud2
 
 TIMELINE = "ts"
 
 
-def _progress(total: int, label: str) -> Callable[[Observation[Any]], None]:
-    """Matches dimos/mapping/utils/cli/map.py:progress."""
-    seen = 0
-    wall_start: float | None = None
-    last_wall: float | None = None
-    first_ts: float | None = None
-
-    def tick(obs: Observation[Any]) -> None:
-        nonlocal seen, wall_start, last_wall, first_ts
-        now = time.monotonic()
-        if wall_start is None:
-            wall_start = now
-            first_ts = obs.ts
-        assert first_ts is not None
-        frame_ms = (now - last_wall) * 1000 if last_wall is not None else 0.0
-        last_wall = now
-        seen += 1
-        pct = 100 * seen // total if total else 100
-        wall = now - wall_start
-        data = obs.ts - first_ts
-        speed = data / wall if wall > 0 else 0.0
-        end = "\n" if seen >= total else ""
-        print(
-            f"\r{label} {pct:>3}% [{seen}/{total}] {data:.1f}s ({speed:.1f} x rt) {frame_ms:.0f}ms/frame",
-            end=end,
-            flush=True,
-        )
-
-    return tick
-
-
 def _log_clouds(
     label: str,
     stream: Stream[PointCloud2],
@@ -99,14 +67,14 @@ def _log_clouds(
     whole pipeline.
     """
     n = total if total is not None else stream.count()
-    cb = _progress(n, label)
-    for obs in stream:
-        cb(obs)
-        rr.set_time(TIMELINE, timestamp=obs.ts)
-        rr.log(
-            entity,
-            obs.data.to_rerun(voxel_size=voxel, mode=point_mode, bottom_cutoff=bottom_cutoff),
-        )
+    with progress(n, label) as bar:
+        for obs in stream:
+            bar(obs)
+            rr.set_time(TIMELINE, timestamp=obs.ts)
+            rr.log(
+                entity,
+                obs.data.to_rerun(voxel_size=voxel, mode=point_mode, bottom_cutoff=bottom_cutoff),
+            )
 
 
 def _log_path(
@@ -122,22 +90,22 @@ def _log_path(
     without a pose are skipped.
     """
     n = stream.count()
-    cb = _progress(n, label)
     points: list[tuple[float, float, float]] = []
     last_ts: float | None = None
     emit_count = 0
-    for obs in stream:
-        cb(obs)
-        if obs.pose_tuple is None:
-            continue
-        points.append(
-            (float(obs.pose_tuple[0]), float(obs.pose_tuple[1]), float(obs.pose_tuple[2]))
-        )
-        last_ts = obs.ts
-        emit_count += 1
-        if emit_every > 0 and emit_count % emit_every == 0 and len(points) >= 2:
-            rr.set_time(TIMELINE, timestamp=obs.ts)
-            rr.log(entity, rr.LineStrips3D([points], colors=[color]))
+    with progress(n, label) as bar:
+        for obs in stream:
+            bar(obs)
+            if obs.pose_tuple is None:
+                continue
+            points.append(
+                (float(obs.pose_tuple[0]), float(obs.pose_tuple[1]), float(obs.pose_tuple[2]))
+            )
+            last_ts = obs.ts
+            emit_count += 1
+            if emit_every > 0 and emit_count % emit_every == 0 and len(points) >= 2:
+                rr.set_time(TIMELINE, timestamp=obs.ts)
+                rr.log(entity, rr.LineStrips3D([points], colors=[color]))
     if (
         last_ts is not None
         and len(points) >= 2
@@ -206,8 +174,8 @@ def main(
     ),
 ) -> None:
     """Dump a recording to .rrd (lidar clouds + camera frames) and open it in rerun."""
-    from dimos.mapping.utils.cli.summary import _stream_payload_types
     from dimos.mapping.voxels import VoxelMapTransformer
+    from dimos.memory2.cli.summary import stream_payload_types
     from dimos.memory2.store.sqlite import SqliteStore
     from dimos.memory2.transform import throttle
     from dimos.msgs.geometry_msgs.PoseStamped import PoseStamped
@@ -224,7 +192,7 @@ def main(
 
     # Resolve which streams to voxelize: all PointCloud2 streams, or the
     # explicit --map-source subset. Validate up front so typos fail fast.
-    pc_streams = [n for n, t in _stream_payload_types(db_path).items() if t is PointCloud2]
+    pc_streams = [n for n, t in stream_payload_types(db_path).items() if t is PointCloud2]
     map_sources = list(map_source) or pc_streams
     if (map or map_final) and (bad := [s for s in map_sources if s not in pc_streams]):
         raise typer.BadParameter(f"--map-source: not PointCloud2 stream(s): {', '.join(bad)}")
@@ -309,20 +277,20 @@ def clipped(name: str, ptype: type[Any]) -> Stream[Any]:
         # fastlio pose axis + path from fastlio_odometry stream.
         if "fastlio_odometry" in store.streams:
             odometry = clipped("fastlio_odometry", Odometry)
-            cb = _progress(odometry.count(), "fastlio_odometry")
-            for obs in odometry:
-                cb(obs)
-                if obs.pose_tuple is None:
-                    continue
-                rr.set_time(TIMELINE, timestamp=obs.ts)
-                x, y, z, qx, qy, qz, qw = obs.pose_tuple
-                rr.log(
-                    "world/fastlio",
-                    rr.Transform3D(
-                        translation=[x, y, z],
-                        quaternion=rr.Quaternion(xyzw=[qx, qy, qz, qw]),
-                    ),
-                )
+            with progress(odometry.count(), "fastlio_odometry") as bar:
+                for obs in odometry:
+                    bar(obs)
+                    if obs.pose_tuple is None:
+                        continue
+                    rr.set_time(TIMELINE, timestamp=obs.ts)
+                    x, y, z, qx, qy, qz, qw = obs.pose_tuple
+                    rr.log(
+                        "world/fastlio",
+                        rr.Transform3D(
+                            translation=[x, y, z],
+                            quaternion=rr.Quaternion(xyzw=[qx, qy, qz, qw]),
+                        ),
+                    )
             _log_path(
                 "  fastlio_path",
                 clipped("fastlio_odometry", Odometry),
@@ -333,20 +301,20 @@ def clipped(name: str, ptype: type[Any]) -> Stream[Any]:
         # Go2 native odom pose axis + path.
         if "odom" in store.streams:
             odom = clipped("odom", PoseStamped)
-            cb = _progress(odom.count(), "        odom")
-            for odom_obs in odom:
-                cb(odom_obs)
-                if odom_obs.pose_tuple is None:
-                    continue
-                rr.set_time(TIMELINE, timestamp=odom_obs.ts)
-                x, y, z, qx, qy, qz, qw = odom_obs.pose_tuple
-                rr.log(
-                    "world/odom",
-                    rr.Transform3D(
-                        translation=[x, y, z],
-                        quaternion=rr.Quaternion(xyzw=[qx, qy, qz, qw]),
-                    ),
-                )
+            with progress(odom.count(), "        odom") as bar:
+                for odom_obs in odom:
+                    bar(odom_obs)
+                    if odom_obs.pose_tuple is None:
+                        continue
+                    rr.set_time(TIMELINE, timestamp=odom_obs.ts)
+                    x, y, z, qx, qy, qz, qw = odom_obs.pose_tuple
+                    rr.log(
+                        "world/odom",
+                        rr.Transform3D(
+                            translation=[x, y, z],
+                            quaternion=rr.Quaternion(xyzw=[qx, qy, qz, qw]),
+                        ),
+                    )
             _log_path(
                 "     odom_path",
                 clipped("odom", PoseStamped),
@@ -359,19 +327,19 @@ def clipped(name: str, ptype: type[Any]) -> Stream[Any]:
             color_image.transform(throttle(1.0 / camera_hz)) if camera_hz > 0 else color_image
         )
         n_img = cam_pipeline.count()
-        cb = _progress(n_img, "  color_image")
-        for img_obs in cam_pipeline:
-            cb(img_obs)
-            rr.set_time(TIMELINE, timestamp=img_obs.ts)
-            if img_obs.pose_tuple is not None:
-                x, y, z, qx, qy, qz, qw = img_obs.pose_tuple
-                rr.log(
-                    "world/camera",
-                    rr.Transform3D(
-                        translation=[x, y, z], quaternion=rr.Quaternion(xyzw=[qx, qy, qz, qw])
-                    ),
-                )
-            rr.log("world/camera/image", img_obs.data.to_rerun())
+        with progress(n_img, "  color_image") as bar:
+            for img_obs in cam_pipeline:
+                bar(img_obs)
+                rr.set_time(TIMELINE, timestamp=img_obs.ts)
+                if img_obs.pose_tuple is not None:
+                    x, y, z, qx, qy, qz, qw = img_obs.pose_tuple
+                    rr.log(
+                        "world/camera",
+                        rr.Transform3D(
+                            translation=[x, y, z], quaternion=rr.Quaternion(xyzw=[qx, qy, qz, qw])
+                        ),
+                    )
+                rr.log("world/camera/image", img_obs.data.to_rerun())
 
     print(f"wrote {out}")
     if no_gui: