diff --git a/dimos/memory/cli/app.py b/dimos/memory/cli/app.py index b360b2a75b3..c121b3a879a 100644 --- a/dimos/memory/cli/app.py +++ b/dimos/memory/cli/app.py @@ -16,9 +16,14 @@ from __future__ import annotations +import json +from pathlib import Path + import typer from dimos.memory.cli.summary import main as _summary_main +from dimos.memory.convert_recording import convert as convert_recording +from dimos.memory.recording_migration import inspect_recording, migrate mem_app = typer.Typer(help="memory store commands", no_args_is_help=True) mem_app.command("summary")(_summary_main) @@ -39,3 +44,37 @@ def rerun( from dimos.memory.cli.render import render_store render_store(open_dataset(path), out=out, seconds=seconds, no_gui=no_gui, root=root) + + +@mem_app.command("convert") +def convert( + source: Path = typer.Argument(..., help="Local recording file or explicit batch directory"), + destination: Path = typer.Argument(..., help="New output file or new batch directory"), + output_format: str = typer.Option( + "mcap", "--format", help="Directory output format: mcap or db" + ), + dry_run: bool = typer.Option( + False, "--dry-run", help="Read-only type/schema/count preflight; no outputs or downloads" + ), +) -> None: + """Migrate legacy recordings to generated CDR; originals are never overwritten.""" + try: + if source.is_dir(): + result = migrate(source, destination, output_format=output_format, dry_run=dry_run) + elif dry_run: + if ( + destination.exists() + or destination.is_symlink() + or Path(str(destination) + ".conversion.jsonl").exists() + or Path(str(destination) + ".conversion.jsonl").is_symlink() + ): + raise FileExistsError("Refusing an existing output or report") + result = inspect_recording(source, destination.suffix.lstrip(".")) + else: + result = convert_recording(source, destination) + except Exception as exc: + typer.echo(f"conversion failed: {exc}", err=True) + raise typer.Exit(2) from exc + typer.echo(json.dumps(result, indent=2)) + if result.get("blocked", 0): + raise typer.Exit(2) diff --git a/dimos/memory/convert_recording.py b/dimos/memory/convert_recording.py new file mode 100644 index 00000000000..23c20b808a7 --- /dev/null +++ b/dimos/memory/convert_recording.py @@ -0,0 +1,682 @@ +# Copyright 2026 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. + +"""Offline migration of explicitly typed legacy memory recordings to generated CDR. + +Never instantiate classes named by a recording. Legacy pickle and external blob +stores are deliberately unsupported. Output publication is exclusive and atomic. +""" + +from __future__ import annotations + +from collections import Counter +from collections.abc import Iterator +from contextlib import contextmanager +from dataclasses import dataclass, field +from functools import lru_cache +import importlib +import io +import json +import math +import os +from pathlib import Path +import re +import sqlite3 +import tempfile +from typing import Any + +from dimos_cdrgen.registry import ( + decode as cdr_decode, + encode as cdr_encode, + initialize, + schema as cdr_schema, +) +from dimos_generated.dimos_msgs.msg import LineSegment3D, LineSegments3D +from dimos_generated.geometry_msgs.msg import Point +import lz4.frame +from mcap.reader import make_reader +from mcap.records import Attachment, Channel, Metadata, Schema +from mcap.stream_reader import StreamReader +import numpy as np +from rosbags.interfaces import Nodetype +from rosbags.interfaces.typing import FieldDesc +import sqlite_vec + +from dimos.memory.store.sqlite import SqliteStore +from dimos.memory.utils.validation import validate_identifier +from dimos.msgs.helpers import resolve_msg_type +from dimos.msgs.time import message_header +from dimos.protocol.cdr_mcap import CdrMcapWriter + +# Deliberate, reviewable mappings. No suffix-based arbitrary Python imports. +TYPES = { + "std_msgs": ("String", "Header", "Bool", "Float32", "Float64", "Int32", "UInt32"), + "builtin_interfaces": ("Time", "Duration"), + "geometry_msgs": ( + "Point", + "PointStamped", + "Vector3", + "Quaternion", + "Pose", + "PoseStamped", + "PoseWithCovariance", + "PoseWithCovarianceStamped", + "Transform", + "TransformStamped", + "Twist", + "TwistStamped", + "TwistWithCovariance", + "TwistWithCovarianceStamped", + "Wrench", + "WrenchStamped", + ), + "sensor_msgs": ( + "Image", + "CompressedImage", + "PointCloud2", + "PointField", + "Imu", + "CameraInfo", + "JointState", + "Joy", + ), + "nav_msgs": ("Odometry", "Path", "OccupancyGrid", "MapMetaData"), + "tf2_msgs": ("TFMessage",), +} +ROS_TYPES = {f"{package}/msg/{name}" for package, names in TYPES.items() for name in names} +LEGACY_TYPES = { + path: f"{package}/msg/{name}" + for package, names in TYPES.items() + for name in names + for path in ( + f"dimos.msgs.{package}.{name}.{name}", + f"dimos.msgs.{package}.{name}", + f"dimos_lcm.{package}.{name}.{name}", + f"lcm_msgs.{package}.{name}.{name}", + ) +} +GENERATED_TYPES = { + f"dimos_generated.{package}.msg.{name}": f"{package}/msg/{name}" + for package, names in TYPES.items() + for name in names +} +REGION_STREAMS = { + "seed_map": ("sensor_msgs/msg/PointCloud2", "dimos_msgs/msg/RegionPointCloud2"), + "map_regions": ("sensor_msgs/msg/PointCloud2", "dimos_msgs/msg/RegionPointCloud2"), + "surface_map": ("sensor_msgs/msg/PointCloud2", "dimos_msgs/msg/RegionPointCloud2"), + "seed_bounds": ("geometry_msgs/msg/PoseStamped", "dimos_msgs/msg/RegionBounds"), + "region_bounds": ("geometry_msgs/msg/PoseStamped", "dimos_msgs/msg/RegionBounds"), + "node_edges": ("nav_msgs/msg/Path", "dimos_msgs/msg/RegionLineSegments3D"), +} +LEGACY_TYPES["dimos.msgs.nav_msgs.LineSegments3D.LineSegments3D"] = "nav_msgs/msg/Path" +LEGACY_TYPES["dimos.msgs.nav_msgs.LineSegments3D"] = "nav_msgs/msg/Path" +ROS_TYPES.update(target for _, target in REGION_STREAMS.values()) + +CODECS = {"lcm", "jpeg", "json", "cdr", "lz4+lcm", "lz4+jpeg", "lz4+json"} + + +@dataclass(frozen=True) +class Stream: + name: str + ros_type: str + codec: str + schema: str = "" + metadata: dict[str, Any] = field(default_factory=dict) + + @property + def region_type(self) -> str | None: + if self.codec not in {"lcm", "lz4+lcm"}: + return None + name = self.name + if name.startswith("dimos/"): + parts = name.split("/") + if len(parts) == 3 and parts[2].rsplit(".", 1)[-1] == self.ros_type.rsplit("/", 1)[-1]: + name = parts[1] + contract = REGION_STREAMS.get(name) + return contract[1] if contract and contract[0] == self.ros_type else None + + @property + def target(self) -> Any: + name = self.region_type or ( + "sensor_msgs/msg/CompressedImage" if self.codec.endswith("jpeg") else self.ros_type + ) + result = resolve_msg_type(name) + if result is None: + raise ValueError(f"Generated message package does not provide {name}") + return result + + +@dataclass +class Row: + stream: Stream + data: bytes + log_ns: int + publish_ns: int | None + sequence: int | None + observation_ts: float + metadata: dict[str, Any] = field(default_factory=dict) + embedding: bytes | None = None + + +def _stream(name: str, payload: str, codec: str, schema: str = "", **metadata: Any) -> Stream: + ros_type = payload if payload in ROS_TYPES else LEGACY_TYPES.get(payload) + if ros_type is None or codec not in CODECS: + raise ValueError( + f"{name}: unsupported type={payload!r}, codec={codec!r} (pickle is never loaded)" + ) + if codec.endswith("jpeg") and ros_type != "sensor_msgs/msg/Image": + raise ValueError(f"{name}: jpeg requires the legacy Image envelope") + if codec.endswith("json") and ros_type != "std_msgs/msg/String": + raise ValueError(f"{name}: json requires explicitly declared std_msgs/String") + result = Stream(name, ros_type, codec, schema, metadata) + if codec == "cdr" and schema != cdr_schema(result.target.__msgtype__): + raise ValueError(f"{name}: CDR schema differs from the installed generated definition") + return result + + +def _mcap_stream(channel: Channel, schemas: dict[int, Schema]) -> Stream: + schema = schemas.get(channel.schema_id) + payload = channel.metadata.get("dimos.payload_type", "") + if channel.message_encoding == "cdr" and schema is not None and schema.encoding == "ros2msg": + payload = schema.name + if not payload and channel.topic.startswith("dimos/"): + # Historical DimOS wire names encode the type as the last segment. + typename = channel.topic.rsplit("/", 1)[-1] + matches = [t for t in ROS_TYPES if t.rsplit("/", 1)[-1] == typename] + if len(matches) == 1: + payload = matches[0] + return _stream( + channel.topic, + payload, + channel.message_encoding, + schema.data.decode() if schema else "", + channel_metadata=channel.metadata, + ) + + +@contextmanager +def read_mcap(path: Path) -> Iterator[tuple[list[Stream], Iterator[Row]]]: + with path.open("rb") as handle: + schemas: dict[int, Schema] = {} + channels: dict[int, Channel] = {} + metadata_records: list[dict[str, Any]] = [] + for record in StreamReader(handle, validate_crcs=True).records: + if isinstance(record, Attachment): + raise ValueError( + "MCAP attachments need a separate export; refusing to discard them" + ) + if isinstance(record, Metadata): + metadata_records.append({"name": record.name, "metadata": record.metadata}) + if isinstance(record, Schema): + schemas[record.id] = record + elif isinstance(record, Channel): + channels[record.id] = record + streams: dict[int, Stream] = {} + errors = [] + names: dict[str, Stream] = {} + for cid, channel in channels.items(): + try: + stream = _mcap_stream(channel, schemas) + if metadata_records: + stream.metadata["source_mcap_metadata"] = metadata_records + if stream.name in names and names[stream.name] != stream: + raise ValueError(f"{stream.name}: conflicting MCAP channel definitions") + names[stream.name] = stream + streams[cid] = stream + except ValueError as exc: + errors.append(str(exc)) + if errors: + raise ValueError("Unsupported streams; nothing converted:\n" + "\n".join(errors)) + handle.seek(0) + + def rows() -> Iterator[Row]: + for _, channel, msg in make_reader(handle, validate_crcs=True).iter_messages( + log_time_order=False + ): + ts = ( + msg.publish_time + if channel.metadata.get("dimos.observation_time") == "publish_time" + else msg.log_time + ) + yield Row( + streams[channel.id], + msg.data, + msg.log_time, + msg.publish_time, + msg.sequence, + ts / 1e9, + ) + + yield list(names.values()), rows() + + +@contextmanager +def read_sqlite( + path: Path, *, allow_vectors: bool = False +) -> Iterator[tuple[list[Stream], Iterator[Row]]]: + conn = sqlite3.connect(path.resolve().as_uri() + "?mode=ro", uri=True) + try: + conn.execute("PRAGMA trusted_schema=OFF") + conn.execute("BEGIN") # Stable read-only snapshot; never reconstruct registry components. + streams = [] + stream_counts: dict[str, int] = {} + errors = [] + for name, text in conn.execute("SELECT name, config FROM _streams ORDER BY name"): + try: + validate_identifier(name) + config = json.loads(text) + blob = config.get("blob_store", {}) + if ( + blob.get("class", "").replace("dimos.memory2.", "dimos.memory.") + != "dimos.memory.blobstore.sqlite.SqliteBlobStore" + or blob.get("config", {}).get("path") is not None + ): + raise ValueError(f"{name}: external/custom blob store is unsupported") + payload_type = config["payload_module"] + schema = "" + if config["codec_id"] == "cdr": + payload_type = GENERATED_TYPES.get(payload_type, payload_type) + # Match only installed allowlisted identities, never import a name + # supplied by the recording (including native rosbags module paths). + for ros_type in sorted(ROS_TYPES): + message_type = resolve_msg_type(ros_type) + if message_type is not None and payload_type in { + ros_type, + f"{message_type.__module__}.{message_type.__qualname__}", + }: + payload_type = ros_type + schema = cdr_schema(ros_type) + break + stream = _stream(name, payload_type, config["codec_id"], schema=schema) + vector_schema = conn.execute( + "SELECT sql FROM sqlite_master WHERE name=?", (name + "_vec",) + ).fetchone() + if vector_schema: + if not allow_vectors: + raise ValueError( + f"{name}: vector embeddings require SQLite output (.db); MCAP is recording-only" + ) + vector = config.get("vector_store", {}) + if ( + vector.get("class", "").replace("dimos.memory2.", "dimos.memory.") + != "dimos.memory.vectorstore.sqlite.SqliteVectorStore" + or vector.get("config", {}).get("path") is not None + ): + raise ValueError(f"{name}: external/custom vector store is unsupported") + match = re.search( + r"USING vec0\(embedding float\[(\d+)\] distance_metric=cosine\)$", + vector_schema[0], + re.IGNORECASE, + ) + if match is None: + raise ValueError(f"{name}: unsupported vector schema") + conn.enable_load_extension(True) + try: + sqlite_vec.load(conn) + finally: + conn.enable_load_extension(False) + orphan = conn.execute( + f'SELECT count(*) FROM "{name}_vec" v LEFT JOIN "{name}" m ON m.id=v.rowid WHERE m.id IS NULL' + ).fetchone()[0] + if orphan: + raise ValueError(f"{name}: {orphan} orphan vectors") + stream.metadata["vector_dimension"] = int(match[1]) + stream.metadata["vector_metric"] = "cosine" + stream_counts[name] = conn.execute(f'SELECT count(*) FROM "{name}"').fetchone()[0] + # Missing blobs must not disappear through an inner join. + missing = ( + 0 + if stream_counts[name] == 0 + else conn.execute( + f'SELECT count(*) FROM "{name}" m LEFT JOIN "{name}_blob" b ON b.id=m.id WHERE b.data IS NULL' + ).fetchone()[0] + ) + if missing: + raise ValueError(f"{name}: {missing} observations have no payload blob") + streams.append(stream) + except (ValueError, KeyError, sqlite3.Error) as exc: + errors.append(str(exc)) + if errors: + raise ValueError("Unsupported streams; nothing converted:\n" + "\n".join(errors)) + + def rows() -> Iterator[Row]: + for stream in streams: + name = stream.name + if not stream_counts[name]: + continue + sql = f'SELECT m.id,m.ts,m.value,m.pose_x,m.pose_y,m.pose_z,m.pose_qx,m.pose_qy,m.pose_qz,m.pose_qw,json(m.tags),b.data FROM "{name}" m JOIN "{name}_blob" b ON b.id=m.id ORDER BY m.id' + for record in conn.execute(sql): + ident, ts, value, *rest = record + if not math.isfinite(ts) or ts < 0: + raise ValueError(f"{name}: invalid observation timestamp {ts}") + embedding = None + if "vector_dimension" in stream.metadata: + vector_row = conn.execute( + f'SELECT embedding FROM "{name}_vec" WHERE rowid=?', (ident,) + ).fetchone() + if vector_row is not None: + embedding = bytes(vector_row[0]) + if ( + len(embedding) != 4 * stream.metadata["vector_dimension"] + or not np.isfinite(np.frombuffer(embedding, dtype=" dict[str, FieldDesc]: + # The runtime registry is frozen; inspect schema metadata once per type, + # rather than rereading the installed package on every nested recording value. + return dict(initialize().fielddefs[ros_type][1]) + + +def _convert_fields(old: Any, sequences: dict[str, int], path: str = "") -> Any: + if isinstance(old, (list, tuple)): + return [_convert_fields(item, sequences, f"{path}[{i}]") for i, item in enumerate(old)] + if not hasattr(old, "__slots__"): + return old + package = type(old).__module__.split(".")[-2] + name = type(old).__name__ + if package == "std_msgs" and name in {"Time", "Duration"}: + package = "builtin_interfaces" + ros_type = f"{package}/msg/{name}" + if ros_type not in ROS_TYPES: + raise ValueError(f"Unsupported nested LCM type {ros_type}") + target = resolve_msg_type(ros_type) + if target is None: + raise ValueError(f"Missing generated {ros_type}") + values = {} + fields = _target_fields(ros_type) + for key in old.__slots__: + value = getattr(old, key) + if ros_type == "std_msgs/msg/Header" and key == "seq": + sequences[f"{path}.seq".lstrip(".")] = value + continue + if key.endswith("_length") and key[:-7] in old.__slots__: + if value != len(getattr(old, key[:-7])): + raise ValueError(f"Invalid LCM sequence length at {path}.{key}") + continue + new_key = "nanosec" if package == "builtin_interfaces" and key == "nsec" else key + if new_key not in target.__annotations__: + raise ValueError(f"No lossless field mapping for {ros_type}.{key}") + converted = _convert_fields(value, sequences, f"{path}.{key}".lstrip(".")) + definition = fields[new_key] + if definition[0] == Nodetype.ARRAY or definition[0] == Nodetype.SEQUENCE: + element = definition[1][0] + if element[0] == Nodetype.BASE and element[1][0] != "string": + scalar = element[1][0] + dtype = np.dtype({"byte": "uint8", "char": "uint8"}.get(scalar, scalar)) + converted = ( + np.frombuffer(converted, dtype=dtype).copy() + if isinstance(converted, bytes) + else np.asarray(converted, dtype=dtype) + ) + values[new_key] = converted + return target(**values) + + +def decode(row: Row) -> tuple[Any, dict[str, int]]: + codec, data = row.stream.codec, row.data + if codec.startswith("lz4+"): + data = lz4.frame.decompress(data) + codec = codec[4:] + if codec == "cdr": + return cdr_decode(data, row.stream.target), {} + if codec == "json": + text = data.decode("utf-8") + json.loads( + text, + parse_constant=lambda value: (_ for _ in ()).throw(ValueError(f"Invalid JSON {value}")), + ) + return row.stream.target(data=text), {} + package, _, name = row.stream.ros_type.split("/") + try: + module = importlib.import_module(f"dimos_lcm.{package}.{name}") + except ImportError as exc: + raise ValueError( + "LCM input needs the optional decoder: pip install dimos-lcm==0.1.4" + ) from exc + buffer = io.BytesIO(data) + old = getattr(module, name).lcm_decode(buffer) # Fingerprint verified by the official decoder. + if buffer.tell() != len(data): + raise ValueError(f"{row.stream.name}: trailing bytes after LCM message") + sequences: dict[str, int] = {} + if codec == "jpeg": + if old.encoding != "jpeg": + raise ValueError(f"{row.stream.name}: jpeg codec contains {old.encoding!r}") + h = _convert_fields(old.header, sequences, "header") + return row.stream.target( + header=h, format="jpeg", data=np.frombuffer(old.data, dtype=np.uint8).copy() + ), sequences + if name == "Image" and old.encoding == "jpeg": + raise ValueError(f"{row.stream.name}: JPEG Image requires explicit jpeg codec metadata") + if row.stream.region_type is not None: + return _convert_region(row, old, sequences), sequences + return _convert_fields(old, sequences), sequences + + +def _convert_region(row: Row, old: Any, sequences: dict[str, int]) -> Any: + """Migrate only the documented built-in region stream contracts.""" + header = _convert_fields(old.header, sequences, "header") + region_id = old.header.seq + if row.stream.region_type == "dimos_msgs/msg/RegionPointCloud2": + return row.stream.target(region_id=region_id, cloud=_convert_fields(old, sequences)) + if row.stream.region_type == "dimos_msgs/msg/RegionBounds": + return row.stream.target( + header=header, + region_id=region_id, + center=Point(x=old.pose.position.x, y=old.pose.position.y, z=old.pose.position.z), + radius=old.pose.orientation.x, + z_min=old.pose.orientation.y, + z_max=old.pose.orientation.z, + ) + if len(old.poses) % 2: + raise ValueError("Legacy region edge Path requires an even number of endpoint poses") + segments = [] + for i in range(0, len(old.poses), 2): + start, end = old.poses[i], old.poses[i + 1] + if ( + start.header.frame_id != old.header.frame_id + or end.header.frame_id != old.header.frame_id + ): + raise ValueError("Legacy region edge endpoints must share the Path frame") + if (start.header.stamp.sec, start.header.stamp.nsec) != ( + old.header.stamp.sec, + old.header.stamp.nsec, + ) or (end.header.stamp.sec, end.header.stamp.nsec) != ( + old.header.stamp.sec, + old.header.stamp.nsec, + ): + raise ValueError("Legacy region edge endpoints must share the Path timestamp") + if ( + any( + getattr(pose.pose.orientation, axis) != 0.0 + for pose in (start, end) + for axis in ("x", "y", "z") + ) + or start.pose.orientation.w != end.pose.orientation.w + ): + raise ValueError("Legacy region edge orientation must carry one matching pair weight") + sequences[f"poses[{i}].header.seq"] = start.header.seq + sequences[f"poses[{i + 1}].header.seq"] = end.header.seq + segments.append( + LineSegment3D( + start=Point( + x=start.pose.position.x, y=start.pose.position.y, z=start.pose.position.z + ), + end=Point(x=end.pose.position.x, y=end.pose.position.y, z=end.pose.position.z), + weight=start.pose.orientation.w, + ) + ) + return row.stream.target( + region_id=region_id, lines=LineSegments3D(header=header, segments=segments) + ) + + +def convert(source: Path, destination: Path) -> dict[str, Any]: + source, destination = source.resolve(), destination.absolute() + report = destination.with_name(destination.name + ".conversion.jsonl") + if destination.exists() or report.exists() or destination.is_symlink() or report.is_symlink(): + raise FileExistsError(f"Refusing to overwrite {destination} or its conversion report") + if destination.suffix not in {".mcap", ".db"}: + raise ValueError("Output must end in .mcap or .db") + if source.suffix not in {".mcap", ".db", ".sqlite"} or not source.is_file(): + raise ValueError( + "Input must be a local MCAP or memory SQLite file; pickle directories are unsupported" + ) + reader = ( + read_mcap(source) + if source.suffix == ".mcap" + else read_sqlite(source, allow_vectors=destination.suffix == ".db") + ) + with reader as (streams, rows): + if not streams: + raise ValueError("Input contains no declared streams") + for stream in streams: + if destination.suffix != ".mcap": + validate_identifier(stream.name) + # The property resolves/validates the generated type before output is created. + stream.target # noqa: B018 + with tempfile.TemporaryDirectory(prefix=".cdr-convert-", dir=destination.parent) as tmp: + temp = Path(tmp) / destination.name + sidecar = Path(tmp) / report.name + counts: Counter[str] = Counter({stream.name: 0 for stream in streams}) + writer: Any = ( + CdrMcapWriter(temp) + if destination.suffix == ".mcap" + else SqliteStore(path=str(temp)) + ) + with writer, sidecar.open("w") as audit: + audit.write( + json.dumps({"source": str(source), "streams": [s.__dict__ for s in streams]}) + + "\n" + ) + if destination.suffix == ".mcap": + for stream in streams: + writer.register_stream( + stream.name, + schema_name=stream.target.__msgtype__, + schema=cdr_schema(stream.target.__msgtype__), + metadata={ + **stream.metadata.get("channel_metadata", {}), + "dimos.payload_type": f"{stream.target.__module__}.{stream.target.__name__}", + }, + ) + else: + for stream in streams: + writer.stream(stream.name, stream.target, codec="cdr") + for row in rows: + try: + value, sequences = decode(row) + payload = cdr_encode(value) + except Exception as exc: + raise ValueError( + f"{row.stream.name}[{counts[row.stream.name]}]: {exc}" + ) from exc + header = message_header(value) + source_ns = ( + (header.stamp.sec * 1_000_000_000 + header.stamp.nanosec) + if header is not None + else None + ) + publish_ns = ( + row.publish_ns + if row.publish_ns is not None + else (source_ns if source_ns is not None else row.log_ns) + ) + sequence = ( + row.sequence + if row.sequence is not None + else sequences.get("header.seq", sequences.get("seq", 0)) & 0xFFFFFFFF + ) + if not ( + 0 <= row.log_ns < 2**64 + and 0 <= publish_ns < 2**64 + and 0 <= sequence < 2**32 + ): + raise ValueError( + f"{row.stream.name}: time/sequence cannot be represented in MCAP" + ) + meta = { + "stream": row.stream.name, + "index": counts[row.stream.name], + "log_time_ns": row.log_ns, + "publish_time_ns": publish_ns, + "source_header_time_ns": source_ns, + "sequence": sequence, + "legacy_header_sequences": sequences, + "original": row.metadata, + } + if destination.suffix == ".mcap": + writer.write( + row.stream.name, + payload, + schema_name=value.__msgtype__, + schema=cdr_schema(value.__msgtype__), + log_time_ns=row.log_ns, + publish_time_ns=publish_ns, + sequence=sequence, + ) + else: + written = writer.stream(row.stream.name).append( + value, + ts=row.observation_ts, + pose=tuple(row.metadata["pose"]) + if row.metadata.get("pose") + and all(v is not None for v in row.metadata["pose"]) + else None, + embedding=importlib.import_module( + "dimos.models.embedding.base" + ).Embedding(np.frombuffer(row.embedding, dtype=" dict[str, Any]: + """Read declarations and count rows; no dynamic registry imports or payload execution.""" + if output_format not in {"mcap", "db"}: + raise ValueError("Format must be mcap or db") + if source.is_symlink(): + raise ValueError("Symlink recordings are not followed; provide their explicit source root") + if source.suffix not in _INPUTS: + raise ValueError( + "Not an expanded MCAP/memory SQLite recording; archives, pickle and raw captures need explicit migration" + ) + with source.open("rb") as file: + if file.read(128).startswith(b"version https://git-lfs.github.com/spec/v1"): + raise ValueError( + "LFS pointer, not recording bytes; materialize it with the project data loader first" + ) + reader = ( + read_mcap(source) + if source.suffix == ".mcap" + else read_sqlite(source, allow_vectors=output_format == "db") + ) + with reader as (streams, rows): + if not streams: + raise ValueError("Recording has no declared streams") + counts: Counter[str] = Counter({stream.name: 0 for stream in streams}) + types = {} + for stream in streams: + if output_format == "db": + validate_identifier(stream.name) + types[stream.name] = { + "input_codec": stream.codec, + "output_type": stream.target.__msgtype__, + } + for row in rows: + counts[row.stream.name] += 1 + return {"counts": dict(counts), "total": sum(counts.values()), "types": types} + + +def _candidates(root: Path) -> Iterator[Path]: + # os.walk does not follow directory symlinks. Explicitly report them rather than + # traversing data outside the requested root or silently declaring it converted. + def fail_walk(error: OSError) -> None: + raise error + + for directory, subdirs, files in os.walk(root, followlinks=False, onerror=fail_walk): + for name in sorted(subdirs): + path = Path(directory) / name + if path.is_symlink(): + yield path + for name in sorted(files): + path = Path(directory) / name + uncompressed = name.removesuffix(".tar.gz") + if ( + path.suffix in _RECORDINGS + or Path(uncompressed).suffix in _RECORDINGS + or name.endswith((".tar.gz", ".zip", ".tar", ".tgz")) + ): + yield path + + +def plan(source: Path, destination: Path, output_format: str = "mcap") -> dict[str, Any]: + """Inventory only the explicitly supplied recording root, preserving relative paths.""" + if output_format not in {"mcap", "db"}: + raise ValueError("Format must be mcap or db") + source, destination = source.absolute(), destination.absolute() + if not source.is_dir() or source.is_symlink(): + raise ValueError("Batch source must be an explicit real directory") + source_resolved, output_resolved = source.resolve(), destination.resolve() + if ( + source_resolved in output_resolved.parents + or output_resolved in source_resolved.parents + or source_resolved == output_resolved + ): + raise ValueError("Input and output directory trees must not overlap") + if destination.exists() or destination.is_symlink(): + raise FileExistsError("Batch destination must be a new directory") + entries: list[dict[str, Any]] = [] + destinations: set[str] = set() + for path in sorted(_candidates(source)): + relative = path.relative_to(source) + # Keep the original suffix to avoid collisions between run.db and run.mcap. + output = destination / relative.parent / (relative.name + ".cdr." + output_format) + entry: dict[str, Any] = {"input": str(path), "output": str(output), "status": "ready"} + try: + if str(output) in destinations: + raise ValueError("Duplicate destination") + destinations.add(str(output)) + entry.update(inspect_recording(path, output_format)) + except Exception as exc: + entry.update(status="blocked", reason=str(exc)) + entries.append(entry) + if not entries: + raise ValueError("No recognized recording files found in the supplied directory") + return { + "source": str(source), + "destination": str(destination), + "format": output_format, + "preflight": "metadata, schema and row counts; payload integrity is checked during conversion", + "files": entries, + "blocked": sum(e["status"] == "blocked" for e in entries), + } + + +def migrate( + source: Path, destination: Path, *, output_format: str = "mcap", dry_run: bool = False +) -> dict[str, Any]: + """Preflight every candidate before publishing any batch member; report every result.""" + result = plan(source, destination, output_format) + if dry_run or result["blocked"]: + return result + destination.mkdir(parents=False, exist_ok=False) + for entry in result["files"]: + entry["status"] = "not_attempted" + try: + for entry in result["files"]: + output = Path(entry["output"]) + output.parent.mkdir(parents=True, exist_ok=True) + try: + converted = convert(Path(entry["input"]), output) + entry.update(status="converted", conversion=converted) + except Exception as exc: + entry.update(status="failed", reason=str(exc)) + result["blocked"] += 1 + break + finally: + # If a later file changes/fails after preflight, successes remain explicitly + # listed; exit is nonzero and no caller can mistake the partial batch for done. + with (destination / "migration-summary.json").open("x") as file: + json.dump(result, file, indent=2) + return result diff --git a/dimos/memory/test_convert_recording.py b/dimos/memory/test_convert_recording.py new file mode 100644 index 00000000000..c672cdad191 --- /dev/null +++ b/dimos/memory/test_convert_recording.py @@ -0,0 +1,573 @@ +# Copyright 2026 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. + +"""Offline migration preserves data and fails closed before publishing outputs.""" + +import copy +import hashlib +import json +from pathlib import Path +import sqlite3 +import sys + +from dimos_cdrgen.registry import decode as cdr_decode +from dimos_generated.std_msgs.msg import String +import lz4.frame +from mcap.reader import make_reader +from mcap.writer import Writer +import numpy as np +import pytest +from rosbags.typesys import Stores, get_types_from_msg, get_typestore +import sqlite_vec + +from dimos.memory import convert_recording +from dimos.memory.convert_recording import convert +from dimos.memory.recording_migration import inspect_recording +from dimos.memory.store.sqlite import SqliteStore +from dimos.models.embedding.base import Embedding + + +@pytest.fixture(autouse=True) +def restore_reference_types_module(): + previous = sys.modules.get("rosbags.usertypes") + try: + yield + finally: + if previous is None: + sys.modules.pop("rosbags.usertypes", None) + else: + sys.modules["rosbags.usertypes"] = previous + + +def write_mcap(path, streams): + with path.open("wb") as f: + writer = Writer(f) + writer.start() + for name, payload_type, codec, messages in streams: + channel = writer.register_channel( + name, + codec, + 0, + metadata={ + "dimos.payload_type": payload_type, + "dimos.observation_time": "publish_time", + }, + ) + for i, data in enumerate(messages): + writer.add_message(channel, 2_000_000_000 + i, data, 1_000_000_000 + i, 17 + i) + writer.finish() + + +def read_mcap(path): + with path.open("rb") as f: + return list(make_reader(f, validate_crcs=True).iter_messages()) + + +def legacy_pose(): + cls = pytest.importorskip("dimos_lcm.geometry_msgs.PoseStamped").PoseStamped + value = cls() + value.header.seq = 42 + value.header.stamp.sec = 1 + value.header.stamp.nsec = 123456789 + value.header.frame_id = "world" + value.pose.position.x = 3.25 + value.pose.orientation.w = 1.0 + return value + + +@pytest.mark.parametrize("suffix", [".mcap", ".db"]) +@pytest.mark.parametrize("codec", ["lcm", "lz4+lcm"]) +def test_pose_preserves_payload_header_and_both_envelope_times(tmp_path, suffix, codec): + old = legacy_pose() + data = old.lcm_encode() + if codec.startswith("lz4"): + data = lz4.frame.compress(data) + source, output = tmp_path / "old.mcap", tmp_path / ("new" + suffix) + write_mcap( + source, [("pose", "dimos.msgs.geometry_msgs.PoseStamped.PoseStamped", codec, [data])] + ) + digest = hashlib.sha256(source.read_bytes()).digest() + assert convert(source, output)["counts"] == {"pose": 1} + assert hashlib.sha256(source.read_bytes()).digest() == digest + if suffix == ".mcap": + [(schema, channel, row)] = read_mcap(output) + assert (row.log_time, row.publish_time, row.sequence) == (2_000_000_000, 1_000_000_000, 17) + assert channel.message_encoding == "cdr" and schema.encoding == "ros2msg" + assert ( + b"MSG: std_msgs/Header" in schema.data + and b"MSG: builtin_interfaces/Time" in schema.data + ) + store = get_typestore(Stores.EMPTY) + store.register(get_types_from_msg(schema.data.decode(), schema.name)) + value = store.deserialize_cdr(row.data, schema.name) + else: + with SqliteStore(path=str(output), must_exist=True) as store: + [obs] = list(store.stream("pose")) + assert obs.ts == 1.0 + assert obs.tags["cdr_conversion"]["log_time_ns"] == 2_000_000_000 + assert obs.tags["cdr_conversion"]["sequence"] == 17 + value = obs.data + with sqlite3.connect(output) as db: + assert ( + json.loads(db.execute("SELECT config FROM _streams").fetchone()[0])["codec_id"] + == "cdr" + ) + assert value.header.stamp.sec == 1 and value.header.stamp.nanosec == 123456789 + assert value.header.frame_id == "world" and value.pose.position.x == 3.25 + report = [ + json.loads(line) + for line in Path(str(output) + ".conversion.jsonl").read_text().splitlines() + ] + assert report[1]["legacy_header_sequences"] == {"header.seq": 42} + + +def test_jpeg_is_standard_compressed_image_without_recompression(tmp_path): + cls = pytest.importorskip("dimos_lcm.sensor_msgs.Image").Image + old = cls() + old.header.stamp.sec = 8 + old.header.stamp.nsec = 7 + old.header.frame_id = "camera" + old.encoding = "jpeg" + # Conversion must preserve encoded bytes, not involve a JPEG decode/encode. + old.data = b"\xff\xd8opaque-compressed-image\xff\xd9" + old.data_length = len(old.data) + source, output = tmp_path / "old.mcap", tmp_path / "new.mcap" + write_mcap( + source, [("image", "dimos.msgs.sensor_msgs.Image.Image", "jpeg", [old.lcm_encode()])] + ) + convert(source, output) + [(schema, _, row)] = read_mcap(output) + assert schema.name == "sensor_msgs/msg/CompressedImage" + store = get_typestore(Stores.EMPTY) + store.register(get_types_from_msg(schema.data.decode(), schema.name)) + msg = store.deserialize_cdr(row.data, schema.name) + assert msg.format == "jpeg" and bytes(msg.data) == old.data + assert msg.header.frame_id == "camera" and msg.header.stamp.nanosec == 7 + + +@pytest.mark.parametrize("codec", ["json", "lz4+json"]) +def test_json_string_preserves_utf8_and_stream_name(tmp_path, codec): + data = '{"note": "世界", "value": 1}'.encode() + source, output = tmp_path / "old.mcap", tmp_path / "new.mcap" + wire = lz4.frame.compress(data) if codec.startswith("lz4") else data + write_mcap(source, [("status/raw", "dimos.msgs.std_msgs.String.String", codec, [wire])]) + convert(source, output) + [(schema, channel, row)] = read_mcap(output) + assert channel.topic == "status/raw" and schema.name == "std_msgs/msg/String" + assert cdr_decode(row.data, String).data.encode() == data + + +def test_all_unsupported_streams_reported_before_output(tmp_path): + source, output = tmp_path / "old.mcap", tmp_path / "new.mcap" + write_mcap( + source, + [ + ("one", "os.system", "pickle", [b"not executed"]), + ("two", "unknown.Type", "lcm", []), + ("safe", "dimos.msgs.std_msgs.String.String", "json", [b"{}"]), + ], + ) + with pytest.raises(ValueError, match="one:.*\n.*two:"): + convert(source, output) + assert not output.exists() and not Path(str(output) + ".conversion.jsonl").exists() + + +@pytest.mark.parametrize("bad", [b"not-json", b'{"bad": NaN}']) +def test_bad_late_payload_publishes_nothing(tmp_path, bad): + source, output = tmp_path / "old.mcap", tmp_path / "new.mcap" + write_mcap(source, [("status", "dimos.msgs.std_msgs.String.String", "json", [b"{}", bad])]) + with pytest.raises(ValueError): + convert(source, output) + assert not output.exists() and not list(tmp_path.glob(".cdr-convert-*")) + + +def test_bad_lcm_fingerprint_rejected(tmp_path): + old = legacy_pose().lcm_encode() + source, output = tmp_path / "old.mcap", tmp_path / "new.mcap" + write_mcap( + source, + [ + ( + "pose", + "dimos.msgs.geometry_msgs.PoseStamped.PoseStamped", + "lcm", + [b"XXXXXXXX" + old[8:]], + ) + ], + ) + with pytest.raises(ValueError, match="Decode error"): + convert(source, output) + assert not output.exists() + + +def test_no_overwrite_or_inplace_conversion(tmp_path): + source, output = tmp_path / "old.mcap", tmp_path / "new.mcap" + source.write_bytes(b"original") + output.write_bytes(b"keep") + for destination in [source, output]: + with pytest.raises(FileExistsError): + convert(source, destination) + assert source.read_bytes() == b"original" and output.read_bytes() == b"keep" + + +def test_empty_stream_is_preserved(tmp_path): + source, output = tmp_path / "old.mcap", tmp_path / "new.mcap" + write_mcap(source, [("status", "dimos.msgs.std_msgs.String.String", "json", [])]) + assert convert(source, output)["counts"] == {"status": 0} + with output.open("rb") as f: + summary = make_reader(f).get_summary() + assert [c.topic for c in summary.channels.values()] == ["status"] + + +def legacy_sqlite(path, *, external=False): + old = legacy_pose() + with sqlite3.connect(path) as db: + db.executescript( + "CREATE TABLE _streams(name TEXT,config TEXT); CREATE TABLE pose(id INTEGER,ts REAL,value NUMERIC,pose_x REAL,pose_y REAL,pose_z REAL,pose_qx REAL,pose_qy REAL,pose_qz REAL,pose_qw REAL,tags TEXT); CREATE TABLE pose_blob(id INTEGER,data BLOB);" + ) + config = { + "payload_module": "dimos.msgs.geometry_msgs.PoseStamped.PoseStamped", + "codec_id": "lcm", + "blob_store": { + "class": "dimos.memory2.blobstore.sqlite.SqliteBlobStore", + "config": {"path": "/do-not-open" if external else None}, + }, + } + db.execute("INSERT INTO _streams VALUES (?,?)", ("pose", json.dumps(config))) + db.execute("INSERT INTO pose VALUES (9,2.5,7,1,2,3,0,0,0,1,?)", ('{"note":"preserve"}',)) + db.execute("INSERT INTO pose_blob VALUES (?,?)", (9, old.lcm_encode())) + + +@pytest.mark.parametrize("suffix", [".mcap", ".db"]) +def test_sqlite_registry_without_dynamic_component_imports(tmp_path, suffix): + source, output = tmp_path / "old.db", tmp_path / ("new" + suffix) + legacy_sqlite(source) + original = source.read_bytes() + convert(source, output) + assert source.read_bytes() == original + report = [ + json.loads(line) + for line in Path(str(output) + ".conversion.jsonl").read_text().splitlines() + ] + row = report[1] + assert row["log_time_ns"] == 2_500_000_000 and row["publish_time_ns"] == 1_123_456_789 + assert row["sequence"] == 42 + assert row["original"]["sqlite_id"] == 9 and row["original"]["tags"] == {"note": "preserve"} + if suffix == ".db": + with SqliteStore(path=str(output), must_exist=True) as store: + [obs] = list(store.stream("pose")) + assert obs.ts == 2.5 and obs.pose_tuple == (1, 2, 3, 0, 0, 0, 1) + assert obs.data.header.stamp.nanosec == 123456789 + + +def test_external_sqlite_components_and_missing_blobs_rejected(tmp_path): + source, output = tmp_path / "old.db", tmp_path / "new.mcap" + legacy_sqlite(source, external=True) + with pytest.raises(ValueError, match="external/custom blob"): + convert(source, output) + source.unlink() + legacy_sqlite(source) + with sqlite3.connect(source) as db: + db.execute("DELETE FROM pose_blob") + with pytest.raises(ValueError, match="no payload blob"): + convert(source, output) + assert not output.exists() + + +@pytest.mark.parametrize( + "typename", + ["sensor_msgs.Imu", "sensor_msgs.PointCloud2", "nav_msgs.Odometry", "tf2_msgs.TFMessage"], +) +def test_standard_nested_and_array_mappings(tmp_path, typename): + module = pytest.importorskip("dimos_lcm." + typename) + cls = getattr(module, typename.rsplit(".", 1)[1]) + old = cls() + source, output = tmp_path / "old.mcap", tmp_path / "new.mcap" + payload_type = "dimos.msgs." + typename + "." + cls.__name__ + write_mcap(source, [("data", payload_type, "lcm", [old.lcm_encode()])]) + convert(source, output) + [(schema, _, row)] = read_mcap(output) + store = get_typestore(Stores.EMPTY) + store.register(get_types_from_msg(schema.data.decode(), schema.name)) + result = store.deserialize_cdr(row.data, schema.name) + assert schema.name == typename.replace(".", "/msg/") + if cls.__name__ == "Imu": + assert len(result.orientation_covariance) == 9 + elif cls.__name__ == "PointCloud2": + assert len(result.fields) == 0 and len(result.data) == 0 + elif cls.__name__ == "Odometry": + assert len(result.pose.covariance) == 36 and len(result.twist.covariance) == 36 + else: + assert len(result.transforms) == 0 + + +def test_existing_report_and_racing_destination_are_not_overwritten(tmp_path, monkeypatch): + source, output = tmp_path / "old.mcap", tmp_path / "new.mcap" + write_mcap(source, [("status", "dimos.msgs.std_msgs.String.String", "json", [b"{}"])]) + report = Path(str(output) + ".conversion.jsonl") + report.write_text("keep") + with pytest.raises(FileExistsError): + convert(source, output) + assert report.read_text() == "keep" + report.unlink() + original_link = convert_recording.os.link + + def competing_link(src, dst): + if dst == output: + output.write_bytes(b"other writer") + return original_link(src, dst) + + monkeypatch.setattr(convert_recording.os, "link", competing_link) + with pytest.raises(FileExistsError): + convert(source, output) + assert output.read_bytes() == b"other writer" and not report.exists() + + +def test_mcap_container_metadata_is_kept_and_attachments_rejected(tmp_path): + source, output = tmp_path / "old.mcap", tmp_path / "new.mcap" + with source.open("wb") as f: + writer = Writer(f) + writer.start() + writer.register_channel( + "status", + "json", + 0, + metadata={"dimos.payload_type": "dimos.msgs.std_msgs.String.String"}, + ) + writer.add_metadata("source-info", {"robot": "offline-fixture"}) + writer.finish() + convert(source, output) + manifest = json.loads(Path(str(output) + ".conversion.jsonl").read_text().splitlines()[0]) + assert manifest["streams"][0]["metadata"]["source_mcap_metadata"] == [ + {"name": "source-info", "metadata": {"robot": "offline-fixture"}} + ] + attached = tmp_path / "attached.mcap" + with attached.open("wb") as f: + writer = Writer(f) + writer.start() + writer.add_attachment(1, 1, "robot.urdf", "text/xml", b"") + writer.finish() + with pytest.raises(ValueError, match="attachments"): + convert(attached, tmp_path / "no.mcap") + assert not (tmp_path / "no.mcap").exists() + + +def test_sqlite_vectors_preserved_by_new_ids_and_rejected_for_mcap(tmp_path): + source, output = tmp_path / "source.db", tmp_path / "converted.db" + with SqliteStore(path=str(source)) as store: + stream = store.stream("images", String, codec="cdr") + stream.append( + String(data="first"), ts=1, embedding=Embedding(np.array([1.0, 0.0], dtype=np.float32)) + ) + stream.append( + String(data="second"), ts=2, embedding=Embedding(np.array([0.0, 1.0], dtype=np.float32)) + ) + with sqlite3.connect(source) as conn: + conn.enable_load_extension(True) + sqlite_vec.load(conn) + conn.execute("UPDATE images SET id=42 WHERE id=2") + conn.execute("UPDATE images_blob SET id=42 WHERE id=2") + vector = conn.execute("SELECT embedding FROM images_vec WHERE rowid=2").fetchone()[0] + conn.execute("DELETE FROM images_vec WHERE rowid=2") + conn.execute("INSERT INTO images_vec(rowid,embedding) VALUES(42,?)", (vector,)) + assert inspect_recording(source, "db")["total"] == 2 + with pytest.raises(ValueError, match="SQLite output"): + convert(source, tmp_path / "rejected.mcap") + convert(source, output) + with SqliteStore(path=str(output)) as store: + hits = ( + store.stream("images") + .search(Embedding(np.array([0.0, 1.0], dtype=np.float32)), k=1) + .to_list() + ) + assert hits[0].data.data == "second" + assert hits[0].similarity == pytest.approx(1) + with sqlite3.connect(output) as conn: + conn.enable_load_extension(True) + sqlite_vec.load(conn) + assert ( + conn.execute("SELECT embedding FROM images_vec WHERE rowid=2").fetchone()[0] == vector + ) + audit = [ + json.loads(line) + for line in Path(str(output) + ".conversion.jsonl").read_text().splitlines() + ] + assert audit[2]["original"]["sqlite_id"] == 42 + assert audit[2]["output_sqlite_id"] == 2 + with sqlite3.connect(source) as conn: + conn.enable_load_extension(True) + sqlite_vec.load(conn) + conn.execute("INSERT INTO images_vec(rowid,embedding) VALUES(999,?)", (vector,)) + with pytest.raises(ValueError, match="orphan vectors"): + convert(source, tmp_path / "orphan.db") + assert not (tmp_path / "orphan.db").exists() + + +@pytest.mark.parametrize( + "typename", ["sensor_msgs.Image", "sensor_msgs.JointState", "sensor_msgs.Imu"] +) +def test_nonempty_numeric_and_byte_arrays_preserve_values(tmp_path, typename): + module = pytest.importorskip("dimos_lcm." + typename) + old = getattr(module, typename.rsplit(".", 1)[1])() + if typename.endswith("Image"): + old.encoding = "rgb8" + old.height, old.width, old.step = 1, 2, 6 + old.data = bytes([0, 1, 127, 128, 254, 255]) + old.data_length = len(old.data) + expected = {"data": np.frombuffer(old.data, dtype=np.uint8)} + elif typename.endswith("JointState"): + old.name = ["left", "right"] + old.name_length = 2 + expected = {} + for field, values in { + "position": [-1.5, 2.25], + "velocity": [0.125, -0.25], + "effort": [3.0, 4.0], + }.items(): + setattr(old, field, values) + setattr(old, field + "_length", len(values)) + expected[field] = np.array(values, dtype=np.float64) + else: + old.orientation_covariance = [i / 8 for i in range(9)] + expected = { + "orientation_covariance": np.array(old.orientation_covariance, dtype=np.float64) + } + source, output = tmp_path / "old.mcap", tmp_path / "new.mcap" + write_mcap(source, [("data", typename.replace(".", "/msg/"), "lcm", [old.lcm_encode()])]) + + convert(source, output) + + [(schema, _, row)] = read_mcap(output) + oracle = get_typestore(Stores.EMPTY) + oracle.register(get_types_from_msg(schema.data.decode(), schema.name)) + value = oracle.deserialize_cdr(row.data, schema.name) + for field, expected_array in expected.items(): + actual = getattr(value, field) + assert actual.dtype == expected_array.dtype + np.testing.assert_array_equal(actual, expected_array) + if typename.endswith("JointState"): + assert value.name == ["left", "right"] + + +@pytest.mark.parametrize("suffix", [".mcap", ".db"]) +@pytest.mark.parametrize( + "stream_name,package,name", + [ + ("map_regions", "sensor_msgs", "PointCloud2"), + ("seed_bounds", "geometry_msgs", "PoseStamped"), + ("node_edges", "nav_msgs", "Path"), + ], +) +def test_builtin_region_conversion_retains_signed_identity_and_geometry( + tmp_path, suffix, stream_name, package, name +): + old = getattr(pytest.importorskip(f"dimos_lcm.{package}.{name}"), name)() + old.header.seq = -196603 + old.header.stamp.sec = 1 + old.header.stamp.nsec = 34 + old.header.frame_id = "map" + if name == "PointCloud2": + old.height = 1 + old.point_step = 12 + elif name == "PoseStamped": + old.pose.position.x, old.pose.position.y = -3.0, 5.0 + old.pose.orientation.x, old.pose.orientation.y, old.pose.orientation.z = 2.5, -1.0, 4.0 + else: + endpoint = pytest.importorskip("dimos_lcm.geometry_msgs.PoseStamped").PoseStamped + old.poses = [] + for x, weight in [(1.0, 0.5), (4.0, 0.5), (9.0, 100.0), (6.0, 100.0)]: + pose = copy.deepcopy(endpoint()) + pose.header.frame_id = "map" + pose.header.stamp.sec, pose.header.stamp.nsec = 1, 34 + pose.pose.position.x = x + pose.pose.orientation.x, pose.pose.orientation.y, pose.pose.orientation.z = ( + 0.0, + 0.0, + 0.0, + ) + pose.pose.orientation.w = weight + old.poses.append(pose) + old.poses_length = len(old.poses) + source, output = tmp_path / "old.mcap", tmp_path / ("new" + suffix) + write_mcap( + source, [(stream_name, f"dimos.msgs.{package}.{name}.{name}", "lcm", [old.lcm_encode()])] + ) + assert convert(source, output)["counts"] == {stream_name: 1} + if suffix == ".mcap": + embedded, channel, record = read_mcap(output)[0] + value = cdr_decode(record.data, convert_recording.resolve_msg_type(embedded.name)) + assert b"MSG: std_msgs/Header" in embedded.data + assert record.sequence == 17 + else: + with SqliteStore(path=str(output), must_exist=True) as store: + value = store.stream(stream_name).first().data + assert value.region_id == -196603 + if name == "PointCloud2": + assert ( + value.cloud.width, + value.cloud.header.frame_id, + value.cloud.header.stamp.nanosec, + ) == (0, "map", 34) + elif name == "PoseStamped": + assert (value.center.x, value.center.y, value.radius, value.z_min, value.z_max) == ( + -3.0, + 5.0, + 2.5, + -1.0, + 4.0, + ) + else: + assert [segment.weight for segment in value.lines.segments] == [0.5, 100.0] + assert [(segment.start.x, segment.end.x) for segment in value.lines.segments] == [ + (1.0, 4.0), + (9.0, 6.0), + ] + + +def test_region_contract_is_not_inferred_for_arbitrary_cloud_streams(): + ordinary = convert_recording.Stream("custom/map_regions", "sensor_msgs/msg/PointCloud2", "lcm") + assert ordinary.region_type is None + standard = convert_recording.Stream("map_regions", "sensor_msgs/msg/PointCloud2", "cdr") + assert standard.region_type is None + + +@pytest.mark.parametrize("invalid", ["count", "frame", "timestamp", "orientation", "weight"]) +def test_region_edges_reject_unrepresentable_endpoint_fields(tmp_path, invalid): + path_type = pytest.importorskip("dimos_lcm.nav_msgs.Path").Path + pose_type = pytest.importorskip("dimos_lcm.geometry_msgs.PoseStamped").PoseStamped + old = copy.deepcopy(path_type()) + old.header.frame_id = "map" + old.header.stamp.sec, old.header.stamp.nsec = 1, 34 + old.poses = [copy.deepcopy(pose_type()), copy.deepcopy(pose_type())] + for pose in old.poses: + pose.header = copy.deepcopy(old.header) + pose.pose.orientation.x = pose.pose.orientation.y = pose.pose.orientation.z = 0.0 + pose.pose.orientation.w = 0.5 + if invalid == "count": + old.poses.pop() + elif invalid == "frame": + old.poses[0].header.frame_id = "odom" + elif invalid == "timestamp": + old.poses[0].header.stamp.nsec += 1 + elif invalid == "orientation": + old.poses[0].pose.orientation.x = 1.0 + else: + old.poses[0].pose.orientation.w = 2.0 + old.poses_length = len(old.poses) + source, output = tmp_path / "old.mcap", tmp_path / "new.mcap" + write_mcap(source, [("node_edges", "dimos.msgs.nav_msgs.Path.Path", "lcm", [old.lcm_encode()])]) + with pytest.raises(ValueError, match="Legacy region edge"): + convert(source, output) + assert not output.exists() + assert not output.with_name(output.name + ".conversion.jsonl").exists() diff --git a/dimos/memory/test_recording_migration.py b/dimos/memory/test_recording_migration.py new file mode 100644 index 00000000000..cc64e02ebf8 --- /dev/null +++ b/dimos/memory/test_recording_migration.py @@ -0,0 +1,119 @@ +# Copyright 2026 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. + +"""Public migration CLI never overwrites or silently skips unsupported recordings.""" + +import json + +from dimos_cdrgen.registry import decode as cdr_decode, encode as cdr_encode +from dimos_generated.builtin_interfaces.msg import Time +from dimos_generated.geometry_msgs.msg import Point, Pose, PoseStamped, Quaternion +from dimos_generated.std_msgs.msg import Header +import pytest +from typer.testing import CliRunner + +from dimos.memory.cli.app import mem_app +from dimos.memory.test_convert_recording import read_mcap, write_mcap +from dimos.msgs.geometry_msgs.PoseStamped import PoseStamped as LegacyPoseStamped + + +def source_file(path, payload=b'"hello"'): + path.parent.mkdir(parents=True, exist_ok=True) + write_mcap(path, [("text", "dimos.msgs.std_msgs.String.String", "json", [payload])]) + + +def test_cli_single_and_batch_dry_run_then_conversion(tmp_path): + source = tmp_path / "input" + source_file(source / "nested" / "run.mcap") + destination = tmp_path / "output" + runner = CliRunner() + dry = runner.invoke(mem_app, ["convert", str(source), str(destination), "--dry-run"]) + assert dry.exit_code == 0, dry.output + assert json.loads(dry.stdout)["files"][0]["total"] == 1 + assert not destination.exists() + done = runner.invoke(mem_app, ["convert", str(source), str(destination)]) + assert done.exit_code == 0, done.output + output = destination / "nested" / "run.mcap.cdr.mcap" + assert len(read_mcap(output)) == 1 + summary = json.loads((destination / "migration-summary.json").read_text()) + assert summary["files"][0]["status"] == "converted" + single = runner.invoke(mem_app, ["convert", str(output), str(tmp_path / "single.db")]) + assert single.exit_code == 0, single.output + assert (tmp_path / "single.db").is_file() + repeat = runner.invoke(mem_app, ["convert", str(source), str(destination)]) + assert repeat.exit_code == 2 + assert len(read_mcap(output)) == 1 + + +@pytest.mark.parametrize( + "blocked", ["unknown.pickle", "archive.tar.gz", "capture.pcap", "pointer.mcap"] +) +def test_batch_reports_blockers_and_publishes_nothing(tmp_path, blocked): + source = tmp_path / "input" + source_file(source / "good.mcap") + (source / blocked).write_bytes(b"version https://git-lfs.github.com/spec/v1\n") + destination = tmp_path / "output" + result = CliRunner().invoke(mem_app, ["convert", str(source), str(destination)]) + assert result.exit_code == 2, result.output + assert json.loads(result.stdout)["blocked"] == 1 + assert not destination.exists() + + +def test_batch_payload_failure_reports_partial_results(tmp_path): + source = tmp_path / "input" + source_file(source / "a.mcap") + source_file(source / "b.mcap", b"not json") + source_file(source / "c.mcap") + destination = tmp_path / "output" + result = CliRunner().invoke(mem_app, ["convert", str(source), str(destination)]) + assert result.exit_code == 2, result.output + summary = json.loads((destination / "migration-summary.json").read_text()) + assert [entry["status"] for entry in summary["files"]] == [ + "converted", + "failed", + "not_attempted", + ] + assert not (destination / "b.mcap.cdr.mcap").exists() + assert not (destination / "c.mcap.cdr.mcap").exists() + + +def test_overlap_symlinks_and_invalid_single_output_are_rejected(tmp_path): + source = tmp_path / "input" + source_file(source / "good.mcap") + runner = CliRunner() + overlap = runner.invoke(mem_app, ["convert", str(source), str(source / "out")]) + assert overlap.exit_code == 2 + (source / "link").symlink_to(tmp_path, target_is_directory=True) + linked = runner.invoke(mem_app, ["convert", str(source), str(tmp_path / "out")]) + assert linked.exit_code == 2 + assert json.loads(linked.stdout)["blocked"] == 1 + invalid = runner.invoke( + mem_app, ["convert", str(source / "good.mcap"), str(tmp_path / "out.txt"), "--dry-run"] + ) + assert invalid.exit_code == 2 + + +def test_retained_legacy_interface_is_explicit_and_cdr_remains_distinct(): + old = LegacyPoseStamped(ts=1.5, frame_id="world", position=(3.25, 0, 0)) + restored = LegacyPoseStamped.lcm_decode(old.lcm_encode()) + assert (restored.ts, restored.frame_id, restored.position.x) == (1.5, "world", 3.25) + current = PoseStamped( + header=Header(stamp=Time(sec=0, nanosec=0), frame_id=""), + pose=Pose( + position=Point(x=0.0, y=0.0, z=0.0), orientation=Quaternion(x=0.0, y=0.0, z=0.0, w=1.0) + ), + ) + assert cdr_decode(cdr_encode(current), PoseStamped) == current + with pytest.raises(ValueError): + cdr_decode(old.lcm_encode(), PoseStamped) diff --git a/docs/development/convert-recordings.md b/docs/development/convert-recordings.md new file mode 100644 index 00000000000..2b68d99f3db --- /dev/null +++ b/docs/development/convert-recordings.md @@ -0,0 +1,379 @@ +# Convert an old recording to CDR + +This offline tool is for reviewing historical dimOS recordings on the CDR proposal. +It does not start a blueprint, router, robot, viewer or network connection. + +From this proposal checkout, use an environment containing dimOS, its matching +`dimos-generated` package and the retained historical decoder. The commands below +require your existing historical recording: replace `old.mcap` or `old.db` with +its path. These input files are not created by the walkthrough. + +```sh skip +pip install dimos-lcm==0.1.4 +dimos mem convert old.mcap converted.mcap +``` + +Open `converted.mcap` directly in Foxglove with **Open local file**. Select your +image topic in an Image panel and point cloud / pose topics in a 3D panel. A local +Foxglove application or an existing signed-in web session may be needed; the +converter does not create an account or upload your recording. + +For an old memory SQLite recording, or for CDR SQLite output: + +```sh skip +dimos mem convert old.db converted.mcap +dimos mem convert old.mcap converted.db +dimos mem convert old.db converted.db +``` + +Choose a **new output filename**. The original file is opened read-only. Existing +outputs and reports are refused, including a destination created concurrently. +Conversion writes to a temporary sibling directory; no output is published if +any stream or payload fails. Output directories must already exist. + +Inspect either output through the current CDR memory reader: + +```sh skip +dimos mem summary converted.mcap +dimos mem summary converted.db +``` + +The converter loads the allowlisted historical decoders from `dimos-lcm`. +Historical `dimos.msgs` names are recognized as input metadata; their old runtime +classes and `lcm_encode`/`lcm_decode` APIs are not restored. This reuses the original +decoder implementation; it introduces no new codec. +New runtime transports and recordings still use generated CDR types; there is +no automatic old-wire fallback. +For JSON-only input it is unnecessary. `mcap`/`lz4` and SQLite support come from +the prepared dimOS environment. No dependencies are downloaded by conversion. + +## Accepted input and mapping + +| Input | Output | +| --- | --- | +| dimOS MCAP `lcm`, `jpeg`, `json`, optionally `lz4+…` | Generated standard ROS-compatible CDR values | +| Memory SQLite `_streams` registry, in-file observation/blob tables, same codecs | Same CDR values through the actual `SqliteStore` | +| Known CDR MCAP with `ros2msg` exactly matching the installed schema | Match the schema and decode/re-encode without changing the type | +| LCM Image with explicit `jpeg` codec | `sensor_msgs/msg/CompressedImage`, `format="jpeg"`, **original JPEG bytes** | +| Raw LCM Image / PointCloud2 / Imu / PoseStamped / Odometry / TFMessage and listed standard messages | Same ROS message shape; `std_msgs.Time.nsec` becomes `builtin_interfaces/Time.nanosec` | +| Explicit JSON `std_msgs.String` | Same UTF-8 JSON document inside generated `std_msgs/msg/String` | + +The full, explicit allowlist is `TYPES` in +`dimos/memory/convert_recording.py`. Only standard messages with compatible field +semantics are admitted; unknown fields/types are errors. LCM length fields are +checked against their arrays, then represented by CDR sequence lengths. The +installed historical decoder checks the LCM fingerprint. + +CDR input must come from trusted producers and match the installed schema. The +native Python decoder is not a hostile-input validation boundary: see the +[accepted native-library limitations](https://tomcc7.github.io/dimos-cdrgen/). +A matching schema does not make malformed bytes safe. Conversion uses the shared +registry codecs and native value types, with no fallback to the old message API. + +MCAP reads `dimos.payload_type` metadata. Historical +`dimos//` wire names are also recognized when the type is uniquely +listed. It never imports a Python class or SQLite component named by an input +file. **Pickle is never loaded.** Unknown streams are listed together before an +output is created; detected payload errors also leave no published output. + +Go2 DDS recordings use different conventions (`rt/utlidar/cloud`, +`rt/frontvideo`, proprietary Unitree types). A topic name alone does not specify +its bytes. Standard CDR channels need their matching full `ros2msg` definition; +undeclared raw video, proprietary DDS types and old Go2 pickle directories are +rejected rather than guessed. They need a separately reviewed, explicit export +from their trusted original producer. + +## What is preserved + +- Stream/topic names and per-stream message count/order, including empty streams. + SQLite requires SQL identifier stream names; use MCAP for names containing `/`. +- All mapped message fields, including Header frame and exact sec/nanosec values. +- MCAP input's exact integer `log_time`, `publish_time` and envelope `sequence`. + They are **not** replaced with conversion wall time or guessed Header time. +- SQLite's existing floating-point observation time. For MCAP output it supplies + log time (rounded to nanoseconds); Header supplies publish time when present. + Precision absent from the source float cannot be recovered. +- SQLite observation ID, value, pose and tags, and legacy ROS1 `Header.seq`, in a + mandatory adjacent `converted.mcap.conversion.jsonl` / `.db.conversion.jsonl` + audit file. MCAP container metadata is retained in the audit manifest. SQLite output additionally carries these in `cdr_conversion` tags + and retains the spatial pose. ROS2 Header has no `seq`; for SQLite input its + legacy value also supplies the MCAP sequence. For MCAP input the envelope + sequence wins and the original Header sequence remains in the audit file. + +MCAP uses `message_encoding="cdr"`, XCDR1 little endian, +`schema_encoding="ros2msg"`, full nested definitions, the `ros2` profile and ZSTD +chunks. It is not LCM bytes relabeled as CDR. SQLite output uses the current +`CdrCodec` and real store registry/blob format. Runtime import paths in output +refer to generated classes. + +This migrates message recordings, not arbitrary Python object databases. External +blob stores, MCAP attachments and pickle payloads are rejected instead of +silently discarded. Keep the audit file with the converted recording. + +## Validation + +With the historical decoder installed: + +```sh skip +python -m pytest dimos/memory/test_convert_recording.py dimos/protocol/test_cdr_mcap.py +``` + +Tests cover both containers, LZ4/JSON/JPEG/LCM, independent rosbags decoding, +Header/type mapping, timestamp/sequence retention, unsupported types, corrupt +payloads, empty streams and exclusive output publication. Legacy-specific tests +are skipped if the historical decoder is missing; that is not a full +converter acceptance run. + +## Convert a directory explicitly + +```sh skip +dimos mem convert ./recordings ./recordings-cdr --dry-run +dimos mem convert ./recordings ./recordings-cdr +dimos mem convert ./recordings ./recordings-cdr-sqlite --format db +``` + +The destination must be new, with an existing parent, and must not overlap the +source tree. Relative directories are preserved: `trip/run.db` becomes +`trip/run.db.cdr.mcap`. No home-directory scan or automatic download occurs. +Dry-run reads schema declarations and counts rows; it does not prove payload +integrity. Both dry-run and conversion exit nonzero when candidates are blocked. +Archives, LFS pointers, pickle, raw captures and symlinks are reported as blocked; +unrelated non-recording files are ignored. Extract/materialize trusted archives +separately before selecting an expanded source directory. + +All candidates pass metadata preflight before any conversion starts. Each output +and its audit file are published exclusively. If a later payload fails, earlier +successful files remain, later files are marked not attempted, and the command +fails. `migration-summary.json` records every candidate and result. Retain the +originals; a successful conversion is not authorization to delete them. + +## Recording versus writable analysis + +MCAP is a cheap bus-recording format, not an embedding database. Convert a +recording into the standard writable SQLite backend before analysis: + +```sh skip +dimos mem convert converted.mcap analysis.db +dimos mem summary analysis.db +``` + +Both filenames must be new when used as outputs. This uses the existing +`SqliteStore`, not a new database format. MCAP integer timestamps and sequence +are retained in `cdr_conversion` observation tags and the mandatory audit file; +SQLite observation timestamps remain floating point. + +For historical SQLite recordings with embeddings, use SQLite output. This path +uses the existing `Embedding` class and requires its PyTorch dependency (already +present in the development environment); it does not load a model or weights: + +```sh skip +uv pip install torch # only if absent from a core-only environment +dimos mem convert go2_short.db go2_short.cdr.db --dry-run && +dimos mem convert go2_short.db go2_short.cdr.db +``` + +In-file sqlite-vec float32/cosine vectors are copied through the existing vector +store, linked to newly assigned observation IDs. The report records the old/new +IDs. Payloads, poses and original tags remain associated with the observation. +External/custom vector stores, unsupported vector schemas and orphaned vectors +are rejected. DB-to-MCAP with vectors remains an error; there is no discard flag, +embedding channel, attachment or automatic sidecar-vector export. + +Existing Python APIs remain `Stream.append(data, embedding=embedding)`, +`stream.transform(EmbedImages(model)).save(destination).drain()` and +`stream.search(query_embedding, k=10)`. The first two append new observations; +they do not attach embeddings to an existing row. Search operates through the +SQLite vector store. There is no embedding-analysis CLI and no MCAP vector search. +Use the same model/preprocessing for queries as for stored vectors: historical +vectors do not establish model identity merely from their dimension. Compressed +images must be decoded to the input type expected by the chosen embedding model. +No model weights are downloaded or embeddings recomputed by conversion. Realtime +capture plus embedding analysis should write to the normal SQLite backend. + +Validation on Linux: the complete `go2_short.db` migration retained 2,546 +observations and all 108 stored 512-dimensional vectors, with exact vector bytes +and verified old/new ID associations. Real stairs MCAP-to-SQLite conversion +retained all 4,451 payloads and exact envelope metadata. A focused regression +checks noncontiguous source IDs and ranked search after migration. This does not +claim newly validated model inference, macOS execution or MCAP embedding support. + +## Ivan's offline acceptance checklist + +Run from the recording-converter branch, `cc/feat/cdr-recording-converter`. +For a fresh checkout, prepare the pinned message inputs before installing the +matching runtime and generated message package: + +```sh skip +export UV_PROJECT_ENVIRONMENT="$PWD/.venv-stairs" +export UV_CACHE_DIR="$PWD/.cache-stairs/uv" +uv run --no-project --python 3.12 --with 'dimos-cdrgen[prepare]==0.1.0' python -m scripts.prepare_builtin_sources +uv sync --frozen --python 3.12 --no-default-groups +uv export --frozen --no-default-groups --extra unitree --no-emit-project --no-emit-local --no-hashes --format requirements-txt --output-file "$PWD/.cache-stairs/demo-constraints.txt" +uv pip install --python "$UV_PROJECT_ENVIRONMENT/bin/python" --constraint "$PWD/.cache-stairs/demo-constraints.txt" 'mcap==1.3.1' 'unitree-webrtc-connect==2.1.2' +source "$UV_PROJECT_ENVIRONMENT/bin/activate" +``` + +The export provides locked versions for the two additional packages and their +dependencies. The current mapping CLI imports Unitree helpers during offline +runs. Preparation and installation require network access unless their inputs +are cached. The following conversion and CPU mapping commands use only local files +and do not start a robot. Use a fresh directory for each run: + +```sh skip +export CDR_REVIEW_DIR="$(mktemp -d "${TMPDIR:-/tmp}/dimos-cdr-review.XXXXXX")" +``` + +### Full real legacy point-cloud recording + +Use `go2_mid360_stairs.db.tar.gz`, an existing approximately 60-second legacy +recording: 54,082,131 compressed bytes and 155,848,704 extracted bytes. It contains +LCM point clouds/poses and JPEG images. This is a direct SQLite-to-CDR conversion; +no Python packaging script or synthetic fixture is needed. Download only this +archive, verify its project LFS hash, and extract into the fresh review directory: + +```sh skip +git lfs pull --include="data/.lfs/go2_mid360_stairs.db.tar.gz" --exclude="" +printf '%s\n' '02d8d2194332291cf71988458965aa9f8019b8d61661ff153e93886dd72e85e9 data/.lfs/go2_mid360_stairs.db.tar.gz' | shasum -a 256 -c - +tar -xzf data/.lfs/go2_mid360_stairs.db.tar.gz -C "$CDR_REVIEW_DIR" +dimos mem convert "$CDR_REVIEW_DIR/go2_mid360_stairs.db" "$CDR_REVIEW_DIR/converted.mcap" --dry-run && +dimos mem convert "$CDR_REVIEW_DIR/go2_mid360_stairs.db" "$CDR_REVIEW_DIR/converted.mcap" +dimos mem summary "$CDR_REVIEW_DIR/converted.mcap" +``` + +Expect 4,451 messages: `color_image` 855, `fastlio_lidar` 576, +`fastlio_odometry` 1,595, `lidar` 301 and `odom` 1,124. Conversion retains every +stream. Images become `sensor_msgs/msg/CompressedImage`, retaining the original +JPEG bytes. Keep the original SQLite file and adjacent conversion audit report. + +`go2_short.db` includes a `color_image_embedded` vector index: migrate it to +`.db`, not `.mcap`, to preserve embeddings. MCAP output rejects vectors explicitly. `alfred_fusion_short.db` converts but +has no point clouds or images. Its current Rerun renderer logs only TF/odometry +transforms, without explicit axes or visible geometry, and skips IMU; a populated +entity tree therefore does not guarantee a visible 3D scene. + +### Memory rendering, mapping and replay + +The setup above includes the Unitree helpers and plotting package used by mapping. + +The following bound visualization to five seconds and use CPU mapping: + +```sh skip +dimos mem rerun "$CDR_REVIEW_DIR/converted.mcap" --seconds 5 --no-gui --out "$CDR_REVIEW_DIR/memory.rrd" +dimos map global "$CDR_REVIEW_DIR/converted.mcap" --lidar lidar --duration 5 --device CPU:0 --block-count 10000 --no-gui --out "$CDR_REVIEW_DIR/global.rrd" +dimos map replay "$CDR_REVIEW_DIR/converted.mcap" --lidar lidar --duration 5 --map-final --map-device CPU:0 --no-gui --out "$CDR_REVIEW_DIR/replay.rrd" +``` + +There is no `dimos mem replay` command. `mem rerun` renders recorded messages, +including compressed images. `map replay` currently selects only uncompressed +`Image` streams, so this recording's JPEG images are absent from its output; +do not pass `--image color_image`. Use `memory.rrd` for images plus clouds and +`global.rrd` for the accumulated map. Neither command starts live blueprint +transport replay. + +`--lidar lidar` is necessary because this recording has two compatible cloud +streams. Automatic selection works only with a unique candidate. Its world-frame +clouds need no stored `obs.pose`; spatial dedup defaults off (`--pgo-tol 0`). +Explicit dedup/PGO requires trajectory metadata. Separate recorded odometry is +not automatically attached as `obs.pose`. Sensor-frame clouds need recorded TF +registration. Missing required data is an error, not an invented pose. + +`map global --markers` and `map replay-marker` additionally need usable camera +calibration, supported image types and poses; this checklist does not claim +marker acceptance. `map rename` and `map pose-fill` remain SQLite-only. + +### Independent validation and manual viewers + +Verify that the installed Rerun version can load each exported recording: + +```sh skip +rerun rrd verify "$CDR_REVIEW_DIR/memory.rrd" +rerun rrd verify "$CDR_REVIEW_DIR/global.rrd" +rerun rrd verify "$CDR_REVIEW_DIR/replay.rrd" +``` + +If the separately installed official Foxglove MCAP CLI is on PATH: + +```sh skip +mcap doctor "$CDR_REVIEW_DIR/converted.mcap" +``` + +The Python `mcap` package is not that executable. No installation is performed by +these commands. Doctor has not been run for this acceptance; CRC/payload checks +are not a substitute for it. + +Open the generated files locally (manual visual acceptance): + +```sh skip +rerun --serve-web --bind 127.0.0.1 --web-viewer-port 9090 "$CDR_REVIEW_DIR/memory.rrd" +``` + +Open http://127.0.0.1:9090 and select timeline `time`. Stop this process with +Ctrl-C before starting another viewer on the same port. Substitute `global.rrd` +for the accumulated map (`/world/raw_map/pointcloud`) or `replay.rrd` for animated +clouds/trajectories (timeline `ts`). Add a 3D view containing the relevant entity +if the viewer does not create it automatically. + +In Foxglove, **Open local file** `converted.mcap`; select `lidar` in a 3D panel +with fixed frame `world`, `color_image` in an Image panel, and inspect `odom` in +Raw Messages. Do not assume camera/odometry overlays align without a matching TF +chain. Existing application/session access may be required; do not upload private +recordings to obtain acceptance. Successful RRD export does not constitute human +visual review. + +### Focused regression entry + +Run this from the project test environment. The converter tests also cover +embedding migration and require PyTorch; the stairs demo above does not. + +```sh skip +python -m pytest dimos/memory/test_convert_recording.py dimos/protocol/test_cdr_mcap.py dimos/memory/store/test_mcap.py dimos/mapping/cli/test_stream_selection.py dimos/mapping/cli/test_pgo_accumulate.py dimos/navigation/go2/loop_closure/test_pgo.py +``` + +At published `6256c71becff76f9ae60af0b4076a145941bdf98`, the actual stairs +archive passed hash verification, registry preflight and full conversion of all +4,451 legacy messages. Linux CPU checks rendered 70 images in the five-second +memory view, retained 39/39 clouds in global mapping and completed map replay. +These are real legacy recording checks, not repackaged CDR fixtures. The runtime +layer separately passed 40 focused mapping/reader/PGO tests and the converter +layer 34 converter/migration/writer tests. macOS installation/execution, doctor, +human visual review, full archive migration and full-suite CI are not claimed. + +The 2026-10-09 native-library integration was also checked locally using the +existing real fixtures: all 4,451 stairs messages matched independently decoded +LCM fields and exact output envelope times; all 2,546 Go2-short observations and +108 stored 512-dimensional vectors were retained, including exact vector bytes +and old/new observation-ID associations. Both source database hashes remained +unchanged. The converter/writer regression suite passed 38 tests, including +nonempty numeric and byte arrays. These results do not establish strict malformed +CDR rejection; the accepted native-library limitations above still apply. + +The 2026-10-19 synchronized checkout was checked with a fresh Python 3.12 +environment using the core install and the two packages above. All six stairs +commands passed, retaining 4,451 messages. The five-second memory view logged +70 images, global mapping kept 39/39 clouds on CPU:0, and replay completed. All +three RRD files passed `rerun rrd verify`. The local web viewer displayed the +image and cloud data in Chrome using software rendering. This is an automated +browser check, not human visual review or Foxglove acceptance. Native hardware +and Nix builds were not validated by this demo. + +## Built-in region streams + +The converter recognizes the old mapper/planner contracts by exact stream name +and source type: `seed_map`, `map_regions`, `surface_map` (PointCloud2), +`seed_bounds`, `region_bounds` (PoseStamped), and `node_edges` (Path). +Historical `dimos//` wire topics are also recognized. Other streams +retain their ordinary type; a matching basename under an arbitrary external topic +is not sufficient to infer region semantics. + +For these legacy LCM streams, signed `Header.seq` becomes `region_id` unchanged. +Clouds remain standard PointCloud2 values inside their region envelope. Bounds +become explicit center/radius/vertical limits. Edge pose pairs become independent +start/end/weight segments; inconsistent frames, timestamps or pair weight layouts +fail conversion rather than losing fields. Endpoint Header sequence values remain +in the conversion sidecar. Empty clouds and edge arrays retain their region ID. + +MCAP's separate unsigned sequence retains the input envelope sequence. When only +a signed legacy Header sequence is available, its 32-bit bit pattern supplies that +MCAP field; the original signed value is retained in the sidecar and region ID. +Typed legacy recordings still require explicit conversion; runtime does not decode +them transparently.