Merge branch '123-api_rec_demo' into 'nix-jazzy'

add settings: compressed and serialization format

See merge request robossembler/robossembler-ros2!79

rbs_utils/rbs_utils/config/recording_demo_skill.json

@@ -7,9 +7,7 @@
       "Launch": { "package": "rbs_utils", "executable": "recording_demo_via_rosbag.py" },
       "BTAction": [
         {
-          "name": "rdConfigure",
-          "type_action": "ACTION",
-          "type": "run",
+          "name": "rdConfigure", "type_action": "ACTION", "type": "run",
           "param": []
         },
         { "name": "rdStop", "type_action": "ACTION", "type": "stop", 
@@ -28,7 +26,9 @@
       "Settings": {
         "output": {
           "params": [
-            { "name": "output_path", "value": "rbs_demo", "description": "!!! Путь, куда запишутся эпизоды в формате rosbag" }
+            { "name": "output_path", "value": "rbs_bag", "description": "!!! Путь, куда запишется rosbag" },
+            { "name": "compressed", "value": "True", "default": "False" },
+            { "name": "serialization", "value": "mcap", "default": "cdr" }
           ]
         }
       }

rbs_utils/rbs_utils/scripts/recording_demo_via_rosbag.py

@@ -20,7 +20,7 @@ import rosbag2_py # 0.26.7 (jazzy)
 from ros2topic.api import get_topic_names_and_types
 from rclpy.serialization import serialize_message
 
-from sensor_msgs.msg import Image, JointState
+from sensor_msgs.msg import JointState, Image, CompressedImage
 
 from rbs_utils.recording_demo import CommandType
 
@@ -30,8 +30,10 @@ PARAM_SKILL_CFG   = "lc_record_demo_cfg"
 OUTPUT_PATH_DEFAULT = "rbs_bag"
 RECORD_PATH = "episode"
 WRITER_ID = "sqlite3"
-TOPIC_TYPES = ["sensor_msgs/msg/JointState", "sensor_msgs/msg/Image"]
-TOPIC_CLASSES = [JointState, Image]
+TOPIC_FILTERS = {
+    "sensor_msgs/msg/JointState": JointState,
+    "sensor_msgs/msg/Image": Image,
+}
 
 class RecordingDemo(Node):
     """ lifecycle node """
@@ -41,6 +43,8 @@ class RecordingDemo(Node):
         self.topics = []
         self._sub = []
         self.output_path = OUTPUT_PATH_DEFAULT
+        self.compressed = "False"  # default: not compressed image
+        self.serialization = "cdr" # default serialization format
 
         # for other nodes
         kwargs["allow_undeclared_parameters"] = True
@@ -51,17 +55,31 @@ class RecordingDemo(Node):
         self.skill_cfg = json.loads(str_cfg)
         sets = self.skill_cfg["Settings"]["output"]["params"]
         for s in sets:
-            if s["name"] == "output_path":
-                self.output_path = s["value"]
+            nam = s["name"]
+            val = s["value"]
+            if nam == "output_path":
+                self.output_path = val
+            elif nam == "compressed":
+                self.compressed = val
+            elif nam == "serialization":
+                self.serialization = val
+
+        if self.compressed == "True":
+            del TOPIC_FILTERS["sensor_msgs/msg/Image"]
+            TOPIC_FILTERS["sensor_msgs/msg/CompressedImage"] = CompressedImage
         
     def get_list_topics(self) -> list:
         topic_names_and_types = get_topic_names_and_types(node=self, include_hidden_topics=False)
 
         topic_list = []
-        for (topic_name, topic_types) in topic_names_and_types:
-            if topic_types[0] in TOPIC_TYPES:
-                i = TOPIC_TYPES.index(topic_types[0])
-                topic_list.append({"name": topic_name, "type": topic_types[0], "class": TOPIC_CLASSES[i]})
+        for topic_name, topic_types in topic_names_and_types:
+            topic_type = topic_types[0]
+            if topic_type in TOPIC_FILTERS:
+                topic_list.append({
+                    "name": topic_name,
+                    "type": topic_type,
+                    "class": TOPIC_FILTERS[topic_type],
+                })
 
         return topic_list
 
@@ -96,9 +114,9 @@ class RecordingDemo(Node):
                 id,
                 name=topic["name"],
                 type=topic["type"],
-                serialization_format="cdr")
+                serialization_format=self.serialization)
             self.writer.create_topic(topic_info)
-            self.get_logger().info(f"Topics: {id}) {topic}")
+            self.get_logger().info(f"Topics: {id+1}) {topic}")
 
         self.get_logger().info("on_configure() is called.")
         return TransitionCallbackReturn.SUCCESS