From 25dc75047dc420a8ba34602044aaf3dfa95520eb Mon Sep 17 00:00:00 2001 From: cc Date: Tue, 20 Oct 2026 22:28:32 -0700 Subject: [PATCH] feat(recordings): restack offline CDR recording conversion --- dimos/memory/cli/app.py | 39 ++ dimos/memory/convert_recording.py | 602 +++++++++++++++++++++++ dimos/memory/recording_migration.py | 161 ++++++ dimos/memory/test_convert_recording.py | 479 ++++++++++++++++++ dimos/memory/test_recording_migration.py | 119 +++++ docs/development/convert-recordings.md | 79 +++ 6 files changed, 1479 insertions(+) create mode 100644 dimos/memory/convert_recording.py create mode 100644 dimos/memory/recording_migration.py create mode 100644 dimos/memory/test_convert_recording.py create mode 100644 dimos/memory/test_recording_migration.py create mode 100644 docs/development/convert-recordings.md 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..88fb44577e9 --- /dev/null +++ b/dimos/memory/convert_recording.py @@ -0,0 +1,602 @@ +# 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 ExitStack, 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, + message_types, + schema as cdr_schema, +) +import lz4.frame +from mcap.reader import make_reader +from mcap.records import Attachment, Channel, Metadata, Schema +from mcap.stream_reader import StreamReader +from mcap.writer import Writer +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 + +# 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 +} +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 target(self) -> Any: + name = "sensor_msgs/msg/CompressedImage" if self.codec.endswith("jpeg") else self.ros_type + result = message_types().get(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 = message_types().get(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 = message_types().get(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": + if len(data) < 4 or data[:2] not in (b"\x00\x00", b"\x00\x01"): + raise ValueError("Expected standalone XCDR1 encapsulation") + 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") + return _convert_fields(old, sequences), sequences + + +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}) + with ExitStack() as resources, sidecar.open("w") as audit: + if destination.suffix == ".mcap": + handle = resources.enter_context(temp.open("wb")) + writer: Any = Writer(handle) + writer.start(profile="ros2", library="dimos") + resources.callback(writer.finish) + channels: dict[str, int] = {} + else: + writer = resources.enter_context(SqliteStore(path=str(temp))) + audit.write( + json.dumps({"source": str(source), "streams": [s.__dict__ for s in streams]}) + + "\n" + ) + if destination.suffix == ".mcap": + for stream in streams: + schema_id = writer.register_schema( + stream.target.__msgtype__, + "ros2msg", + cdr_schema(stream.target.__msgtype__).encode(), + ) + channels[stream.name] = writer.register_channel( + stream.name, + "cdr", + schema_id, + metadata={ + "offered_qos_profiles": "[]", + **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 = getattr(value, "header", None) + 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)) + ) + 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.add_message( + channel_id=channels[row.stream.name], + data=payload, + log_time=row.log_ns, + publish_time=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..de20554ef0b --- /dev/null +++ b/dimos/memory/test_convert_recording.py @@ -0,0 +1,479 @@ +# 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 hashlib +import json +from pathlib import Path +import sqlite3 +import sys + +from dimos_cdrgen.registry import decode as cdr_decode, encode as cdr_encode, schema as cdr_schema +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.messages.std_msgs.msg import String +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"] + + +def test_cdr_input_with_wrong_encapsulation_publishes_no_converted_output(tmp_path): + source, output = tmp_path / "bad-cdr.mcap", tmp_path / "converted.mcap" + with source.open("wb") as handle: + writer = Writer(handle) + writer.start() + schema_id = writer.register_schema( + "std_msgs/msg/String", "ros2msg", cdr_schema("std_msgs/msg/String").encode() + ) + channel = writer.register_channel("status", "cdr", schema_id) + payload = cdr_encode(String("body is otherwise valid")) + writer.add_message(channel, 1, b"bad!" + payload[4:], 1) + writer.finish() + original = source.read_bytes() + with pytest.raises(ValueError, match="XCDR1 encapsulation"): + convert(source, output) + assert source.read_bytes() == original + assert not output.exists() + assert not Path(str(output) + ".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..482708449bf --- /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 +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.messages.builtin_interfaces.msg import Time +from dimos.messages.geometry_msgs.msg import Point, Pose, PoseStamped, Quaternion +from dimos.messages.std_msgs.msg import Header +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..c5730e51b04 --- /dev/null +++ b/docs/development/convert-recordings.md @@ -0,0 +1,79 @@ +# Convert an existing recording to CDR + +Conversion is an explicit offline command. It never starts a robot, downloads data, +or negotiates an old runtime wire format. Install this proposal's DimOS package; +built-in generated messages are included in it. The historical LCM decoder comes from the retained `dimos-lcm>=0.1.6` +dependency while runtime migration is incomplete. + +Use your existing local MCAP or memory SQLite file and choose a **new** output name: + +```sh skip +dimos mem convert old.db converted.mcap --dry-run +dimos mem convert old.db converted.mcap +dimos mem summary converted.mcap +``` + +`--dry-run` checks declared types, schemas and row counts; payload integrity is +checked during conversion. It creates no output. Read the JSON report and resolve +any blocked inputs before converting. No compressed archive is downloaded or +extracted by these commands. + +Either container can be an input or output: + +```sh skip +dimos mem convert old.mcap converted.db +dimos mem convert old.db converted.db +``` + +MCAP is the portable recording format, with full dependency `ros2msg` schemas and +CDR payloads embedded. SQLite is the writable analysis/vector-search format. Vector +embeddings require SQLite output; attempting to export them to MCAP fails rather +than dropping them. Vector preservation uses the existing Embedding implementation +and its PyTorch prerequisite, loaded only if vectors are present. + +An explicit directory can be migrated while preserving relative filenames: + +```sh skip +dimos mem convert ./old-recordings ./cdr-recordings --format mcap --dry-run +dimos mem convert ./old-recordings ./cdr-recordings --format mcap +``` + +The new output directory must not overlap the source tree. Every candidate is +preflighted before any conversion. Unsupported inputs are listed as blocked. +If a payload fails after a batch has started, completed files remain; the command +returns nonzero and `migration-summary.json` records converted, failed and +not-attempted files. A batch is not an atomic transaction across files. + +## Fidelity and failure behavior + +- Originals are opened read-only. Existing outputs, reports and racing output + writers are never overwritten. A single-file failure publishes no output. +- All mapped fields, exact Header seconds/nanoseconds and frame IDs are preserved. + LCM array length fields are validated; CDR encodes the corresponding sequences. +- MCAP log time, publish time and envelope sequence retain their exact integer values. + SQLite observation timestamps retain their original floating-point precision. +- Legacy ROS1 `Header.seq`, SQLite observation IDs/value/pose/tags and container + metadata are retained in the mandatory adjacent `.conversion.jsonl` audit file. + SQLite output additionally carries conversion metadata in tags, spatial poses + and vector-to-observation associations. IDs may change and are mapped in the audit. +- An explicitly declared JPEG Image maps to standard CompressedImage without + decoding or recompressing its bytes. JSON String retains its UTF-8 document. + +The allowlist lives in `dimos/memory/convert_recording.py`. Supported standard +geometry, sensor, navigation, TF, time and scalar message shapes use the installed +historical decoder and generated CDR registry. Unknown/custom/proprietary types, +pickle, raw captures, external blob/vector stores and MCAP attachments are rejected. +No class named by a recording is imported or instantiated. Declared CDR input must +match the installed full schema. Untrusted native-format recordings are not a +sandboxed input format; use recordings from trusted producers. + +Open `converted.mcap` as a local file in Foxglove. +Foxglove may require your existing desktop installation or signed-in browser; +conversion does not create accounts or upload a recording. Viewer and human +acceptance remain separate from successful data conversion. + +The focused regression suite runs without data downloads or hardware: + +```sh +uv run pytest --noconftest --import-mode=importlib -o addopts='' -q dimos/memory/test_convert_recording.py dimos/memory/test_recording_migration.py +```