diff --git a/examples/message-codegen/demo_blueprint_recording.py b/examples/message-codegen/demo_blueprint_recording.py new file mode 100644 index 00000000000..82986eb6dd2 --- /dev/null +++ b/examples/message-codegen/demo_blueprint_recording.py @@ -0,0 +1,174 @@ +# 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. + +"""Record the native relay blueprint to MCAP, stop it, then replay every CDR value.""" + +import argparse +from contextlib import ExitStack +from functools import partial +from pathlib import Path +import socket +import threading +from typing import Any +from unittest.mock import patch + +from demo_blueprint import PythonRelayDemo, blueprint +from dimos_cdrgen.registry import encode as cdr_encode +from dimos_generated.dimos_msgs.msg import LineSegments3D +from dimos_generated.geometry_msgs.msg import PoseStamped +from dimos_generated.sensor_msgs.msg import Image +from mcap.reader import make_reader +import numpy as np +from rosbags.typesys import Stores, get_types_from_msg, get_typestore + +from dimos.core.coordination.module_coordinator import ModuleCoordinator +from dimos.core.stream import In +from dimos.experimental.memory.rust_recorder import RustMcapStoreConfig, RustRecorder +from dimos.memory.recording_policy import OnExisting +from dimos.memory.replay_module import replay_module +from dimos.memory.store.mcap import McapStore +from dimos.msgs.time import to_nanoseconds +from dimos.protocol.service.system_configurator.base import configure_system +from dimos.protocol.service.zenohservice import ZenohConfig, ZenohSessionPool + + +class RelayRecorder(RustRecorder): + l2: In[LineSegments3D] + i2: In[Image] + p0: In[PoseStamped] + + +def verify_and_replay(artifact: Path, samples: int) -> None: + expected: dict[str, list[bytes]] = {} + independent = get_typestore(Stores.ROS2_JAZZY) + with artifact.open("rb") as source: + for schema, channel, row in make_reader(source).iter_messages(): + assert schema is not None and schema.encoding == "ros2msg" + assert channel.message_encoding == "cdr" + independent.register(get_types_from_msg(schema.data.decode(), schema.name)) + independent.deserialize_cdr(row.data, schema.name) + with McapStore(path=str(artifact)) as store: + for name in ("l2", "i2", "p0"): + messages = [obs.data for obs in store.stream(name)] + expected[name] = [cdr_encode(message) for message in messages] + observed_samples = set() + for message in messages: + delta = to_nanoseconds(message.header.stamp) - 1700000000123456789 + sample, remainder = divmod(delta, 100_000_000) + assert remainder == 0 + assert 0 <= sample < samples + observed_samples.add(sample) + if name == "l2": + assert message.segments[0].weight == sample + 2 + assert message.segments[0].start.x == sample + elif name == "p0": + assert ( + message.pose.position.x, + message.pose.position.y, + message.pose.position.z, + ) == (sample, 0, 0) + assert message.pose.orientation.w == 1 + else: + assert message.encoding == "rgb8" + assert (message.width, message.height, message.step) == (640, 480, 1920) + np.testing.assert_array_equal( + message.data.view(), np.full(921600, sample % 256, dtype=np.uint8) + ) + assert observed_samples == set(range(samples)), (name, observed_samples) + print( + f"MCAP {name}: verified {len(messages)} values, samples={sorted(observed_samples)}" + ) + + module = replay_module(str(artifact))(dataset=str(artifact)) + received: dict[str, list[bytes]] = {name: [] for name in expected} + done = threading.Event() + lock = threading.Lock() + + def receive(name: str, message: Any) -> None: + with lock: + received[name].append(cdr_encode(message)) + if all(len(received[key]) >= len(values) for key, values in expected.items()): + done.set() + + with ExitStack() as stack: + for name in expected: + stack.callback( + module.outputs[name].subscribe(lambda message, name=name: receive(name, message)) + ) + stack.callback(module.stop) + module.start() + assert done.wait(15), {name: len(values) for name, values in received.items()} + assert received == expected + print("Replay: every recorded CDR value matched in per-stream order after producers stopped") + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--transport", choices=("lcm", "zenoh"), default="zenoh") + parser.add_argument("--output", type=Path, required=True) + parser.add_argument("--samples", type=int, default=3) + parser.add_argument("--cpp", type=Path, default=Path("build/native-cpp-examples/cdr_relay")) + parser.add_argument("--rust", type=Path, default=Path("target/debug/cdr_relay")) + parser.add_argument("--recorder", type=Path, default=Path("target/debug/dimos-memory-recorder")) + args = parser.parse_args() + if args.samples < 1: + parser.error("--samples must be positive") + if args.output.exists(): + parser.error("--output must be a new file; existing recordings are preserved") + for executable in (args.cpp, args.rust, args.recorder): + if not executable.is_file(): + parser.error(f"Build the native executable first: {executable}") + args.output.parent.mkdir(parents=True, exist_ok=True) + recording = RelayRecorder.blueprint( + executable=str(args.recorder.resolve()), + build_command="", + cwd=str(args.recorder.resolve().parent), + store=RustMcapStoreConfig(path=str(args.output.resolve())), + on_existing=OnExisting.ERROR, + record_tf=False, + encoding_threads=2, + ) + with ExitStack() as stack: + stack.enter_context( + patch( + "dimos.protocol.service.system_configurator.base.configure_system", + partial(configure_system, check_only=True), + ) + ) + bp = blueprint(args.transport, args.cpp, args.rust, recording=recording) + if args.transport == "zenoh": + with socket.socket() as reservation: + reservation.bind(("127.0.0.1", 0)) + endpoint = f"tcp/127.0.0.1:{reservation.getsockname()[1]}" + pool = ZenohSessionPool() + stack.callback(pool.close_all) + pool.acquire( + ZenohConfig( + mode="router", listen=[endpoint], connect=[], multicast=False, gossip=False + ) + ) + bp = bp.global_config( + zenoh_mode="client", zenoh_connect=endpoint, zenoh_multicast=False + ) + coordinator = ModuleCoordinator.build(bp) + stack.callback(coordinator.stop) + producer = coordinator.get_instance(PythonRelayDemo) + for sample in range(args.samples): + print(producer.exchange(sample), flush=True) + verify_and_replay(args.output.resolve(), args.samples) + print(f"Retained self-describing {args.transport} recording: {args.output}") + + +if __name__ == "__main__": + main() diff --git a/examples/message-codegen/demo_native_recording.py b/examples/message-codegen/demo_native_recording.py new file mode 100644 index 00000000000..23a21b29931 --- /dev/null +++ b/examples/message-codegen/demo_native_recording.py @@ -0,0 +1,181 @@ +# 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. + +"""Record an independently generated custom type with an unchanged Rust binary.""" + +import argparse +from contextlib import ExitStack +import json +import os +from pathlib import Path +import socket +import subprocess +import threading +from typing import Any +import uuid + +from dimos_cdrgen.registry import decode as cdr_decode +from dimos_generated.builtin_interfaces.msg import Time +from dimos_generated.std_msgs.msg import Header +from mcap.reader import make_reader +from recording_messages.recording_msgs.msg import Reading + +from dimos.core.transport import LCMTransport, ZenohTransport +from dimos.experimental.memory.rust_recorder import RustStreamSpec +from dimos.protocol.service.zenohservice import ZenohConfig, ZenohSessionPool + + +def free_port(kind: int) -> int: + with socket.socket(socket.AF_INET, kind) as listener: + listener.bind(("127.0.0.1", 0)) + return int(listener.getsockname()[1]) + + +def stop_process(process: subprocess.Popen[bytes]) -> None: + if process.poll() is None: + process.terminate() + try: + process.wait(timeout=5) + except subprocess.TimeoutExpired: + process.kill() + process.wait(timeout=5) + + +def record(backend: str, executable: Path, output: Path) -> None: + endpoint = f"tcp/127.0.0.1:{free_port(socket.SOCK_STREAM)}" + url = f"udpm://239.255.76.67:{free_port(socket.SOCK_DGRAM)}?ttl=0" + ready, written = threading.Event(), threading.Event() + transcript: list[str] = [] + with ExitStack() as stack: + pool = ZenohSessionPool() + stack.callback(pool.close_all) + topic = f"demo/{uuid.uuid4().hex[:8]}" + publisher: Any + if backend == "lcm": + publisher = LCMTransport(topic, Reading, url=url) + else: + publisher = ZenohTransport( + topic, + Reading, + session_pool=pool, + scouting=False, + multicast=False, + gossip=False, + listen=[endpoint], + connect=[], + ) + stack.callback(publisher.stop) + publisher.start() + spec = RustStreamSpec.from_type( + port="telemetry", name="telemetry", payload_type=Reading, codec="cdr" + ) + session = ZenohConfig( + mode="client", + connect=[endpoint], + listen=[], + multicast=False, + gossip=False, + scouting=False, + connect_timeout=10, + ).to_wire() + process = stack.enter_context( + subprocess.Popen( + [str(executable.resolve())], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + env={ + **os.environ, + "DIMOS_TRANSPORT": backend, + "LCM_DEFAULT_URL": url, + "RUST_LOG": "info,dimos_memory_recorder=debug", + }, + ) + ) + + def read_logs() -> None: + assert process.stdout is not None + for raw in process.stdout: + line = raw.decode(errors="replace") + transcript.append(line) + if "memory recorder ready" in line: + ready.set() + if "memory recorder batch written" in line: + written.set() + + reader = threading.Thread(target=read_logs, name="recorder-demo-logs") + reader.start() + stack.callback(reader.join, 5) + stack.callback(stop_process, process) + assert process.stdin is not None + launch = { + "topics": {"telemetry": publisher.channel}, + "session": session if backend == "zenoh" else {}, + "config": { + "store": {"kind": "mcap", "path": str(output.resolve())}, + "encoding_threads": 2, + "streams": [spec.model_dump()], + }, + } + process.stdin.write(json.dumps(launch).encode() + b"\n") + process.stdin.close() + assert ready.wait(15), "".join(transcript) + expected = [] + for index in range(3): + message = Reading( + header=Header(stamp=Time(sec=0, nanosec=0), frame_id="map"), + value=float(index), + application_note=f"locally-added-field-{index}", + ) + message.header.stamp.sec = 1700000000 + message.header.stamp.nanosec = 123456789 + index + expected.append(message) + written.clear() + publisher.publish(message) + assert written.wait(5), "".join(transcript) + stop_process(process) + assert process.returncode == 0, "".join(transcript) + with output.open("rb") as stream: + records = list(make_reader(stream).iter_messages()) + assert len(records) == len(expected) + for (schema, channel, row), original in zip(records, expected, strict=True): + assert schema is not None and schema.encoding == "ros2msg" + assert b"application_note" in schema.data + assert channel.message_encoding == "cdr" + assert row.publish_time == row.log_time # Unknown stamp layout uses reception time. + decoded = cdr_decode(row.data, Reading) + assert decoded == original + print( + f"{backend}: {decoded.application_note}, payload stamp={decoded.header.stamp.sec}.{decoded.header.stamp.nanosec:09d}" + ) + output.with_suffix(".log").write_text("".join(transcript)) + print(f"Recorded {len(records)} independently generated messages with full schema: {output}") + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--executable", type=Path, default=Path("target/debug/dimos-memory-recorder") + ) + parser.add_argument("--output", type=Path, default=Path("build/message-codegen/demo/evidence")) + parser.add_argument("--transport", choices=("lcm", "zenoh", "both"), default="both") + args = parser.parse_args() + args.output.mkdir(parents=True, exist_ok=True) + for backend in ("lcm", "zenoh") if args.transport == "both" else (args.transport,): + record(backend, args.executable, args.output / f"external-native-recording-{backend}.mcap") + print("PASS: custom field recorded without a recorder decoder or rebuild") + + +if __name__ == "__main__": + main() diff --git a/examples/message-codegen/demo_storage_replay.py b/examples/message-codegen/demo_storage_replay.py new file mode 100644 index 00000000000..ef37885d8dd --- /dev/null +++ b/examples/message-codegen/demo_storage_replay.py @@ -0,0 +1,103 @@ +# 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. + +"""Record generated values to SQLite and MCAP, then replay SQLite through module ports.""" + +from pathlib import Path +from tempfile import TemporaryDirectory +from threading import Event, Lock +from typing import Any + +from dimos_cdrgen.registry import encode as cdr_encode, schema as cdr_schema +from dimos_generated.dimos_msgs.msg import LineSegment3D, LineSegments3D +from dimos_generated.geometry_msgs.msg import Point +from dimos_generated.std_msgs.msg import Header + +from dimos.memory.codecs.base import codec_for, codec_id +from dimos.memory.replay_module import replay_module +from dimos.memory.store.mcap import McapStore +from dimos.memory.store.sqlite import SqliteStore +from dimos.msgs.time import time_from_nanoseconds, to_nanoseconds +from dimos.protocol.cdr_mcap import CdrMcapWriter + + +def main() -> None: + messages = [ + LineSegments3D( + header=Header(frame_id="map", stamp=time_from_nanoseconds(1700000000123456789 + index)), + segments=[ + LineSegment3D( + start=Point(x=index, y=0.0, z=0.0), + end=Point(x=0.0, y=2.0, z=0.0), + weight=index + 4, + ) + ], + ) + for index in range(3) + ] + with TemporaryDirectory(prefix="dimos-cdr-storage-") as directory: + database = Path(directory) / "memory.db" + artifact = Path(directory) / "recording.mcap" + with SqliteStore(path=str(database)) as store, CdrMcapWriter(artifact) as writer: + stream = store.stream("segments", LineSegments3D) + for index, message in enumerate(messages): + stream.append(message, ts=1 + index * 0.05) + writer.write( + "segments", + cdr_encode(message), + schema_name=message.__msgtype__, + schema=cdr_schema(message.__msgtype__), + log_time_ns=1000000000 + index * 50000000, + publish_time_ns=to_nanoseconds(message.header.stamp), + ) + print(f"SQLite default codec: {codec_id(codec_for(LineSegments3D))}") + for source in [ + SqliteStore(path=str(database), must_exist=True), + McapStore(path=str(artifact)), + ]: + with source: + decoded = [observation.data for observation in source.stream("segments")] + assert decoded == messages + print( + f"{type(source).__name__}: weights={[value.segments[0].weight for value in decoded]}" + ) + replay = replay_module(str(database))(dataset=str(database)) + received: list[Any] = [] + ready = Event() + lock = Lock() + + def receive(message: LineSegments3D) -> None: + with lock: + received.append(message) + print( + f"Module replay: weight={message.segments[0].weight}, stamp={to_nanoseconds(message.header.stamp)}" + ) + if len(received) == len(messages): + ready.set() + + unsubscribe = replay.outputs["segments"].subscribe(receive) + try: + replay.start() + assert ready.wait(5), "Replay did not deliver all messages" + assert received == messages + finally: + unsubscribe() + replay.stop() + print( + "PASS: CDR persisted, reopened by type/schema, replayed through Out ports; temporary files removed" + ) + + +if __name__ == "__main__": + main() diff --git a/examples/message-codegen/evidence/blueprint-recording.md b/examples/message-codegen/evidence/blueprint-recording.md new file mode 100644 index 00000000000..fe1d61c68d4 --- /dev/null +++ b/examples/message-codegen/evidence/blueprint-recording.md @@ -0,0 +1,186 @@ +# Native blueprint recording and replay + +Verified locally on 2026-09-30 using the current runtime batch on top of +`f4a269ce4bc766a1ff876ed2b6f5ac432e6d3900`. + +- 33 recorder, SQLite path, stamped-message and TF regression tests passed. +- Strict scoped mypy passed for RustRecorder and the SQLite replay helper. +- The actual Python -> C++ -> Rust -> Python Zenoh relay produced three + custom LineSegments3D samples, three 640x480 RGB images, and three PoseStamped + samples. Native edits incremented line weight twice; image bytes were exact. +- Rust recorder reported received=9, written=9, encode_errors=0 and exited cleanly. +- MCAP had CDR channels and embedded ros2msg schemas. Independent rosbags Jazzy + decoding succeeded for every row; all three streams contained samples 0,1,2. +- After all producers stopped, DimOS ReplayModule emitted every recorded value + with byte equality and per-stream order preserved. + +The retained local artifact is +`build/message-codegen/demo/evidence/blueprint-current-zenoh-v3.mcap`; terminal +log is `/tmp/cdr-blueprint-recording-v3.log`. Earlier failed artifacts were +preserved. This demo uses installed built-in custom messages; arbitrary installed +external-message capture has separate acceptance requirements. + +The first recorder run exposed inherited TF wiring being passed despite +`record_tf=False`. RustRecorder now filters native launch topics to declared +recording streams; regression tests cover TF enabled and disabled. Another run +exposed float-second replay treating 1 ns spaced epoch timestamps as static. +The demo now uses realistic 100 ms intervals while checking exact nanoseconds; +sub-float-resolution scheduling is not claimed. + +No hardware ran and host tuning was check-only. Default-interface LCM, human +Foxglove/Rerun inspection, full dependency retirement and final installed-package +acceptance remain open. This evidence does not mark stages 5 or 6 complete. + +## Follow-up regression batch + +The published parent `f4a269ce4` codegen workflow 36763634296 succeeded. Main +workflow 36763630343 ended cancelled after ARM reported 6170 passed, 228 skipped +and one documentation-branding failure. Both offending spellings in +`docs/usage/lcm.md` are repaired; other Python jobs were cancelled, not passed. + +Follow-up local checks: 15 Joy/Header/LineSegments3D/branding tests and 53 +stamped-covariance/time tests passed. Migrated suites retain numeric payloads, +empty and large sequences, covariance patterns, copy independence, explicit +clock and datetime conversions, and add independent ROS CDR checks. Legacy +convenience constructors, custom string formatting, inheritance, Path-based +line packing and implicit timestamps are intentionally retired. Generated ROS +time overflow now reports an explicit ValueError; both int32 boundaries and +values immediately outside them are covered. Scoped mypy passed time.py. + +## Future cross-stack compatibility note + +Parent review on 2026-09-30 identified EpisodeStatus in the independent open +chain #3854 -> #3855 -> #3921 -> #3931 -> #3942. It is used by typed live Out, +collection recorder input/source timestamp extraction, offline episode boundaries +and TUI RPC. That type is absent from this CDR checkout. Its eventual integration +would need a coordinated generated CDR design if that proposal is adopted; a JSON substitution would not satisfy +the typed CDR recorder contract. The user has not selected the integration design. +No code or commits from that chain were imported or changed. This is an explicit +future compatibility note, not a regression introduced by this branch. +CDR is not a prerequisite for merging that separate stack: it targets current +main and remains independent. No merge is authorized for this CDR draft. + +## Image and value-type retirement follow-up + +Local verification after `0bac7f454`: 52 Path/basic-covariance checks, five +CameraInfo calibration checks, and 58 generated-image/compression/detection/heavy-import +checks passed. Strict mypy passed image.py, Rerun message_helpers.py and both +manual message tools. The original JPEG/JXL mean-error and compressed-size +thresholds, lossless PNG/JXL depth, codec effort comparison, camera calibration +values, Rerun box geometry/labels and path sequence behavior are retained. + +PNG/JXL encoding and decoding live in external functions over generated values. +Source headers and 16UC1 depth units survive, including big-endian inputs. Raw +64FC1, 16SC1 and 32FC3 views remain supported; unsupported compression types are +rejected explicitly. The local venv was missing the existing lockfile dependency +imagecodecs 2026.6.26; its 27.5 MiB wheel was installed without changing project +dependencies or system packages. + +Image file/selection tests use deterministic temporary pixels and real sharpness +calculation in virtual-time windows. The manual pickle tool retains only its +raw-array input path, converting to generated images after loading; no old typed +message decoding was added. Manual publishers and hardware tools were not run. + +## Geometry and map follow-up + +Local checks: 42 wrench/geometry, 12 odometry, 78 point-cloud/viewer and 32 occupancy +tests passed. Strict scoped mypy passed geometry.py, pointcloud.py, occupancy.py +and the Rerun helper. Force/torque arrays and axis-aligned finite cloud bounds +are external functions; generated types stay value containers. Point-field tests +retain intensity, zero offset_time, tag, line and all original overlap scenarios. +Occupancy tests retain unknown/free/occupied counts, threshold filtering, origin +yaw, coordinate conversion and obstacle-preserving reduction. Borrowed views +remain read-only; tests mutate declared message data explicitly. ROS float32 +resolution is compared with numerical tolerances instead of assuming Python +float64 storage. + +## Generated geometry regression continuation + +The Vector3, Quaternion, Pose, Transform, and Twist suites now use generated +values and explicit external math. Non-unit Hamilton products/inverses, zero +errors, rotated local offsets, frame composition, independent ROS decoding, +and original numerical tolerances remain covered. Constructor coverage now +checks explicit nested fields, deep-copy isolation, covariance lengths and +layout, zero/default timestamps, exact nanoseconds, and rejection of retired +polymorphic fields. Stamped values are distinct from their payloads. + +Offline message-directory validation: **614 passed** with only +`tf2_msgs/test_TFMessage_lcmpub.py` excluded. The attempted whole-directory run +also produced 614 passes, but that real multicast test failed host LCM self-test +and thread teardown; this remains an environment-dependent gate, not a pass. +No host routing/firewall configuration was changed. Strict scoped mypy for +`dimos/msgs/geometry.py` and repository hooks passed. + +Published commit `0bac7f454bcd2f61766c0246e38a40016c62746c` has terminal success +for main CI 36765815202 and codegen CI 36765824915. Self-hosted lanes were +skipped. These results do not cover subsequent local commits until their own +exact-HEAD workflows complete. + +## Python dependency retirement and remaining consumers + +The 47 obsolete Python message implementation files have been removed after +migrating their remaining Python imports. The root project now declares raw +`lcm-dimos-fork` directly and no longer depends on `dimos-lcm`. The task venv's +installed `dimos-lcm` was also uninstalled: **624 offline message/voxel tests +passed without it**. The lock removes only `dimos-lcm` and its unused +`foxglove-websocket` dependency; no remaining package versions changed. +Re-resolution required restricting the existing optional `a750-control` wheel +to Python 3.12, its only published/locked ABI, instead of requesting it on 3.10/3.11. + +The retired typed-pickle voxel regression fixture is now deterministic CDR +geometry, with exact voxel counts at three resolutions, repeated-ingestion +invariance, numerical coordinates, and range preservation. Raw PLY/image +occupancy fixtures remain supported. Full local test collection reached 6555 +tests with 16 collection errors for absent optional dependencies (coacd, h5py, +open_clip, plotext, reportlab, requests_mock, trimesh, ultralytics, yourdfpy), +not obsolete message imports. This is not a full-suite pass. + +Remaining examples and the optional LeRobot runtime use generated image, +pose, point-cloud and UInt32 button fields. The runtime copies image pixels +before retaining an observation. Its own scoped mypy configuration passes, +but its behavioral suite still requires the absent LeRobot environment. +The virtual robot's pure unicycle integration passed a numerical check; +the C++ controller compiled against the installed CMake message package and +raw LCM, without being run. The obsolete Lua and TypeScript LCM-codec demos +are retired explicitly; Lua generation is deferred and the browser CDR SDK +is the supported web path. Transform documentation passed 5 executable Python +blocks; eval documentation passed all 6 blocks (4 no-result, 2 with results). + +Outstanding dependency gate: the M20 onboard DrDDS/Zenoh C++ bridge still +fetches/uses old message headers. The local environment lacks the vendor DrDDS +SDK, and its `sensor_msgs::msg`/`nav_msgs::msg` types collide with generated C++ +type names. A verified vendor boundary/codec migration is still required; +no board build, deployment or robot operation has been performed. Historical +manual replay fixtures and final viewer/transport acceptance also remain open. + +The shared Go2 moment fixture now takes an explicit CDR SQLite path; its tests +create generated recordings instead of resolving the old pickle dataset name. +The manual memory import/query tool now copies 1024 deterministic CDR samples +into an isolated destination. Its original >1000 counts, >10-second duration, +pagination, ordering, overlap, lazy decoding and pose assertions remain. +All **18 tool checks passed**, including cached CPU CLIP embedding/search with +model network access disabled; no new weight download was needed. These +synthetic image results validate the pipeline, not real-scene recognition. + +The separate real detection fixture `unitree_go2_lidar_corrected` is absent +locally and its LFS archive is 1,212,727,745 bytes. Regenerating it while retaining +its semantic assertions is still an explicit outstanding fixture gate. + +### DimSim standalone codec audit + +The remaining browser/Deno `@dimos/msgs` consumers now use pinned Foxglove CDR +codecs and a standalone full-schema bundle exported from canonical generated +messages. RGB publishes raw RGBA Image, depth publishes 16UC1 Image, and lidar, +odometry and cmd_vel use PointCloud2/PoseStamped/Twist with ROS 2 channel names. +The existing LCM transport and LC02 WebSocket envelope remain. JPEG stays only +in eval/sidebar previews. No legacy decoding fallback was introduced. + +Verified locally: Vite production build; complete Deno CLI type check with a +frozen lock preserving unrelated original dependency versions; three socket-free +codec tests (large image envelope, all velocity components, malformed input); +canonical-schema drift pytest; strict Python exporter typing; pre-commit checks. +Four Deno-encoded payloads (Twist, PoseStamped, Image and PointCloud2) decoded in +both generated Python and independent rosbags with exact values/source stamps; +a generated Python Twist decoded in Deno with exact command components. +Full browser/physics simulation acceptance remains unverified. These checks +neither operate hardware nor prove host multicast or end-to-end rendering. diff --git a/examples/message-codegen/evidence/remaining-consumer-cutover.md b/examples/message-codegen/evidence/remaining-consumer-cutover.md new file mode 100644 index 00000000000..e73f7aba6d8 --- /dev/null +++ b/examples/message-codegen/evidence/remaining-consumer-cutover.md @@ -0,0 +1,5 @@ +## Remaining consumer cutover + +See [final-consumer-cutover.md](evidence/final-consumer-cutover.md) for generated +entity/control/PGO/eval migration, exact previous CI failures, offline evidence +and remaining legacy retirement boundaries. diff --git a/examples/message-codegen/test_native_recording.py b/examples/message-codegen/test_native_recording.py new file mode 100644 index 00000000000..b60c99c25d1 --- /dev/null +++ b/examples/message-codegen/test_native_recording.py @@ -0,0 +1,89 @@ +# 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. + +import os +from pathlib import Path +import subprocess +import sys + +from dimos_cdrgen.generate import generate +from dimos_cdrgen.ownership import Dependency +from dimos_generated_schemas.provider import package_root +import pytest + + +@pytest.mark.native_e2e +@pytest.mark.parametrize("backend", ["lcm", "zenoh"]) +def test_external_schema_records_on_both_transports_with_unchanged_binary(tmp_path, backend): + executable = Path( + os.environ.get("DIMOS_NATIVE_RECORDER_BIN", "target/debug/dimos-memory-recorder") + ).resolve() + if not executable.is_file(): + pytest.skip("Build the native recorder before running the native_e2e recording gate") + source = tmp_path / "interfaces/recording_msgs/msg/Reading.msg" + source.parent.mkdir(parents=True) + source.write_text("std_msgs/Header header\nfloat64 value\nstring application_note\n") + builtin = Dependency.load(Path(package_root())) + package = tmp_path / "generated" + generate( + [source.parents[2]], + package, + ["recording_msgs/msg/Reading"], + "recording_messages", + dependencies=(builtin,), + languages=("python",), + ) + site = tmp_path / "site" + subprocess.run( + [ + sys.executable, + "-m", + "pip", + "install", + "--no-deps", + "--no-build-isolation", + "--target", + str(site), + str(package / "python"), + ], + check=True, + capture_output=True, + text=True, + timeout=60, + ) + env = { + **os.environ, + "PYTHONPATH": os.pathsep.join( + [str(site), str(Path.cwd()), os.environ.get("PYTHONPATH", "")] + ), + } + result = subprocess.run( + [ + sys.executable, + str(Path(__file__).with_name("demo_native_recording.py")), + "--executable", + str(executable), + "--transport", + backend, + "--output", + str(tmp_path), + ], + capture_output=True, + text=True, + env=env, + timeout=60, + ) + assert result.returncode == 0, result.stdout + result.stderr + assert f"{backend}: locally-added-field-2" in result.stdout + assert "PASS:" in result.stdout diff --git a/examples/message-codegen/viewer-checker/check-native-recording.mjs b/examples/message-codegen/viewer-checker/check-native-recording.mjs new file mode 100644 index 00000000000..efb5cb7f03e --- /dev/null +++ b/examples/message-codegen/viewer-checker/check-native-recording.mjs @@ -0,0 +1,40 @@ +import assert from 'node:assert/strict'; +import { readFile } from 'node:fs/promises'; +import { McapIndexedReader } from '@mcap/core'; +import { parse, fixupTypes } from '@foxglove/rosmsg'; +import { MessageReader } from '@foxglove/rosmsg2-serialization'; +import zstd from '@foxglove/wasm-zstd'; + +await zstd.isLoaded; +const bytes = await readFile(process.argv[2]); +const reader = await McapIndexedReader.Initialize({ + readable: { + size: async () => BigInt(bytes.length), + read: async (offset, size) => bytes.subarray(Number(offset), Number(offset + size)), + }, + decompressHandlers: { zstd: (data, size) => zstd.decompress(data, Number(size)) }, +}); +assert.equal(reader.header.profile, 'ros2'); +const decoders = new Map(); +for (const schema of reader.schemasById.values()) { + assert.equal(schema.encoding, 'ros2msg'); + const definitions = parse(new TextDecoder().decode(schema.data), { ros2: true, skipTypeFixup: true }); + definitions[0].name = schema.name.replace('/msg/', '/'); + fixupTypes(definitions); + decoders.set(schema.id, new MessageReader(definitions)); +} +let count = 0; +for await (const message of reader.readMessages({ validateCrcs: true })) { + const channel = reader.channelsById.get(message.channelId); + assert.equal(channel.messageEncoding, 'cdr'); + assert.equal(channel.topic, 'telemetry'); + const value = decoders.get(channel.schemaId).readMessage(message.data); + assert.equal(value.application_note, `locally-added-field-${count}`); + assert.equal(value.header.stamp.sec, 1700000000); + assert.equal(value.header.stamp.nanosec, 123456789 + count); + assert.equal(message.logTime, message.publishTime); + console.log(`Foxglove decoded ${value.application_note} from embedded schema.`); + count += 1; +} +assert.equal(count, 3); +console.log('PASS: native recorder custom messages need no viewer-side message package');