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

update skills: recording demo, object detection

Browse source
This commit is contained in:
shalenikol 2024-10-07 17:23:19 +03:00 • committed by Bill Finger
parent b42c2e633e
commit ec604bccb8
2 changed files with 59 additions and 41 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -3,7 +3,7 @@
recording_demo_lifecycle_node_via_RosBag
ROS 2 program for recording a demo via RosBag
@shalenikol release 0.1
@shalenikol release 0.2
"""
import os
import shutil
@ -46,20 +46,24 @@ class RecordingDemo(Node):
str_cfg = self.get_parameter(PARAM_SKILL_CFG).get_parameter_value().string_value
self.skill_cfg = json.loads(str_cfg)
sets = self.skill_cfg["Settings"]
sets = self.skill_cfg["Settings"]["output"]["params"]
for s in sets:
if s["name"] == "output_path":
self.output_path = s["value"]
# check for output folder existence
if os.path.isdir(self.output_path):
shutil.rmtree(self.output_path)
x = 1
while os.path.isdir(self.output_path + str(x)):
x += 1
shutil.move(self.output_path, self.output_path + str(x))
self.writer = rosbag2_py.SequentialWriter()
storage_options = rosbag2_py._storage.StorageOptions(
uri=self.output_path,
storage_id=WRITER_ID)
converter_options = rosbag2_py._storage.ConverterOptions("", "")
self.writer.open(storage_options, converter_options)
# storage_options = rosbag2_py._storage.StorageOptions(
# uri=self.output_path,
# storage_id=WRITER_ID)
# converter_options = rosbag2_py._storage.ConverterOptions("", "")
# self.writer.open(storage_options, converter_options)
def get_list_topics(self) -> list:
topic_names_and_types = get_topic_names_and_types(node=self, include_hidden_topics=False)
@ -82,6 +86,9 @@ class RecordingDemo(Node):
TransitionCallbackReturn.FAILURE transitions to "unconfigured".
TransitionCallbackReturn.ERROR or any uncaught exceptions to "errorprocessing"
"""
storage_options = rosbag2_py._storage.StorageOptions(uri=self.output_path, storage_id=WRITER_ID)
converter_options = rosbag2_py._storage.ConverterOptions("", "")
self.writer.open(storage_options, converter_options)
self.topics = self.get_list_topics()
for id, topic in enumerate(self.topics):
@ -125,6 +132,8 @@ class RecordingDemo(Node):
self.destroy_subscription(sub)
self._sub.clear()
self.writer.close()
self.get_logger().info('on_cleanup() is called.')
return TransitionCallbackReturn.SUCCESS