Python: Handle ros2 uint8[] field as str without ValueError

python/mcap-ros2-support/mcap_ros2/_dynamic.py

@@ -2,6 +2,8 @@
 
 import os
 import re
+import base64
+import binascii
 from io import BytesIO
 from types import SimpleNamespace
 from typing import Any, Callable, Dict, List, Optional, Tuple, Union
@@ -422,6 +424,12 @@ def _write_complex_type(
                 array: Union[List[Any], Tuple[Any], Any] = _get_property(
                     ros2_msg, field.name
                 )
+                if ftype.type == "uint8" and isinstance(array, str):
+                    # Special handling for bytes in JSON
+                    try:
+                        array = base64.b64decode(array)
+                    except binascii.Error:
+                        pass
                 if array is None:
                     array = []
                 if (

python/mcap-ros2-support/tests/test_ros2_writer.py

@@ -1,5 +1,8 @@
 from io import BytesIO
 
+import json
+import base64
+
 from mcap_ros2.decoder import DecoderFactory
 from mcap_ros2.writer import Writer as Ros2Writer
 
@@ -59,3 +62,34 @@ def test_write_std_msgs_empty_messages():
         assert msg.message.log_time == index
         assert msg.message.publish_time == index
         assert msg.message.sequence == index
+
+
+def test_write_bytes_object_from_json_deserialization():
+    output = BytesIO()
+    ros_writer = Ros2Writer(output=output)
+    schema = ros_writer.register_msgdef("test_msgs/BytesMsg", "uint8[] data")
+
+    sample_bytes = bytes([0x77, 0x67, 0x65, 0x80])
+    to_serialize_dict = {"data": base64.b64encode(sample_bytes).decode("utf-8")}
+    sample_json = json.dumps(to_serialize_dict)
+    deserialized_dict = json.loads(sample_json)
+
+    for i in range(0, 10):
+        ros_writer.write_message(
+            topic="/test",
+            schema=schema,
+            message=deserialized_dict,
+            log_time=i,
+            publish_time=i,
+            sequence=i,
+        )
+    ros_writer.finish()
+
+    output.seek(0)
+    for index, msg in enumerate(read_ros2_messages(output)):
+        assert msg.channel.topic == "/test"
+        assert msg.schema.name == "test_msgs/BytesMsg"
+        assert msg.decoded_message.data == sample_bytes
+        assert msg.message.log_time == index
+        assert msg.message.publish_time == index
+        assert msg.message.sequence == index