re_ros_msg

Author: oxkitsune

ARCHITECTURE.md

@@ -232,6 +232,7 @@ Update instructions:
 | re_mcap            | Convert MCAP into Rerun-compatible data.                                             |
 | re_memory          | Run-time memory tracking and profiling.                                              |
 | re_perf_telemetry  | In and out of process performance profiling utilities for Rerun & Redap              |
+| re_ros_msg         | Parsing and deserializing ROS   messages                                             |
 | re_smart_channel   | A channel that keeps track of latency and queue length.                              |
 | re_span            | An integer range that always has a non-negative length                               |
 | re_string_interner | Yet another string interning library                                                 |

Cargo.lock

@@ -8141,6 +8141,7 @@ dependencies = [
  "re_format_arrow",
  "re_log",
  "re_log_types",
+ "re_ros_msg",
  "re_tracing",
  "re_types",
  "serde",
@@ -8451,6 +8452,23 @@ dependencies = [
  "winit",
 ]
 
+[[package]]
+name = "re_ros_msg"
+version = "0.26.0-alpha.1+dev"
+dependencies = [
+ "ahash",
+ "anyhow",
+ "byteorder",
+ "cdr-encoding",
+ "insta",
+ "re_format_arrow",
+ "re_log",
+ "re_tracing",
+ "serde",
+ "serde_bytes",
+ "thiserror 1.0.69",
+]
+
 [[package]]
 name = "re_sdk"
 version = "0.26.0-alpha.1+dev"

Cargo.toml

@@ -104,6 +104,7 @@ re_log = { path = "crates/utils/re_log", version = "=0.26.0-alpha.1", default-fe
 re_mcap = { path = "crates/utils/re_mcap", version = "=0.26.0-alpha.1", default-features = false }
 re_memory = { path = "crates/utils/re_memory", version = "=0.26.0-alpha.1", default-features = false }
 re_perf_telemetry = { path = "crates/utils/re_perf_telemetry", version = "=0.26.0-alpha.1", default-features = false }
+re_ros_msg = { path = "crates/utils/re_ros_msg", version = "=0.26.0-alpha.1", default-features = false }
 re_smart_channel = { path = "crates/utils/re_smart_channel", version = "=0.26.0-alpha.1", default-features = false }
 re_span = { path = "crates/utils/re_span", version = "=0.26.0-alpha.1", default-features = false }
 re_string_interner = { path = "crates/utils/re_string_interner", version = "=0.26.0-alpha.1", default-features = false }

crates/utils/re_mcap/Cargo.toml

@@ -18,6 +18,7 @@ workspace = true
 re_chunk.workspace = true
 re_log.workspace = true
 re_log_types.workspace = true
+re_ros_msg.workspace = true
 re_tracing.workspace = true
 re_types = { workspace = true, features = ["ecolor", "glam", "image", "video"] }
 

crates/utils/re_mcap/src/layers/ros2_reflection.rs

@@ -1,3 +1,4 @@
+use anyhow::Context as _;
 use arrow::{
     array::{
         ArrayBuilder, ArrowPrimitiveType, BinaryBuilder, BooleanBuilder, FixedSizeListBuilder,
@@ -10,18 +11,43 @@ use arrow::{
         Int64Type, UInt8Type, UInt16Type, UInt32Type, UInt64Type,
     },
 };
+use cdr_encoding::CdrDeserializer;
 use re_chunk::{Chunk, ChunkId};
-use re_types::{ComponentDescriptor, reflection::ComponentDescriptorExt as _};
-
-use crate::parsers::ros2msg::reflection::{
+use re_ros_msg::{
     MessageSchema,
-    deserialize::primitive_array::PrimitiveArray,
-    deserialize::{Value, decode_bytes},
+    deserialize::{MapResolver, MessageSeed, Value, primitive_array::PrimitiveArray},
     message_spec::{ArraySize, BuiltInType, ComplexType, MessageSpecification, Type},
 };
-use crate::parsers::{MessageParser, ParserContext};
+use re_types::{ComponentDescriptor, reflection::ComponentDescriptorExt as _};
+use serde::de::DeserializeSeed as _;
+
+use crate::parsers::{MessageParser, ParserContext, dds};
 use crate::{Error, LayerIdentifier, MessageLayer};
 
+pub fn decode_bytes(top: &MessageSchema, buf: &[u8]) -> anyhow::Result<Value> {
+    // 4-byte encapsulation header
+    if buf.len() < 4 {
+        anyhow::bail!("short encapsulation");
+    }
+
+    let representation_identifier = dds::RepresentationIdentifier::from_bytes([buf[0], buf[1]])
+        .with_context(|| "failed to parse CDR representation identifier")?;
+
+    let resolver = MapResolver::new(top.dependencies.iter().map(|dep| (dep.name.clone(), dep)));
+
+    let seed = MessageSeed::new(&top.spec, &resolver);
+
+    if representation_identifier.is_big_endian() {
+        let mut de = CdrDeserializer::<byteorder::BigEndian>::new(&buf[4..]);
+        seed.deserialize(&mut de)
+            .with_context(|| "failed to deserialize CDR message")
+    } else {
+        let mut de = CdrDeserializer::<byteorder::LittleEndian>::new(&buf[4..]);
+        seed.deserialize(&mut de)
+            .with_context(|| "failed to deserialize CDR message")
+    }
+}
+
 struct Ros2ReflectionMessageParser {
     num_rows: usize,
     message_schema: MessageSchema,
@@ -260,7 +286,7 @@ fn append_value(
 
             struct_builder.append(true);
         }
-        Value::Array(vec) | Value::Seq(vec) => {
+        Value::Array(vec) | Value::Sequence(vec) => {
             let list_builder = downcast_err::<ListBuilder<Box<dyn ArrayBuilder>>>(builder)?;
 
             for val in vec {

crates/utils/re_mcap/src/parsers/ros2msg/mod.rs

@@ -7,8 +7,6 @@ pub mod scalar_parser;
 pub mod sensor_msgs;
 pub mod std_msgs;
 
-pub mod reflection;
-
 /// Trait for ROS2 message parsers that can be constructed with just a row count.
 pub trait Ros2MessageParser: MessageParser {
     /// Create a new parser instance.

crates/utils/re_ros_msg/Cargo.toml

@@ -0,0 +1,32 @@
+[package]
+name = "re_ros_msg"
+description = "Parse and decode ROS messages"
+authors.workspace = true
+edition.workspace = true
+homepage.workspace = true
+include.workspace = true
+license.workspace = true
+publish = true
+repository.workspace = true
+rust-version.workspace = true
+version.workspace = true
+
+[lints]
+workspace = true
+
+[dependencies]
+re_log.workspace = true
+re_tracing.workspace = true
+
+ahash.workspace = true
+anyhow.workspace = true
+byteorder.workspace = true
+cdr-encoding.workspace = true
+serde.workspace = true
+serde_bytes.workspace = true
+thiserror.workspace = true
+
+[dev-dependencies]
+re_format_arrow.workspace = true
+
+insta = { workspace = true, features = ["filters", "redactions"] }