robossembler/runtime
7
0
Fork 0
Code Issues 4 Pull requests Projects Releases Packages Wiki Activity Actions

add settings: compressed and serialization format

Browse source
This commit is contained in:
shalenikol 2025-05-22 14:27:22 +03:00
parent 9c6cdfa291
commit 125fb14cb7
2 changed files with 33 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

8
rbs_utils/rbs_utils/config/recording_demo_skill.json
View file

@ -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" }
]
}
}

40
rbs_utils/rbs_utils/scripts/recording_demo_via_rosbag.py
View file

@ -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