diff --git a/.github/workflows/cdr-native.yml b/.github/workflows/cdr-native.yml new file mode 100644 index 00000000000..55c20c9980f --- /dev/null +++ b/.github/workflows/cdr-native.yml @@ -0,0 +1,69 @@ +name: CDR native interoperability + +on: + pull_request: + workflow_dispatch: + push: + branches: [main] + +permissions: + contents: read + +jobs: + native: + runs-on: ubuntu-latest + timeout-minutes: 30 + env: + DIMOS_ALLOW_MISSING_COCKPIT: "1" + CARGO_TARGET_DIR: ${{ github.workspace }}/target + steps: + - uses: actions/checkout@v7 + with: + lfs: false + persist-credentials: false + - uses: astral-sh/setup-uv@v10.2.0 + with: + python-version: '3.12' + enable-cache: true + - uses: denoland/setup-deno@v2 + with: + deno-version: v2.6.10 + - name: Install build and media dependencies + run: | + sudo apt-get update + sudo apt-get install -y cmake pkg-config clang libclang-dev portaudio19-dev libturbojpeg ffmpeg + - name: Install Python project and test dependencies + run: uv sync --frozen --group tests + - name: Test SDK and build native consumers + run: | + cargo test --locked --manifest-path native/rust/dimos-module/Cargo.toml + cargo test --locked --manifest-path native/rust/dimos-module-macros/Cargo.toml + cargo test --locked --manifest-path dimos/mapping/ray_tracing/rust/Cargo.toml --lib + cargo test --locked --manifest-path dimos/navigation/global_planner/mls_planner/rust/Cargo.toml --lib + cargo test --locked --manifest-path dimos/navigation/local_planner/rust/Cargo.toml --features module,test-bodies + cargo test --locked --manifest-path dimos/navigation/trajectory_follower/fancy/rust/Cargo.toml --features module + cargo test --locked --manifest-path dimos/hardware/sensors/camera/v4l2/rust/Cargo.toml --lib + cargo check --locked --manifest-path dimos/perception/depth2depth_cloud/rust/Cargo.toml --lib + cargo check --locked --manifest-path dimos/hardware/sensors/camera/realsense/rust/Cargo.toml --all-targets --features docs-only + cargo build --locked --manifest-path dimos/mapping/ray_tracing/rust/Cargo.toml --bin voxel_ray_tracing + cargo build --locked --manifest-path dimos/navigation/global_planner/mls_planner/rust/Cargo.toml --bin mls_planner + cargo build --locked --manifest-path dimos/navigation/local_planner/rust/Cargo.toml --features module --bin local_planner + cargo build --locked --manifest-path dimos/navigation/trajectory_follower/fancy/rust/Cargo.toml --features module --bin trajectory_follower + - name: Prepare DimSim dependencies + working-directory: misc/DimSim/cli + run: deno install --node-modules-dir=auto --entrypoint bridge/cdr_test.ts + - name: Exchange generated messages with real native and TypeScript consumers + run: | + export DIMOS_NATIVE_RAY_TRACING_BINARY="$CARGO_TARGET_DIR/debug/voxel_ray_tracing" + export DIMOS_NATIVE_MLS_BINARY="$CARGO_TARGET_DIR/debug/mls_planner" + export DIMOS_NATIVE_LOCAL_PLANNER_BINARY="$CARGO_TARGET_DIR/debug/local_planner" + export DIMOS_NATIVE_FOLLOWER_BINARY="$CARGO_TARGET_DIR/debug/trajectory_follower" + export DIMOS_DIMSIM_DENO="$(command -v deno)" + uv run --no-sync pytest --noconftest --import-mode=importlib -o addopts='' -q \ + dimos/mapping/ray_tracing/test_generated_native.py \ + dimos/navigation/global_planner/mls_planner/test_generated_native.py \ + dimos/navigation/local_planner/test_generated_native.py \ + dimos/robot/unitree/test_dimsim_connection.py \ + dimos/navigation/trajectory_follower/fancy/test_generated_native.py \ + dimos/hardware/sensors/camera/realsense/test_camera.py \ + dimos/perception/depth2depth_cloud/test_depth2depth_cloud_config.py diff --git a/bin/generate-native-tf-messages b/bin/generate-native-tf-messages new file mode 100755 index 00000000000..85706252620 --- /dev/null +++ b/bin/generate-native-tf-messages @@ -0,0 +1,42 @@ +#!/usr/bin/env python3 +# 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. + +"""Regenerate the SDK's standard TF message crate with dimos-cdrgen 0.1.2.""" + +from importlib.metadata import version +from pathlib import Path +import shutil +from tempfile import TemporaryDirectory + +from dimos_cdrgen.generate import generate + +root = Path(__file__).resolve().parents[1] +crate = root / "native/rust/dimos-tf-messages" +if version("dimos-cdrgen") != "0.1.2": + raise RuntimeError("Install dimos-cdrgen==0.1.2 before regenerating TF messages") +with TemporaryDirectory() as temporary: + output = Path(temporary) / "generated" + generate( + [crate / "interfaces"], + output, + ["tf2_msgs/msg/TFMessage"], + module="dimos_tf", + version="0.1.0", + languages=("rust",), + ) + shutil.copytree(output / "rust", crate, dirs_exist_ok=True) + generated_lib = crate / "src/lib.rs" + license_header = (crate / "build.rs").read_text().split("use ros2msg", 1)[0] + generated_lib.write_text(license_header + generated_lib.read_text()) diff --git a/native/message-source.nix b/native/message-source.nix new file mode 100644 index 00000000000..d54683ac53e --- /dev/null +++ b/native/message-source.nix @@ -0,0 +1,52 @@ +# Copyright 2026 Dimensional Inc. +# SPDX-License-Identifier: Apache-2.0 + +# Generate the built-in definitions through the same configuration and core as pip. +# Nix fetches fixed-hash inputs; the generation builder itself is offline. +{ pkgs }: +let + generatorWheel = pkgs.fetchurl { + url = "https://files.pythonhosted.org/packages/29/9f/a4fe6f8ea4dcafd3001c868ebbc751bb0af83aecb1fe6523765c12cdedba/dimos_cdrgen-0.1.2-py3-none-any.whl"; + sha256 = "9857c8cc40de58b03f94cb81e7d27ae3caac98101c96cdaa2965413a332e685f"; + }; + generator = pkgs.runCommand "dimos-cdrgen-0.1.2-source" { nativeBuildInputs = [ pkgs.unzip ]; } '' + mkdir -p $out + unzip -q ${generatorWheel} -d $out + ''; + python = pkgs.python3.withPackages (p: [ + p.setuptools p.wheel p.pydantic p.empy p.lark p.rosbags p.numpy p.requests + ]); + product = pkgs.lib.fileset.toSource { + root = ../.; + fileset = pkgs.lib.fileset.unions [ + ../pyproject.toml + ../dimos/message_codegen/build.py + ../packages/dimos-generated/schema-sources.json + ../packages/dimos-generated/custom + ]; + }; + schemaLock = builtins.fromJSON (builtins.readFile ../packages/dimos-generated/schema-sources.json); + schemaArchives = map (source: { + name = builtins.hashString "sha256" source.url + ".tar.gz"; + archive = pkgs.fetchurl { inherit (source) url sha256; }; + }) schemaLock.sources; + source = pkgs.runCommand "dimos-generated-message-source" { nativeBuildInputs = [ python ]; } '' + cp -r ${product} product + chmod -R u+w product + mkdir -p cache/archives + ${pkgs.lib.concatMapStringsSep "\n" (item: "cp ${item.archive} cache/archives/${item.name}") schemaArchives} + export PYTHONPATH="$PWD/product:${generator}:$PYTHONPATH" + python3 - <<'PYTHON' +from pathlib import Path +from dimos.message_codegen.build import prepare_builtin_sources +from dimos_cdrgen.source_package import prepare +prepare_builtin_sources(Path("cache"), offline=True) +prepare(Path("product")) +PYTHON + cp -r product/build/message-source $out + test -f $out/cpp/CMakeLists.txt + test -f $out/rust/Cargo.toml + ''; +in { + inherit generator python source; +} diff --git a/native/rust/dimos-module-macros/src/lib.rs b/native/rust/dimos-module-macros/src/lib.rs index af89a0680b4..ff19a4af1ad 100644 --- a/native/rust/dimos-module-macros/src/lib.rs +++ b/native/rust/dimos-module-macros/src/lib.rs @@ -454,20 +454,33 @@ fn port_decls(classified: &[ClassifiedField], want_input: bool) -> Vec FieldKind::Tf => want_input, _ => false, }) - .map(|f| PortDecl { - name: f.name.to_string(), - // `Tf` carries no generic payload. The port is TFMessage by - // construction. - // An explicit `msg` wins: a view type's wire payload and the dimos - // message it stands for are allowed to differ. - ty: match f.kind { - FieldKind::Tf => TF_PAYLOAD_TYPE.to_string(), - _ => f - .msg - .clone() - .or_else(|| port_payload_type(f.ty)) - .unwrap_or_default(), - }, + .flat_map(|f| { + let port = PortDecl { + name: f.name.to_string(), + // `Tf` carries no generic payload. The port is TFMessage by + // construction. + // An explicit `msg` wins: a view type's wire payload and the dimos + // message it stands for are allowed to differ. + ty: match f.kind { + FieldKind::Tf => TF_PAYLOAD_TYPE.to_string(), + _ => f + .msg + .clone() + .or_else(|| port_payload_type(f.ty)) + .unwrap_or_default(), + }, + }; + if matches!(f.kind, FieldKind::Tf) { + vec![ + port, + PortDecl { + name: "tf_static".into(), + ty: TF_PAYLOAD_TYPE.into(), + }, + ] + } else { + vec![port] + } }) .collect() } @@ -537,7 +550,7 @@ fn check_port_table(entry: &toml::Value, kind: &str, declared: &[PortDecl]) -> R let msg = msg .as_str() .ok_or_else(|| format!("`{kind}.{}` must be a string", port.name))?; - let want = msg.rsplit('.').next().unwrap_or(msg); + let want = msg.rsplit(['.', '/']).next().unwrap_or(msg); if want != port.ty { return Err(format!( "port `{}` is declared `{msg}` in `{kind}` but the struct field carries `{}`", @@ -649,6 +662,12 @@ fn classify_field(field: &Field, name: &Ident) -> syn::Result { if found.is_some() { return Err(syn::Error::new_spanned(attr, ONE_ATTR_ONLY)); } + if !matches!(&attr.meta, syn::Meta::Path(_)) { + return Err(syn::Error::new_spanned( + attr, + "#[tf] uses standard ROS TF messages and takes no codec arguments", + )); + } found = Some(FieldKind::Tf); } } @@ -736,6 +755,36 @@ mod tests { }; use syn::parse_quote; + #[test] + fn tf_accepts_legacy_or_explicit_codec_pair_and_keeps_registry_type() { + for field in [parse_quote!(#[tf] tf: Tf)] { + let attr = super::classify_field(&field, &parse_quote!(tf)).expect("valid TF field"); + let ty: syn::Type = parse_quote!(Tf); + let name = parse_quote!(tf); + let fields = [ClassifiedField { + name: &name, + ty: &ty, + kind: attr.kind, + msg: attr.msg, + }]; + let ports = super::port_decls(&fields, true); + assert_eq!(ports.len(), 2); + assert_eq!(ports[0].ty, "TFMessage"); + assert_eq!(ports[1].name, "tf_static"); + } + } + + #[test] + fn tf_rejects_partial_or_unknown_codec_settings() { + for field in [ + parse_quote!(#[tf(decode = wire::decode)] tf: Tf), + parse_quote!(#[tf(encode = wire::encode)] tf: Tf), + parse_quote!(#[tf(codec = wire::codec)] tf: Tf), + ] { + assert!(super::classify_field(&field, &parse_quote!(tf)).is_err()); + } + } + #[test] fn io_ports_are_rejected_by_the_registry_check() { let name: syn::LitStr = parse_quote!("demo"); @@ -790,6 +839,29 @@ mod tests { .expect("matching ports should pass"); } + #[test] + fn accepts_canonical_cdr_names_and_still_rejects_type_mismatch() { + let manifest = MANIFEST.replace("sensor_msgs.PointCloud2", "sensor_msgs/msg/PointCloud2"); + check_manifest_ports( + &manifest, + "demo", + &[port("lidar", "PointCloud2")], + &[port("global_map", "PointCloud2")], + ) + .expect("canonical ROS message names match their imported Rust type"); + let err = check_manifest_ports( + &manifest, + "demo", + &[port("lidar", "Image")], + &[port("global_map", "PointCloud2")], + ) + .expect_err("canonical names must still enforce the payload type"); + assert!( + err.contains("sensor_msgs/msg/PointCloud2") && err.contains("Image"), + "{err}" + ); + } + #[test] fn rejects_missing_metadata_section() { let err = check_manifest_ports(MANIFEST, "other", &[], &[]) diff --git a/native/rust/dimos-module/Cargo.lock b/native/rust/dimos-module/Cargo.lock index cbd83d0f707..7143d4af75a 100644 --- a/native/rust/dimos-module/Cargo.lock +++ b/native/rust/dimos-module/Cargo.lock @@ -249,6 +249,15 @@ dependencies = [ "shlex", ] +[[package]] +name = "cdr" +version = "0.2.4" +source = "git+https://github.com/TomCC7/cdr-rs?rev=38a4368255d408faf374726653b39d4a4e9735cb#38a4368255d408faf374726653b39d4a4e9735cb" +dependencies = [ + "byteorder", + "serde", +] + [[package]] name = "cfg-if" version = "1.0.4" @@ -582,7 +591,7 @@ dependencies = [ [[package]] name = "dimos-lcm" version = "0.1.0" -source = "git+https://github.com/dimensionalOS/dimos-lcm.git?branch=rust-codegen#04d78e8622500244123ba9cefa4c51b4cb454549" +source = "git+https://github.com/dimensionalOS/dimos-lcm.git?rev=04d78e8622500244123ba9cefa4c51b4cb454549#04d78e8622500244123ba9cefa4c51b4cb454549" dependencies = [ "byteorder", "socket2 0.5.10", @@ -591,10 +600,12 @@ dependencies = [ [[package]] name = "dimos-module" -version = "0.1.0" +version = "0.2.0" dependencies = [ + "cdr", "dimos-lcm", "dimos-module-macros", + "dimos-tf-messages", "lcm-msgs", "nalgebra", "rayon", @@ -604,6 +615,7 @@ dependencies = [ "tracing", "tracing-subscriber", "tracing-test", + "transforms", "url", "validator", "zenoh", @@ -619,6 +631,17 @@ dependencies = [ "toml 1.1.6+spec-1.1.0", ] +[[package]] +name = "dimos-tf-messages" +version = "0.1.0" +dependencies = [ + "cdr", + "heapless", + "ros2msg", + "serde", + "serde-big-array", +] + [[package]] name = "dirs" version = "6.0.0" @@ -950,6 +973,15 @@ version = "0.33.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e762b9b188634fc508d4ec54b62ea3d352c43fd431d18d05d46f559f217f4725" +[[package]] +name = "hash32" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47d60b12902ba28e2730cd37e95b8c9223af2808df9e902d4df49588d1470606" +dependencies = [ + "byteorder", +] + [[package]] name = "hashbrown" version = "0.12.3" @@ -991,6 +1023,28 @@ name = "hashbrown" version = "0.17.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" +dependencies = [ + "allocator-api2", + "equivalent", + "foldhash 0.2.0", +] + +[[package]] +name = "heapless" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0bfb9eb618601c89945a70e254898da93b13be0388091d42117462b265bb3fad" +dependencies = [ + "hash32", + "serde", + "stable_deref_trait", +] + +[[package]] +name = "heck" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" [[package]] name = "hermit-abi" @@ -1385,7 +1439,7 @@ checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" [[package]] name = "lcm-msgs" version = "0.1.0" -source = "git+https://github.com/dimensionalOS/dimos-lcm.git?branch=rust-codegen#04d78e8622500244123ba9cefa4c51b4cb454549" +source = "git+https://github.com/dimensionalOS/dimos-lcm.git?rev=04d78e8622500244123ba9cefa4c51b4cb454549#04d78e8622500244123ba9cefa4c51b4cb454549" dependencies = [ "byteorder", ] @@ -1926,6 +1980,16 @@ dependencies = [ "zerocopy", ] +[[package]] +name = "prettyplease" +version = "0.2.37" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "479ca8adacdd7ce8f1fb39ce9ecccbfe93a3f1344b3d0d97f20bc0196208f62b" +dependencies = [ + "proc-macro2", + "syn 2.0.119", +] + [[package]] name = "proc-macro-crate" version = "3.5.0" @@ -2280,6 +2344,23 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "ros2msg" +version = "0.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2fe291be5cb6d4b8e6bf9c9de80c40bd1cb09ecac50deafa2fdcae0027e04988" +dependencies = [ + "heck", + "pest", + "pest_derive", + "prettyplease", + "proc-macro2", + "quote", + "regex", + "syn 2.0.119", + "thiserror", +] + [[package]] name = "rustc-hash" version = "2.1.3" @@ -2520,6 +2601,15 @@ dependencies = [ "serde_derive", ] +[[package]] +name = "serde-big-array" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "11fc7cc2c76d73e0f27ee52abbd64eec84d46f370c88371120433196934e4b7f" +dependencies = [ + "serde", +] + [[package]] name = "serde_core" version = "1.0.229" @@ -3216,6 +3306,18 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "transforms" +version = "2.2.0" +source = "git+https://github.com/TomCC7/transforms?rev=5ae9563de0c02638c04e393aadfad28444a4398e#5ae9563de0c02638c04e393aadfad28444a4398e" +dependencies = [ + "approx", + "hashbrown 0.17.1", + "libm", + "serde", + "thiserror", +] + [[package]] name = "twox-hash" version = "1.6.3" diff --git a/native/rust/dimos-module/Cargo.toml b/native/rust/dimos-module/Cargo.toml index b79f793a92c..57cb27bd429 100644 --- a/native/rust/dimos-module/Cargo.toml +++ b/native/rust/dimos-module/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "dimos-module" -version = "0.1.0" +version = "0.2.0" edition = "2021" description = "Rust native module SDK for dimos NativeModule framework" license = "Apache-2.0" @@ -8,9 +8,12 @@ license = "Apache-2.0" [workspace] [dependencies] -dimos-lcm = { git = "https://github.com/dimensionalOS/dimos-lcm.git", branch = "rust-codegen" } +dimos-tf-messages = { path = "../dimos-tf-messages" } +transforms = { version = "=2.2.0", git = "https://github.com/TomCC7/transforms", rev = "5ae9563de0c02638c04e393aadfad28444a4398e" } +cdr = { version = "=0.2.4", git = "https://github.com/TomCC7/cdr-rs", rev = "38a4368255d408faf374726653b39d4a4e9735cb" } +dimos-lcm = { git = "https://github.com/dimensionalOS/dimos-lcm.git", rev = "04d78e8622500244123ba9cefa4c51b4cb454549" } dimos-module-macros = { version = "=0.1.0", path = "../dimos-module-macros" } -lcm-msgs = { git = "https://github.com/dimensionalOS/dimos-lcm.git", branch = "rust-codegen" } +lcm-msgs = { git = "https://github.com/dimensionalOS/dimos-lcm.git", rev = "04d78e8622500244123ba9cefa4c51b4cb454549" } nalgebra = "0.35.0" rayon = "1" tokio = { version = "1", features = ["rt-multi-thread", "macros", "sync", "time", "signal", "io-std", "io-util"] } diff --git a/native/rust/dimos-module/README.md b/native/rust/dimos-module/README.md new file mode 100644 index 00000000000..77054406d08 --- /dev/null +++ b/native/rust/dimos-module/README.md @@ -0,0 +1,28 @@ +# Native module SDK + +`#[tf]` binds standard `tf2_msgs/TFMessage` streams named `tf` and `tf_static`. +Both ports must appear in the native module registry and Python wrapper. +The SDK uses generated `TransformStamped` and `Time` values directly. TF +fields accept no codec parameters; CDR is the default wire format. + +`lookup_transform(target, source, time)` maps source coordinates into the +target frame. ROS zero time selects the latest common time. `lookup().at()` +also takes a ROS `Time`; `within()` waits for missing data. Lookups interpolate +and reject extrapolation. The `transforms` library owns the frame tree, +history, composition and interpolation. Its current registry pins each child's +parent and static/dynamic kind; changing either requires rebuilding the buffer. + +`publish()` sends dynamic transforms and `publish_static()` sends static +transforms. Static edges never expire from a receiving buffer and do not +constrain the common time of a dynamic chain. Invalid batches do not change +the graph or enqueue a packet. A failed publication queue leaves the graph +unchanged. + +The SDK's LCM and Zenoh transports do not provide ROS DDS transient-local +history. A static publisher must remain available and reannounce its complete +mount set for late subscribers. Existing Python static-mount publishers still +reannounce on dynamic `tf` for compatibility with the Python query layer; they +have not been migrated to `tf_static` in this SDK change. + +Older modules pinned to SDK 0.1.0 retain their explicit legacy wire path. SDK +0.2.0 does not fall back to legacy TF decoding. diff --git a/native/rust/dimos-module/src/cdr.rs b/native/rust/dimos-module/src/cdr.rs new file mode 100644 index 00000000000..833ac0976bd --- /dev/null +++ b/native/rust/dimos-module/src/cdr.rs @@ -0,0 +1,123 @@ +// 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. + +//! SDK error and message-boundary adapter for the upstream complete CDR codec. +use ::cdr::{CdrBe, CdrLe, Infinite}; +use serde::{de::DeserializeOwned, Serialize}; +use std::io::{self, Cursor}; + +pub fn encode(message: &T) -> io::Result> { + encode_endian(message, true) +} + +pub fn encode_endian(message: &T, little_endian: bool) -> io::Result> { + if little_endian { + ::cdr::serialize::<_, _, CdrLe>(message, Infinite) + } else { + ::cdr::serialize::<_, _, CdrBe>(message, Infinite) + } + .map_err(|error| io::Error::new(io::ErrorKind::InvalidInput, error)) +} + +pub fn decode(bytes: &[u8]) -> io::Result { + // The SDK accepts standalone plain XCDR1, not parameter lists or other DDS representations. + if !matches!(bytes.get(..4), Some([0, 0 | 1, 0, 0])) { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "Expected plain CDR/XCDR1 encapsulation", + )); + } + let mut reader = Cursor::new(bytes); + let value = ::cdr::deserialize_from(&mut reader, ::cdr::Bounded(bytes.len() as u64)) + .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?; + if reader.position() != bytes.len() as u64 { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "Trailing bytes after CDR message", + )); + } + Ok(value) +} + +#[cfg(test)] +mod tests { + use super::*; + use serde::{Deserialize, Serialize}; + + #[derive(Debug, PartialEq, Serialize, Deserialize)] + struct Int32 { + data: i32, + } + + #[test] + fn literal_bytes_match_ros_cdr_in_both_byte_orders() { + let value = Int32 { data: 42 }; + assert_eq!(encode(&value).unwrap(), vec![0, 1, 0, 0, 42, 0, 0, 0]); + assert_eq!( + encode_endian(&value, false).unwrap(), + vec![0, 0, 0, 0, 0, 0, 0, 42] + ); + assert_eq!(decode::(&[0, 1, 0, 0, 42, 0, 0, 0]).unwrap(), value); + assert_eq!(decode::(&[0, 0, 0, 0, 0, 0, 0, 42]).unwrap(), value); + } + + #[test] + fn serialization_failure_is_an_error() { + struct Unencodable; + impl Serialize for Unencodable { + fn serialize(&self, _: S) -> Result { + Err(serde::ser::Error::custom("invalid message")) + } + } + assert_eq!( + encode(&Unencodable).unwrap_err().kind(), + io::ErrorKind::InvalidInput + ); + } + + #[test] + fn oversized_or_unterminated_strings_fail_at_the_library_boundary() { + for bytes in [ + vec![0, 1, 0, 0, 255, 255, 255, 255], + vec![0, 1, 0, 0, 0, 0, 0, 0], + vec![0, 1, 0, 0, 2, 0, 0, 0, b'a', b'b'], + ] { + assert_eq!( + decode::(&bytes).unwrap_err().kind(), + io::ErrorKind::InvalidData + ); + } + assert_eq!( + decode::(&encode(&"相机/café").unwrap()).unwrap(), + "相机/café" + ); + } + + #[test] + fn reject_wrong_encapsulation_truncation_and_trailing_bytes() { + for bytes in [ + vec![], + vec![1, 1, 0, 0, 42, 0, 0, 0], + vec![0, 1, 1, 0, 42, 0, 0, 0], + vec![0, 1, 0, 0, 42], + vec![0, 3, 0, 0, 42, 0, 0, 0], + vec![0, 1, 0, 0, 42, 0, 0, 0, 0], + ] { + assert_eq!( + decode::(&bytes).unwrap_err().kind(), + io::ErrorKind::InvalidData + ); + } + } +} diff --git a/native/rust/dimos-module/src/host.rs b/native/rust/dimos-module/src/host.rs index cc7982b7e55..fa01c48f320 100644 --- a/native/rust/dimos-module/src/host.rs +++ b/native/rust/dimos-module/src/host.rs @@ -891,7 +891,7 @@ mod tests { fn build(builder: &mut Builder, config: ProbeConfig) -> Self { Self { key: config.key, - ping: builder.output("ping", |b: &Vec| b.clone()), + ping: builder.output("ping", |b: &Vec| Ok(b.clone())), } } diff --git a/native/rust/dimos-module/src/lib.rs b/native/rust/dimos-module/src/lib.rs index 5816445c3f4..c2345519993 100644 --- a/native/rust/dimos-module/src/lib.rs +++ b/native/rust/dimos-module/src/lib.rs @@ -17,6 +17,7 @@ #[cfg(test)] extern crate self as dimos_module; +pub mod cdr; pub mod host; pub mod lcm; pub mod log; @@ -32,7 +33,7 @@ pub use dimos_module_macros::{native_config, Module}; pub use host::{host_main, HostSpec, ModuleEntry}; pub use lcm::LcmTransport; pub use module::{run, Builder, Input, Io, Module, ModuleConfig, NativeConfig, NoConfig, Output}; -pub use tf::{Lookup, Tf, Transform}; +pub use tf::{Lookup, Tf, TransformStamped}; pub use transport::{SharedTransport, Transport}; pub use workers::worker_pool; pub use zenoh::ZenohTransport; diff --git a/native/rust/dimos-module/src/module.rs b/native/rust/dimos-module/src/module.rs index 23fe965ba36..75b5aa44a94 100644 --- a/native/rust/dimos-module/src/module.rs +++ b/native/rust/dimos-module/src/module.rs @@ -123,13 +123,13 @@ impl Input { #[derive(Clone)] pub struct Output { pub topic: String, - encode: fn(&T) -> Vec, + encode: fn(&T) -> io::Result>, sender: mpsc::Sender>, } impl Output { pub async fn publish(&self, msg: &T) -> io::Result<()> { - publish_encoded(&self.sender, (self.encode)(msg)).await + publish_encoded(&self.sender, (self.encode)(msg)?).await } } @@ -140,7 +140,7 @@ impl Output { pub struct Io { pub topic: String, receiver: mpsc::Receiver, - encode: fn(&T) -> Vec, + encode: fn(&T) -> io::Result>, sender: mpsc::Sender>, } @@ -150,7 +150,7 @@ impl Io { } pub async fn publish(&self, msg: &T) -> io::Result<()> { - publish_encoded(&self.sender, (self.encode)(msg)).await + publish_encoded(&self.sender, (self.encode)(msg)?).await } } @@ -386,7 +386,7 @@ impl Builder { Input { topic, receiver } } - pub fn output(&mut self, port: &str, encode: fn(&T) -> Vec) -> Output { + pub fn output(&mut self, port: &str, encode: fn(&T) -> io::Result>) -> Output { let topic = self.topic_for(port); let sender = self.add_publisher(&topic); Output { @@ -401,7 +401,7 @@ impl Builder { &mut self, port: &str, decode: fn(&[u8]) -> io::Result, - encode: fn(&T) -> Vec, + encode: fn(&T) -> io::Result>, ) -> Io { let topic = self.topic_for(port); let receiver = self.add_route(&topic, decode); @@ -414,19 +414,22 @@ impl Builder { } } - /// A handle that answers transform queries and publishes on the `tf` topic. - /// - /// The graph fills in the background as `tf` messages arrive. Repeated calls - /// share one graph. + /// Standard ROS TF streams. Dynamic and static messages share one frame tree. pub fn tf(&mut self) -> crate::tf::Tf { if let Some(tf) = &self.tf { return tf.clone(); } let topic = self.topic_for("tf"); + let static_topic = self.topic_for("tf_static"); let sender = self.add_publisher(&topic); - let (tf, route) = - crate::tf::tf_subscription(topic.clone(), crate::tf::DEFAULT_TF_WINDOW_SECS, sender); + let static_sender = self.add_publisher(&static_topic); + let (tf, route, static_route) = + crate::tf::tf_subscription(sender, static_sender, crate::tf::DEFAULT_TF_WINDOW); self.routes.entry(topic).or_default().push(route); + self.routes + .entry(static_topic) + .or_default() + .push(static_route); self.tf = Some(tf.clone()); tf } @@ -942,7 +945,7 @@ mod tests { #[test] fn output_uses_mapped_topic() { let mut builder = builder_with_topics(&[("cmd_vel", "/robot/cmd_vel")]); - let output = builder.output("cmd_vel", |b: &Vec| b.clone()); + let output = builder.output("cmd_vel", |b: &Vec| Ok(b.clone())); assert_eq!(output.topic, "/robot/cmd_vel"); } @@ -950,7 +953,7 @@ mod tests { fn topics_matching_ports_exactly_pass() { let mut builder = builder_with_topics(&[("cmd", "/robot/cmd"), ("odom", "/robot/odom")]); builder.input("cmd", |b| Ok(b.to_vec())); - builder.output("odom", |b: &Vec| b.clone()); + builder.output("odom", |b: &Vec| Ok(b.clone())); builder.enforce_topics_match_ports().expect("exact match"); } @@ -958,7 +961,7 @@ mod tests { fn a_port_the_coordinator_never_sent_is_rejected() { let mut builder = builder_with_topics(&[("cmd", "/robot/cmd")]); builder.input("cmd", |b| Ok(b.to_vec())); - builder.output("odom", |b: &Vec| b.clone()); + builder.output("odom", |b: &Vec| Ok(b.clone())); let err = builder .enforce_topics_match_ports() .expect_err("odom has no topic"); @@ -977,7 +980,10 @@ mod tests { #[test] fn a_tf_field_claims_the_tf_topic() { - let mut builder = builder_with_topics(&[("tf", "/tf#tf2_msgs.TFMessage")]); + let mut builder = builder_with_topics(&[ + ("tf", "/tf#tf2_msgs.TFMessage"), + ("tf_static", "/tf_static#tf2_msgs.TFMessage"), + ]); builder.tf(); builder.enforce_topics_match_ports().expect("tf claimed"); } @@ -993,14 +999,14 @@ mod tests { #[test] fn io_uses_mapped_topic() { let mut builder = builder_with_topics(&[("cmd", "/robot/cmd")]); - let io = builder.io("cmd", |b| Ok(b.to_vec()), |b: &Vec| b.clone()); + let io = builder.io("cmd", |b| Ok(b.to_vec()), |b: &Vec| Ok(b.clone())); assert_eq!(io.topic, "/robot/cmd"); } #[test] fn io_registers_one_route_and_one_publisher_on_the_same_topic() { let mut builder = builder_with_topics(&[("cmd", "/robot/cmd")]); - let _io = builder.io("cmd", |b| Ok(b.to_vec()), |b: &Vec| b.clone()); + let _io = builder.io("cmd", |b| Ok(b.to_vec()), |b: &Vec| Ok(b.clone())); assert_eq!(builder.routes.get("/robot/cmd").map(Vec::len), Some(1)); assert_eq!(builder.outputs.len(), 1); assert_eq!(builder.outputs[0].0, "/robot/cmd"); @@ -1009,7 +1015,7 @@ mod tests { #[tokio::test] async fn io_receives_on_its_route_and_publishes_to_its_queue() { let mut builder = builder_with_topics(&[("cmd", "/robot/cmd")]); - let mut io = builder.io("cmd", |b| Ok(b.to_vec()), |b: &Vec| b.clone()); + let mut io = builder.io("cmd", |b| Ok(b.to_vec()), |b: &Vec| Ok(b.clone())); builder.routes["/robot/cmd"][0].try_dispatch(b"inbound"); assert_eq!(io.recv().await.expect("inbound message"), b"inbound"); @@ -1022,7 +1028,7 @@ mod tests { #[tokio::test] async fn io_publish_errors_when_the_publish_worker_is_gone() { let mut builder = builder_with_topics(&[]); - let io = builder.io("cmd", |b| Ok(b.to_vec()), |b: &Vec| b.clone()); + let io = builder.io("cmd", |b| Ok(b.to_vec()), |b: &Vec| Ok(b.clone())); builder.outputs.clear(); let err = io .publish(&b"x".to_vec()) @@ -1044,10 +1050,44 @@ mod tests { let mut builder = builder_with_topics(&[("tf", "/tf")]); builder.tf(); builder.tf(); - assert_eq!(builder.outputs.len(), 1); + assert_eq!(builder.outputs.len(), 2); assert_eq!(builder.routes.get("/tf").map(Vec::len), Some(1)); } + #[tokio::test] + async fn port_encode_error_does_not_enqueue_or_close_the_port() { + fn encode(value: &Vec) -> io::Result> { + if value == b"bad" { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "encoding failed", + )); + } + Ok(value.clone()) + } + let mut builder = builder_with_topics(&[("output", "/out"), ("io", "/io")]); + let output = builder.output("output", encode); + let io = builder.io("io", |bytes| Ok(bytes.to_vec()), encode); + assert_eq!( + output.publish(&b"bad".to_vec()).await.unwrap_err().kind(), + io::ErrorKind::InvalidInput + ); + assert_eq!( + io.publish(&b"bad".to_vec()).await.unwrap_err().kind(), + io::ErrorKind::InvalidInput + ); + for (_, receiver) in &mut builder.outputs { + assert!(matches!( + receiver.try_recv(), + Err(tokio::sync::mpsc::error::TryRecvError::Empty) + )); + } + output.publish(&b"out".to_vec()).await.unwrap(); + io.publish(&b"io".to_vec()).await.unwrap(); + assert_eq!(builder.outputs[0].1.recv().await.unwrap(), b"out"); + assert_eq!(builder.outputs[1].1.recv().await.unwrap(), b"io"); + } + // recv/publish concurrency #[tokio::test(flavor = "multi_thread", worker_threads = 2)] @@ -1065,7 +1105,7 @@ mod tests { let mut builder = Builder::new(topics(&[("data", "/data"), ("out", "/out")])); let _input = builder.input("data", |b| Ok(b.to_vec())); - let output = builder.output("out", |b: &Vec| b.clone()); + let output = builder.output("out", |b: &Vec| Ok(b.clone())); subscribe_routes(&transport, builder.routes).await.unwrap(); let transport = Arc::new(transport); @@ -1115,7 +1155,7 @@ mod tests { } Ok(b.to_vec()) }); - let output = builder.output("out", |b: &Vec| b.clone()); + let output = builder.output("out", |b: &Vec| Ok(b.clone())); subscribe_routes(&transport, builder.routes).await.unwrap(); let transport = Arc::new(transport); @@ -1166,8 +1206,8 @@ mod tests { }); let mut builder = Builder::new(topics(&[("block_out", "/block"), ("fast_out", "/fast")])); - let block_out = builder.output("block_out", |b: &Vec| b.clone()); - let fast_out = builder.output("fast_out", |b: &Vec| b.clone()); + let block_out = builder.output("block_out", |b: &Vec| Ok(b.clone())); + let fast_out = builder.output("fast_out", |b: &Vec| Ok(b.clone())); let _pub_tasks = spawn_publish_tasks(Arc::clone(&transport), builder.outputs); // Wedge the block channel, then publish on the fast channel. @@ -1278,8 +1318,8 @@ mod tests { Ok(Msg(bytes.to_vec())) } - fn encode(msg: &Msg) -> Vec { - msg.0.clone() + fn encode(msg: &Msg) -> io::Result> { + Ok(msg.0.clone()) } #[derive(crate::Module)] diff --git a/native/rust/dimos-module/src/pointcloud.rs b/native/rust/dimos-module/src/pointcloud.rs index 31b8ef0aee6..1e6b7ddba7a 100644 --- a/native/rust/dimos-module/src/pointcloud.rs +++ b/native/rust/dimos-module/src/pointcloud.rs @@ -58,45 +58,78 @@ pub fn extract_xyz(msg: &PointCloud2) -> Result, ExtractError> { if step == 0 { return Err(ExtractError("point_step is 0")); } - let n = count(msg.width, "negative width")? - .checked_mul(count(msg.height, "negative height")?) - .ok_or(ExtractError("width*height overflows"))?; - let needed = n + let width = count(msg.width, "negative width")?; + let height = count(msg.height, "negative height")?; + let row_step = width .checked_mul(step) - .ok_or(ExtractError("width*height*point_step overflows"))?; - if msg.data.len() < needed { - return Err(ExtractError( - "data buffer shorter than width*height*point_step", - )); - } - for off in [xo, yo, zo] { - if off.checked_add(4).is_none_or(|end| end > step) { + .ok_or(ExtractError("row size overflows"))?; + extract_xyz_rows( + &msg.data, + [width, height], + [step, row_step], + [xo, yo, zo], + false, + ) +} + +/// Read finite float32 xyz values from validated field offsets, respecting row padding. +/// Message adapters retain their own schema types and pass only the wire layout here. +pub fn extract_xyz_rows( + data: &[u8], + dimensions: [usize; 2], + steps: [usize; 2], + offsets: [usize; 3], + big_endian: bool, +) -> Result, ExtractError> { + let [width, height] = dimensions; + let [step, row_step] = steps; + if step == 0 { + return Err(ExtractError("point_step is 0")); + } + let packed = width + .checked_mul(step) + .ok_or(ExtractError("row size overflows"))?; + if row_step < packed { + return Err(ExtractError("row_step is shorter than a row")); + } + let needed = height + .checked_mul(row_step) + .ok_or(ExtractError("row buffer size overflows"))?; + if data.len() < needed { + return Err(ExtractError("data buffer shorter than rows")); + } + for offset in offsets { + if offset.checked_add(4).is_none_or(|end| end > step) { return Err(ExtractError( "xyz field offsets do not fit within point_step", )); } } - - let mut out = Vec::with_capacity(n); - for base in (0..needed).step_by(step) { - let x = read_f32_le(&msg.data, base + xo); - let y = read_f32_le(&msg.data, base + yo); - let z = read_f32_le(&msg.data, base + zo); - if x.is_finite() && y.is_finite() && z.is_finite() { - out.push([x, y, z]); + let mut out = Vec::new(); + if width == 0 || height == 0 { + return Ok(out); + } + for row in 0..height { + for column in 0..width { + let base = row * row_step + column * step; + let point = offsets.map(|offset| { + let bytes: [u8; 4] = data[base + offset..base + offset + 4] + .try_into() + .expect("bounds checked above"); + if big_endian { + f32::from_be_bytes(bytes) + } else { + f32::from_le_bytes(bytes) + } + }); + if point.iter().all(|value| value.is_finite()) { + out.push(point); + } } } Ok(out) } -#[inline] -fn read_f32_le(buf: &[u8], off: usize) -> f32 { - let bytes: [u8; 4] = buf[off..off + 4] - .try_into() - .expect("bounds checked by caller"); - f32::from_le_bytes(bytes) -} - #[cfg(test)] mod tests { use super::*; @@ -129,6 +162,40 @@ mod tests { } } + #[test] + fn padded_rows_and_both_byte_orders_preserve_finite_points() { + for big_endian in [false, true] { + let mut bytes = Vec::new(); + for xyz in [[1.0f32, 2.0, 3.0], [4.0, 5.0, 6.0]] { + for value in xyz { + bytes.extend(if big_endian { + value.to_be_bytes() + } else { + value.to_le_bytes() + }); + } + bytes.extend([255u8; 4]); + } + assert_eq!( + extract_xyz_rows(&bytes, [1, 2], [12, 16], [0, 4, 8], big_endian).unwrap(), + vec![[1., 2., 3.], [4., 5., 6.]] + ); + } + } + + #[test] + fn malformed_row_layouts_are_rejected_before_allocation_or_indexing() { + for (dimensions, steps, offsets) in [ + ([1, 1], [12, 11], [0, 4, 8]), + ([1, 2], [12, 16], [0, 4, 8]), + ([1, 1], [12, 12], [0, 4, usize::MAX]), + ([usize::MAX, 1], [12, 12], [0, 4, 8]), + ([1, usize::MAX], [12, 16], [0, 4, 8]), + ] { + assert!(extract_xyz_rows(&[0; 16], dimensions, steps, offsets, false).is_err()); + } + } + #[test] fn drops_non_finite_points_and_moves_nothing() { let cloud = cloud_of(&[[1.0, 2.0, 0.3], [f32::NAN, 0.0, 0.0], [0.0, 0.0, 0.0]]); diff --git a/native/rust/dimos-module/src/tf.rs b/native/rust/dimos-module/src/tf.rs index a6c1611f9e3..c5a81cd1d37 100644 --- a/native/rust/dimos-module/src/tf.rs +++ b/native/rust/dimos-module/src/tf.rs @@ -12,1194 +12,613 @@ // See the License for the specific language governing permissions and // limitations under the License. -//! Transform client for native modules. -//! -//! Each `/tf` edge is buffered per `(parent, child)`, and [`Tf::lookup`] composes -//! the shortest path through the frame graph. Lookups are nearest-in-time within -//! a tolerance, not interpolated. [`Tf::publish`] sends transforms onto the same -//! topic and feeds the local graph. - -use std::collections::{HashMap, HashSet, VecDeque}; -use std::io; -use std::sync::atomic::AtomicU64; -use std::sync::{Arc, Mutex, RwLock}; -use std::time::Duration; - -use nalgebra::{Isometry3, Quaternion, Translation3, UnitQuaternion, Vector3}; -use tokio::sync::{mpsc, Notify}; -use tracing::warn; +//! ROS TF messages backed by the `transforms` frame tree. use crate::module::Route; -use crate::time::now_secs; - -/// How many seconds of history each edge keeps. -pub(crate) const DEFAULT_TF_WINDOW_SECS: f64 = 10.0; - -const WARN_INTERVAL: Duration = Duration::from_secs(1); - -/// A rigid transform from `parent` to `child` at a point in time. -/// -/// It maps a point expressed in `child` coordinates into `parent` coordinates. -#[derive(Clone, Debug)] -pub struct Transform { - pub parent: String, - pub child: String, - pub ts: f64, - iso: Isometry3, +use nalgebra::{Isometry3, Quaternion, Translation3, UnitQuaternion}; +use std::{ + io, + sync::{Arc, RwLock}, + time::{Duration, SystemTime, UNIX_EPOCH}, +}; +use tokio::sync::{mpsc, Notify}; +use transforms::{ + geometry::{ + Quaternion as FrameQuaternion, Transform as FrameTransform, Vector3 as FrameVector, + }, + time::Stamp, + Registry, +}; + +pub use dimos_tf_messages::{ + builtin_interfaces::msg::time::Time, geometry_msgs::msg::transform_stamped::TransformStamped, + tf2_msgs::msg::tf_message::TFMessage, +}; +use dimos_tf_messages::{ + geometry_msgs::msg::{ + quaternion::Quaternion as WireQuaternion, transform::Transform, vector3::Vector3, + }, + std_msgs::msg::header::Header, +}; + +pub(crate) const DEFAULT_TF_WINDOW: Duration = Duration::from_secs(10); + +fn invalid(error: impl std::fmt::Display) -> io::Error { + io::Error::new(io::ErrorKind::InvalidData, error.to_string()) } -impl Transform { - pub fn new( - parent: impl Into, - child: impl Into, - ts: f64, - iso: Isometry3, - ) -> Self { - Self { - parent: parent.into(), - child: child.into(), - ts, - iso, - } - } - - pub fn translation(&self) -> Vector3 { - self.iso.translation.vector - } - - pub fn rotation(&self) -> UnitQuaternion { - self.iso.rotation - } - - fn inverse(&self) -> Transform { - Transform { - parent: self.child.clone(), - child: self.parent.clone(), - ts: self.ts, - iso: self.iso.inverse(), - } - } - - fn compose(&self, other: &Transform) -> Transform { - Transform { - parent: self.parent.clone(), - child: other.child.clone(), - ts: self.ts, - iso: self.iso * other.iso, - } +pub fn nanoseconds(time: &Time) -> io::Result { + if time.nanosec >= 1_000_000_000 { + return Err(invalid("unnormalized ROS time")); } + Ok(i64::from(time.sec) * 1_000_000_000 + i64::from(time.nanosec)) } -struct Sample { - ts: f64, - iso: Isometry3, +pub fn time_from_nanoseconds(ns: i64) -> io::Result