diff --git a/.gitignore b/.gitignore index 26bc611653..7ce28ef6df 100644 --- a/.gitignore +++ b/.gitignore @@ -5,3 +5,4 @@ cmake-build-release/ build/ venv/ **/.pytest_cache/ +**/__pycache__/** diff --git a/README.md b/README.md index f86ba25b6f..1fa5a3c0f6 100644 --- a/README.md +++ b/README.md @@ -174,6 +174,14 @@ Topic information: Topic: /chatter | Type: std_msgs/String | Count: 9 | Serializ Topic: /my_chatter | Type: std_msgs/String | Count: 18 | Serialization Format: cdr ``` +### Reindexing Data + +In the event that a bag file becomes unreadable, due to either a corrupted or missing `metadata.yaml` file within the bag, you can attempt to recover the bag by reindexing: +``` +$ ros2 bag reindex +``` +This will attempt to read the internal storage of the bag file to reconstruct a workable `metadata.yaml` file. It is not guaranteed to be the exact same as a file generated during the recording process. You may have to check the file manually to make sure all the details are correct. + ### Overriding QoS Profiles When starting a recording or playback workflow, you can pass a YAML file that contains QoS profile settings for a specific topic. diff --git a/ros2bag/ros2bag/verb/reindex.py b/ros2bag/ros2bag/verb/reindex.py new file mode 100644 index 0000000000..3ecdfc77b0 --- /dev/null +++ b/ros2bag/ros2bag/verb/reindex.py @@ -0,0 +1,58 @@ +# Copyright 2018 Open Source Robotics Foundation, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import os + +from ros2bag.api import check_path_exists +from ros2bag.verb import VerbExtension + + +class ReindexVerb(VerbExtension): + """Generate metadata from a bag.""" + + def add_arguments(self, parser, cli_name): # noqa: D102 + parser.add_argument( + 'bag_file', type=check_path_exists, help='Bag file to reindex') + parser.add_argument( + '-s', '--storage', default='sqlite3', + help="storage identifier to be used, defaults to 'sqlite3'") + parser.add_argument( + '-f', '--serialization-format', default='', + help='rmw serialization format in which the messages are saved, defaults to the' + ' rmw currently in use') + parser.add_argument( + '--compression-format', type=str, default='', choices=['zstd'], + help='Specify the compression format/algorithm. Default is none.' + ) + self._subparser = parser + + def main(self, *, args): # noqa: D102 + uri = args.bag_file + + # NOTE(hidmic): in merged install workspaces on Windows, Python entrypoint lookups + # combined with constrained environments (as imposed by colcon test) + # may result in DLL loading failures when attempting to import a C + # extension. Therefore, do not import rosbag2_transport at the module + # level but on demand, right before first use. + from rosbag2_transport import rosbag2_transport_py + + rosbag2_transport_py.reindex( + uri=uri, + storage_id=args.storage, + serialization_format=args.serialization_format, + compression_format=args.compression_format + ) + + if os.path.isdir(uri) and not os.listdir(uri): + os.rmdir(uri) diff --git a/ros2bag/setup.py b/ros2bag/setup.py index f6efbec71f..a45e914287 100644 --- a/ros2bag/setup.py +++ b/ros2bag/setup.py @@ -42,6 +42,7 @@ 'list = ros2bag.verb.list:ListVerb', 'play = ros2bag.verb.play:PlayVerb', 'record = ros2bag.verb.record:RecordVerb', + 'reindex = ros2bag.verb.reindex:ReindexVerb' ], } ) diff --git a/ros2bag/test/resources/reindex_test_bags/file_compression/file_compression_0.db3.zstd b/ros2bag/test/resources/reindex_test_bags/file_compression/file_compression_0.db3.zstd new file mode 100644 index 0000000000..47efd7e928 Binary files /dev/null and b/ros2bag/test/resources/reindex_test_bags/file_compression/file_compression_0.db3.zstd differ diff --git a/ros2bag/test/resources/reindex_test_bags/file_compression/file_compression_1.db3.zstd b/ros2bag/test/resources/reindex_test_bags/file_compression/file_compression_1.db3.zstd new file mode 100644 index 0000000000..46a9604a27 Binary files /dev/null and b/ros2bag/test/resources/reindex_test_bags/file_compression/file_compression_1.db3.zstd differ diff --git a/ros2bag/test/resources/reindex_test_bags/file_compression/file_compression_2.db3.zstd b/ros2bag/test/resources/reindex_test_bags/file_compression/file_compression_2.db3.zstd new file mode 100644 index 0000000000..f901849587 Binary files /dev/null and b/ros2bag/test/resources/reindex_test_bags/file_compression/file_compression_2.db3.zstd differ diff --git a/ros2bag/test/resources/reindex_test_bags/file_compression/file_compression_3.db3.zstd b/ros2bag/test/resources/reindex_test_bags/file_compression/file_compression_3.db3.zstd new file mode 100644 index 0000000000..53c962965d Binary files /dev/null and b/ros2bag/test/resources/reindex_test_bags/file_compression/file_compression_3.db3.zstd differ diff --git a/ros2bag/test/resources/reindex_test_bags/file_compression/file_compression_target.yaml b/ros2bag/test/resources/reindex_test_bags/file_compression/file_compression_target.yaml new file mode 100644 index 0000000000..764e22a4c1 --- /dev/null +++ b/ros2bag/test/resources/reindex_test_bags/file_compression/file_compression_target.yaml @@ -0,0 +1,34 @@ +rosbag2_bagfile_information: + version: 4 + storage_identifier: sqlite3 + relative_file_paths: + - file_compression/file_compression_0.db3.zstd + - file_compression/file_compression_1.db3.zstd + - file_compression/file_compression_2.db3.zstd + - file_compression/file_compression_3.db3.zstd + duration: + nanoseconds: 59997866716 + starting_time: + nanoseconds_since_epoch: 1604357089919704830 + message_count: 775 + topics_with_message_count: + - topic_metadata: + name: /topic + type: std_msgs/msg/String + serialization_format: cdr + offered_qos_profiles: "- history: 3\n depth: 0\n reliability: 1\n durability: 2\n deadline:\n sec: 2147483647\n nsec: 4294967295\n lifespan:\n sec: 2147483647\n nsec: 4294967295\n liveliness: 1\n liveliness_lease_duration:\n sec: 2147483647\n nsec: 4294967295\n avoid_ros_namespace_conventions: false" + message_count: 383 + - topic_metadata: + name: /parameter_events + type: rcl_interfaces/msg/ParameterEvent + serialization_format: cdr + offered_qos_profiles: "- history: 3\n depth: 0\n reliability: 1\n durability: 2\n deadline:\n sec: 2147483647\n nsec: 4294967295\n lifespan:\n sec: 2147483647\n nsec: 4294967295\n liveliness: 1\n liveliness_lease_duration:\n sec: 2147483647\n nsec: 4294967295\n avoid_ros_namespace_conventions: false" + message_count: 0 + - topic_metadata: + name: /rosout + type: rcl_interfaces/msg/Log + serialization_format: cdr + offered_qos_profiles: "- history: 3\n depth: 0\n reliability: 1\n durability: 1\n deadline:\n sec: 2147483647\n nsec: 4294967295\n lifespan:\n sec: 10\n nsec: 0\n liveliness: 1\n liveliness_lease_duration:\n sec: 2147483647\n nsec: 4294967295\n avoid_ros_namespace_conventions: false" + message_count: 392 + compression_format: zstd + compression_mode: FILE \ No newline at end of file diff --git a/ros2bag/test/resources/reindex_test_bags/message_compression/message_compression_0.db3 b/ros2bag/test/resources/reindex_test_bags/message_compression/message_compression_0.db3 new file mode 100644 index 0000000000..cb496d68a3 Binary files /dev/null and b/ros2bag/test/resources/reindex_test_bags/message_compression/message_compression_0.db3 differ diff --git a/ros2bag/test/resources/reindex_test_bags/message_compression/message_compression_1.db3 b/ros2bag/test/resources/reindex_test_bags/message_compression/message_compression_1.db3 new file mode 100644 index 0000000000..fe3aba22e7 Binary files /dev/null and b/ros2bag/test/resources/reindex_test_bags/message_compression/message_compression_1.db3 differ diff --git a/ros2bag/test/resources/reindex_test_bags/message_compression/message_compression_2.db3 b/ros2bag/test/resources/reindex_test_bags/message_compression/message_compression_2.db3 new file mode 100644 index 0000000000..11a87eb30c Binary files /dev/null and b/ros2bag/test/resources/reindex_test_bags/message_compression/message_compression_2.db3 differ diff --git a/ros2bag/test/resources/reindex_test_bags/message_compression/message_compression_3.db3 b/ros2bag/test/resources/reindex_test_bags/message_compression/message_compression_3.db3 new file mode 100644 index 0000000000..9b189dfa13 Binary files /dev/null and b/ros2bag/test/resources/reindex_test_bags/message_compression/message_compression_3.db3 differ diff --git a/ros2bag/test/resources/reindex_test_bags/message_compression/message_compression_target.yaml b/ros2bag/test/resources/reindex_test_bags/message_compression/message_compression_target.yaml new file mode 100644 index 0000000000..147265e2c7 --- /dev/null +++ b/ros2bag/test/resources/reindex_test_bags/message_compression/message_compression_target.yaml @@ -0,0 +1,34 @@ +rosbag2_bagfile_information: + version: 4 + storage_identifier: sqlite3 + relative_file_paths: + - message_compression/message_compression_0.db3 + - message_compression/message_compression_1.db3 + - message_compression/message_compression_2.db3 + - message_compression/message_compression_3.db3 + duration: + nanoseconds: 59872587605 + starting_time: + nanoseconds_since_epoch: 1604581538133709023 + message_count: 1127 + topics_with_message_count: + - topic_metadata: + name: /parameter_events + type: rcl_interfaces/msg/ParameterEvent + serialization_format: cdr + offered_qos_profiles: "- history: 3\n depth: 0\n reliability: 1\n durability: 2\n deadline:\n sec: 2147483647\n nsec: 4294967295\n lifespan:\n sec: 2147483647\n nsec: 4294967295\n liveliness: 1\n liveliness_lease_duration:\n sec: 2147483647\n nsec: 4294967295\n avoid_ros_namespace_conventions: false\n- history: 3\n depth: 0\n reliability: 1\n durability: 2\n deadline:\n sec: 2147483647\n nsec: 4294967295\n lifespan:\n sec: 2147483647\n nsec: 4294967295\n liveliness: 1\n liveliness_lease_duration:\n sec: 2147483647\n nsec: 4294967295\n avoid_ros_namespace_conventions: false" + message_count: 0 + - topic_metadata: + name: /rosout + type: rcl_interfaces/msg/Log + serialization_format: cdr + offered_qos_profiles: "- history: 3\n depth: 0\n reliability: 1\n durability: 1\n deadline:\n sec: 2147483647\n nsec: 4294967295\n lifespan:\n sec: 10\n nsec: 0\n liveliness: 1\n liveliness_lease_duration:\n sec: 2147483647\n nsec: 4294967295\n avoid_ros_namespace_conventions: false\n- history: 3\n depth: 0\n reliability: 1\n durability: 1\n deadline:\n sec: 2147483647\n nsec: 4294967295\n lifespan:\n sec: 10\n nsec: 0\n liveliness: 1\n liveliness_lease_duration:\n sec: 2147483647\n nsec: 4294967295\n avoid_ros_namespace_conventions: false" + message_count: 758 + - topic_metadata: + name: /topic + type: std_msgs/msg/String + serialization_format: cdr + offered_qos_profiles: "" + message_count: 369 + compression_format: zstd + compression_mode: MESSAGE \ No newline at end of file diff --git a/ros2bag/test/resources/reindex_test_bags/multiple_files/multiple_files_0.db3 b/ros2bag/test/resources/reindex_test_bags/multiple_files/multiple_files_0.db3 new file mode 100644 index 0000000000..ffcf41d90b Binary files /dev/null and b/ros2bag/test/resources/reindex_test_bags/multiple_files/multiple_files_0.db3 differ diff --git a/ros2bag/test/resources/reindex_test_bags/multiple_files/multiple_files_0.db3-shm b/ros2bag/test/resources/reindex_test_bags/multiple_files/multiple_files_0.db3-shm new file mode 100644 index 0000000000..fe9ac2845e Binary files /dev/null and b/ros2bag/test/resources/reindex_test_bags/multiple_files/multiple_files_0.db3-shm differ diff --git a/ros2bag/test/resources/reindex_test_bags/multiple_files/multiple_files_0.db3-wal b/ros2bag/test/resources/reindex_test_bags/multiple_files/multiple_files_0.db3-wal new file mode 100644 index 0000000000..e69de29bb2 diff --git a/ros2bag/test/resources/reindex_test_bags/multiple_files/multiple_files_1.db3 b/ros2bag/test/resources/reindex_test_bags/multiple_files/multiple_files_1.db3 new file mode 100644 index 0000000000..d742510b96 Binary files /dev/null and b/ros2bag/test/resources/reindex_test_bags/multiple_files/multiple_files_1.db3 differ diff --git a/ros2bag/test/resources/reindex_test_bags/multiple_files/multiple_files_1.db3-shm b/ros2bag/test/resources/reindex_test_bags/multiple_files/multiple_files_1.db3-shm new file mode 100644 index 0000000000..fe9ac2845e Binary files /dev/null and b/ros2bag/test/resources/reindex_test_bags/multiple_files/multiple_files_1.db3-shm differ diff --git a/ros2bag/test/resources/reindex_test_bags/multiple_files/multiple_files_1.db3-wal b/ros2bag/test/resources/reindex_test_bags/multiple_files/multiple_files_1.db3-wal new file mode 100644 index 0000000000..e69de29bb2 diff --git a/ros2bag/test/resources/reindex_test_bags/multiple_files/multiple_files_2.db3 b/ros2bag/test/resources/reindex_test_bags/multiple_files/multiple_files_2.db3 new file mode 100644 index 0000000000..ced88a9cb7 Binary files /dev/null and b/ros2bag/test/resources/reindex_test_bags/multiple_files/multiple_files_2.db3 differ diff --git a/ros2bag/test/resources/reindex_test_bags/multiple_files/multiple_files_2.db3-shm b/ros2bag/test/resources/reindex_test_bags/multiple_files/multiple_files_2.db3-shm new file mode 100644 index 0000000000..fe9ac2845e Binary files /dev/null and b/ros2bag/test/resources/reindex_test_bags/multiple_files/multiple_files_2.db3-shm differ diff --git a/ros2bag/test/resources/reindex_test_bags/multiple_files/multiple_files_2.db3-wal b/ros2bag/test/resources/reindex_test_bags/multiple_files/multiple_files_2.db3-wal new file mode 100644 index 0000000000..e69de29bb2 diff --git a/ros2bag/test/resources/reindex_test_bags/multiple_files/multiple_files_3.db3 b/ros2bag/test/resources/reindex_test_bags/multiple_files/multiple_files_3.db3 new file mode 100644 index 0000000000..6137336d10 Binary files /dev/null and b/ros2bag/test/resources/reindex_test_bags/multiple_files/multiple_files_3.db3 differ diff --git a/ros2bag/test/resources/reindex_test_bags/multiple_files/multiple_files_3.db3-shm b/ros2bag/test/resources/reindex_test_bags/multiple_files/multiple_files_3.db3-shm new file mode 100644 index 0000000000..fe9ac2845e Binary files /dev/null and b/ros2bag/test/resources/reindex_test_bags/multiple_files/multiple_files_3.db3-shm differ diff --git a/ros2bag/test/resources/reindex_test_bags/multiple_files/multiple_files_3.db3-wal b/ros2bag/test/resources/reindex_test_bags/multiple_files/multiple_files_3.db3-wal new file mode 100644 index 0000000000..e69de29bb2 diff --git a/ros2bag/test/resources/reindex_test_bags/multiple_files/multiple_files_target.yaml b/ros2bag/test/resources/reindex_test_bags/multiple_files/multiple_files_target.yaml new file mode 100644 index 0000000000..a251989f7d --- /dev/null +++ b/ros2bag/test/resources/reindex_test_bags/multiple_files/multiple_files_target.yaml @@ -0,0 +1,34 @@ +rosbag2_bagfile_information: + version: 4 + storage_identifier: sqlite3 + relative_file_paths: + - multiple_files_0.db3 + - multiple_files_1.db3 + - multiple_files_2.db3 + - multiple_files_3.db3 + duration: + nanoseconds: 59997863418 + starting_time: + nanoseconds_since_epoch: 1604356847420232242 + message_count: 766 + topics_with_message_count: + - topic_metadata: + name: /topic + type: std_msgs/msg/String + serialization_format: cdr + offered_qos_profiles: "- history: 3\n depth: 0\n reliability: 1\n durability: 2\n deadline:\n sec: 2147483647\n nsec: 4294967295\n lifespan:\n sec: 2147483647\n nsec: 4294967295\n liveliness: 1\n liveliness_lease_duration:\n sec: 2147483647\n nsec: 4294967295\n avoid_ros_namespace_conventions: false" + message_count: 378 + - topic_metadata: + name: /parameter_events + type: rcl_interfaces/msg/ParameterEvent + serialization_format: cdr + offered_qos_profiles: "- history: 3\n depth: 0\n reliability: 1\n durability: 2\n deadline:\n sec: 2147483647\n nsec: 4294967295\n lifespan:\n sec: 2147483647\n nsec: 4294967295\n liveliness: 1\n liveliness_lease_duration:\n sec: 2147483647\n nsec: 4294967295\n avoid_ros_namespace_conventions: false" + message_count: 0 + - topic_metadata: + name: /rosout + type: rcl_interfaces/msg/Log + serialization_format: cdr + offered_qos_profiles: "- history: 3\n depth: 0\n reliability: 1\n durability: 1\n deadline:\n sec: 2147483647\n nsec: 4294967295\n lifespan:\n sec: 10\n nsec: 0\n liveliness: 1\n liveliness_lease_duration:\n sec: 2147483647\n nsec: 4294967295\n avoid_ros_namespace_conventions: false" + message_count: 388 + compression_format: "" + compression_mode: "" \ No newline at end of file diff --git a/ros2bag/test/test_reindex.py b/ros2bag/test/test_reindex.py new file mode 100644 index 0000000000..6e07053b06 --- /dev/null +++ b/ros2bag/test/test_reindex.py @@ -0,0 +1,322 @@ +# Copyright 2020 DCS Corporation, All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +# DISTRIBUTION A. Approved for public release; distribution unlimited. +# OPSEC #4584. +# +# Delivered to the U.S. Government with Unlimited Rights, as defined in DFARS +# Part 252.227-7013 or 7014 (Feb 2014). +# +# This notice must appear in all copies of this file and its derivatives. + +import contextlib +from pathlib import Path +from typing import Any +import unittest + +from launch import LaunchDescription +from launch.actions import ExecuteProcess +import launch_testing +import launch_testing.actions +import launch_testing.asserts +import launch_testing.markers +import launch_testing.tools + +import pytest +import yaml + + +RESOURCES_PATH = Path(__file__).parent / 'resources' / 'reindex_test_bags' +# TEST_NODE = 'ros2bag_record_qos_profile_test_node' +# TEST_NAMESPACE = 'ros2bag_record_qos_profile' +TEST_NODE = 'ros2bag_reindex_node' +TEST_NAMESPACE = 'ros2bag_reindex' +ERROR_STRING = r'\[ERROR] \[ros2bag]:' + + +@pytest.mark.rostest +@launch_testing.markers.keep_alive +def generate_test_description(): + return LaunchDescription([launch_testing.actions.ReadyToTest()]) + + +def check_version(target_base_node: Any, test_base_node: Any) -> None: + target_version = target_base_node.get('version') + test_version = test_base_node.get('version') + assert target_base_node.get('version') == test_base_node.get('version'), \ + print('Reindex generated wrong version. Expected "{}"; got "{}"'.format( + target_version, test_version + )) + + +def check_storage_identifier(target_base_node: Any, test_base_node: Any) -> None: + target_si = target_base_node.get('storage_identifier') + test_si = test_base_node.get('storage_identifier') + assert target_si == test_si, \ + print('Reindex generated wrong storage identifier. ' + 'Expected "{}"; got "{}"'.format( + target_si, test_si + )) + + +def check_relative_filepaths(target_base_node: Any, test_base_node: Any) -> None: + target_fp = target_base_node.get('relative_file_paths') + test_fp = test_base_node.get('relative_file_paths') + assert target_fp == test_fp, \ + print('Reindex generated wrong relative file paths. ' + 'Expected "{}"; got "{}"'.format( + target_fp, test_fp + )) + + +def check_duration(target_base_node: Any, test_base_node: Any) -> None: + target_duration = target_base_node.get('duration') + test_duration = test_base_node.get('duration') + assert target_duration == test_duration, \ + print('Reindex generated wrong duration. ' + 'Expected "{}"; got "{}"'.format( + target_duration, test_duration + )) + + +def check_starting_time(target_base_node: Any, test_base_node: Any) -> None: + target_start = target_base_node.get('starting_time') + test_start = test_base_node.get('starting_time') + assert target_start == test_start, \ + print('Reindex generated wrong starting time. ' + 'Expected "{}"; got "{}"'.format( + target_start, test_start + )) + + +def check_message_count(target_base_node: Any, test_base_node: Any) -> None: + target_mc = target_base_node.get('message_count') + test_mc = test_base_node.get('message_count') + assert target_mc == test_mc, \ + print('Reindex generated wrong message count. ' + 'Expected "{}"; got "{}'.format( + target_mc, test_mc + )) + + +def check_topic_type(target_topic_metadata: Any, test_topic_metadata: Any) -> None: + target_name = target_topic_metadata.get('topic_metadata').get('name') + test_type = test_topic_metadata.get('topic_metadata').get('type') + target_type = target_topic_metadata.get('topic_metadata').get('type') + assert test_type == target_type, \ + print('Reindex generated incorrect target type for "{}". ' + 'Expected "{}", got "{}'.format(target_name, target_type, test_type)) + + +def check_topic_ser_fmt(target_topic_metadata: Any, test_topic_metadata: Any) -> None: + target_name = target_topic_metadata.get('topic_metadata').get('name') + test_ser_fmt = test_topic_metadata.get('topic_metadata').get('serialization_format') + target_ser_fmt = target_topic_metadata.get('topic_metadata').get('serialization_format') + assert test_ser_fmt == target_ser_fmt, \ + print('Reindex generated incorrect serialization format for "{}". ' + 'Expected "{}", got "{}"'.format(target_name, target_ser_fmt, test_ser_fmt)) + + +def check_topic_qos(target_topic_metadata: Any, test_topic_metadata: Any) -> None: + target_name = target_topic_metadata.get('topic_metadata').get('name') + test_qos = test_topic_metadata.get('topic_metadata').get('offered_qos_profiles') + target_qos = test_topic_metadata.get('topic_metadata').get('offered_qos_profiles') + assert test_qos == target_qos, \ + print('Reindex generated incorrect QOS profiles for "{}". ' + 'Expected "{}", got "{}"'.format(target_name, target_qos, test_qos)) + + +def check_topic_message_count(target_topic_metadata: Any, test_topic_metadata: Any) -> None: + target_name = target_topic_metadata.get('topic_metadata').get('name') + test_count = test_topic_metadata.get('message_count') + target_count = target_topic_metadata.get('message_count') + assert test_count == target_count, \ + print('Reindex generated incorrect message count for "{}". ' + 'Expected "{}", got "{}"'.format(target_name, target_count, test_count)) + + +def check_topics(target_base_node: Any, test_base_node: Any) -> None: + target_topics = target_base_node.get('topics_with_message_count') + test_topics = test_base_node.get('topics_with_message_count') + assert test_topics, print('Reindex generated no topics block') + + if test_topics: + for test_topic in test_topics: + test_name = test_topic.get('topic_metadata').get('name') + target_topic = next((x for x in target_topics + if x.get('topic_metadata').get('name') == test_name)) + + assert target_topic, print('Reindex generated extra topic: "{}"'.format(test_name)) + if target_topic: + check_topic_type(target_topic, test_topic) + check_topic_ser_fmt(target_topic, test_topic) + check_topic_qos(target_topic, test_topic) + check_topic_message_count(target_topic, test_topic) + + +def check_compression_fmt(target_base_node: Any, test_base_node: Any) -> None: + target_c_fmt = target_base_node.get('compression_format') + test_c_fmt = test_base_node.get('compression_format') + assert target_c_fmt == test_c_fmt, \ + print('Reindex generated incorrect compression format. ' + 'Expected "{}", got "{}"'.format(target_c_fmt, test_c_fmt)) + + +def check_compression_mode(target_base_node: Any, test_base_node: Any) -> None: + target_mode = target_base_node.get('compression_mode') + test_mode = test_base_node.get('compression_mode') + assert target_mode == test_mode, \ + print('Reindex generated incorrect compression mode. ' + 'Expected "{}", got "{}"'.format(target_mode, test_mode)) + + +def compare_metadata_files(target_file: Path, test_file: Path): + target_yaml = yaml.safe_load(target_file.open()) + generated_yaml = yaml.safe_load(test_file.open()) + + # Check that base node exists + target_base_node = target_yaml.get('rosbag2_bagfile_information') + test_base_node = generated_yaml.get('rosbag2_bagfile_information') + assert test_base_node, print('Reindex was unable to generate base node') + + check_version(target_base_node, test_base_node) + check_storage_identifier(target_base_node, test_base_node) + + # INCONSISTENT BETWEEN COMPRESSED / NON COMPRESSED BAGS + # Disabling for now + # check_relative_filepaths(target_base_node, test_base_node) + + # MAY NOT BE ABLE TO GUARANTEE THIS # + # check_duration(target_base_node, test_base_node) + # check_starting_time(target_base_node, test_base_node) + + check_message_count(target_base_node, test_base_node) + check_topics(target_base_node, test_base_node) + check_compression_fmt(target_base_node, test_base_node) + check_compression_mode(target_base_node, test_base_node) + + +class TestRos2BagReindexMultiFile(unittest.TestCase): + + @classmethod + def setUpClass(cls, launch_service, proc_info, proc_output): + @contextlib.contextmanager + def launch_bag_command(self, arguments, **kwargs): + pkg_command_action = ExecuteProcess( + cmd=['ros2', 'bag', *arguments], + additional_env={'PYTHONUNBUFFERED': '1'}, + name='ros2bag-cli', + output='screen', + **kwargs + ) + with launch_testing.tools.launch_process( + launch_service, pkg_command_action, proc_info, proc_output + ) as pkg_command: + yield pkg_command + cls.launch_bag_command = launch_bag_command + + @classmethod + def tearDown(cls) -> None: + metadata_file = RESOURCES_PATH / 'multiple_files' / 'metadata.yaml' + metadata_file.unlink(True) + + def test_multiple_files(self): + bag_path = RESOURCES_PATH / 'multiple_files' + metadata_file = bag_path / 'metadata.yaml' + target_file = bag_path / 'multiple_files_target.yaml' + + arguments = ['reindex', bag_path.as_posix()] + with self.launch_bag_command(arguments=arguments) as bag_command: + bag_command.wait_for_shutdown(timeout=5) + + # Metadata.yaml file should be created at this point + compare_metadata_files(target_file, metadata_file) + + +# class TestRos2BagReindexMessageCompression(unittest.TestCase): + +# @classmethod +# def setUpClass(cls, launch_service, proc_info, proc_output): +# @contextlib.contextmanager +# def launch_bag_command(self, arguments, **kwargs): +# pkg_command_action = ExecuteProcess( +# cmd=['ros2', 'bag', *arguments], +# additional_env={'PYTHONUNBUFFERED': '1'}, +# name='ros2bag-cli', +# output='screen', +# **kwargs +# ) +# with launch_testing.tools.launch_process( +# launch_service, pkg_command_action, proc_info, proc_output +# ) as pkg_command: +# yield pkg_command +# cls.launch_bag_command = launch_bag_command + +# @classmethod +# def tearDown(cls) -> None: +# metadata_file = RESOURCES_PATH / 'message_compression' / 'metadata.yaml' +# metadata_file.unlink(True) + +# def test_compressed(self): +# bag_path = RESOURCES_PATH / 'message_compression' +# metadata_file = bag_path / 'metadata.yaml' +# target_file = bag_path / 'message_compression_target.yaml' + +# arguments = ['reindex', bag_path.as_posix(), +# '--compression-format', 'zstd', +# '--compression-mode', 'message'] +# with self.launch_bag_command(arguments=arguments) as bag_command: +# bag_command.wait_for_shutdown(timeout=5) + +# # Metadata.yaml file should be created at this point +# compare_metadata_files(target_file, metadata_file) + + +# class TestRos2BagReindexFileCompression(unittest.TestCase): + +# @classmethod +# def setUpClass(cls, launch_service, proc_info, proc_output): +# @contextlib.contextmanager +# def launch_bag_command(self, arguments, **kwargs): +# pkg_command_action = ExecuteProcess( +# cmd=['ros2', 'bag', *arguments], +# additional_env={'PYTHONUNBUFFERED': '1'}, +# name='ros2bag-cli', +# output='screen', +# **kwargs +# ) +# with launch_testing.tools.launch_process( +# launch_service, pkg_command_action, proc_info, proc_output +# ) as pkg_command: +# yield pkg_command +# cls.launch_bag_command = launch_bag_command + +# @classmethod +# def tearDown(cls) -> None: +# metadata_file = RESOURCES_PATH / 'file_compression' / 'metadata.yaml' +# metadata_file.unlink(True) + +# def test_compressed(self): +# bag_path = RESOURCES_PATH / 'file_compression' +# metadata_file = bag_path / 'metadata.yaml' +# target_file = bag_path / 'file_compression_target.yaml' + +# arguments = ['reindex', bag_path.as_posix(), +# '--compression-format', 'zstd', +# '--compression-mode', 'file'] +# with self.launch_bag_command(arguments=arguments) as bag_command: +# bag_command.wait_for_shutdown(timeout=5) + +# # Metadata.yaml file should be created at this point +# compare_metadata_files(target_file, metadata_file) diff --git a/rosbag2_cpp/CMakeLists.txt b/rosbag2_cpp/CMakeLists.txt index 1d78d5831f..3788e1a65d 100644 --- a/rosbag2_cpp/CMakeLists.txt +++ b/rosbag2_cpp/CMakeLists.txt @@ -57,9 +57,12 @@ add_library(${PROJECT_NAME} SHARED src/rosbag2_cpp/typesupport_helpers.cpp src/rosbag2_cpp/types/introspection_message.cpp src/rosbag2_cpp/writer.cpp - src/rosbag2_cpp/writers/sequential_writer.cpp) + src/rosbag2_cpp/writers/sequential_writer.cpp + src/rosbag2_cpp/reindexer.cpp + src/rosbag2_cpp/reindexers/sequential_reindexer.cpp) ament_target_dependencies(${PROJECT_NAME} + PUBLIC ament_index_cpp pluginlib rcpputils diff --git a/rosbag2_cpp/include/rosbag2_cpp/reindexer.hpp b/rosbag2_cpp/include/rosbag2_cpp/reindexer.hpp new file mode 100644 index 0000000000..29c9534443 --- /dev/null +++ b/rosbag2_cpp/include/rosbag2_cpp/reindexer.hpp @@ -0,0 +1,80 @@ +// Copyright 2020 DCS Corporation, All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// +// DISTRIBUTION A. Approved for public release; distribution unlimited. +// OPSEC #4584. +// +// Delivered to the U.S. Government with Unlimited Rights, as defined in DFARS +// Part 252.227-7013 or 7014 (Feb 2014). +// +// This notice must appear in all copies of this file and its derivatives. + +#ifndef ROSBAG2_CPP__REINDEXER_HPP_ +#define ROSBAG2_CPP__REINDEXER_HPP_ + +#include + +#include "rosbag2_cpp/converter_options.hpp" +#include "rosbag2_cpp/visibility_control.hpp" + +#include "rosbag2_storage/bag_metadata.hpp" +#include "rosbag2_storage/serialized_bag_message.hpp" +#include "rosbag2_storage/storage_filter.hpp" +#include "rosbag2_storage/storage_options.hpp" +#include "rosbag2_storage/topic_metadata.hpp" + +// This is necessary because of using stl types here. It is completely safe, because +// a) the member is not accessible from the outside +// b) there are no inline functions. +#ifdef _WIN32 +# pragma warning(push) +# pragma warning(disable:4251) +#endif + +namespace rosbag2_cpp +{ + +namespace reindexer_interfaces +{ +class BaseReindexerInterface; +} // namespace reindexer_interfaces + +/** + * The reindexer attempts to construct a metadata file based on information contained in the bag. + */ +class ROSBAG2_CPP_PUBLIC Reindexer final +{ +public: + explicit Reindexer(std::unique_ptr reindexer_impl); + + ~Reindexer(); + + /** + * Attempts to create a metadata file from the specified bag + * + * \throws runtime_error if the Reader is not open. + */ + void reindex(const rosbag2_storage::StorageOptions & storage_options); + +private: + std::unique_ptr reindexer_impl; +}; + +} // namespace rosbag2_cpp + +#ifdef _WIN32 +# pragma warning(pop) +#endif + +#endif // ROSBAG2_CPP__REINDEXER_HPP_ diff --git a/rosbag2_cpp/include/rosbag2_cpp/reindexer_interfaces/base_reindexer_interface.hpp b/rosbag2_cpp/include/rosbag2_cpp/reindexer_interfaces/base_reindexer_interface.hpp new file mode 100644 index 0000000000..b54074e83e --- /dev/null +++ b/rosbag2_cpp/include/rosbag2_cpp/reindexer_interfaces/base_reindexer_interface.hpp @@ -0,0 +1,54 @@ +// Copyright 2020 DCS Corporation, All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// +// DISTRIBUTION A. Approved for public release; distribution unlimited. +// OPSEC #4584. +// +// Delivered to the U.S. Government with Unlimited Rights, as defined in DFARS +// Part 252.227-7013 or 7014 (Feb 2014). +// +// This notice must appear in all copies of this file and its derivatives. + +#ifndef ROSBAG2_CPP__REINDEXER_INTERFACES__BASE_REINDEXER_INTERFACE_HPP_ +#define ROSBAG2_CPP__REINDEXER_INTERFACES__BASE_REINDEXER_INTERFACE_HPP_ + +#include +#include + +#include "rosbag2_cpp/converter_options.hpp" +#include "rosbag2_cpp/visibility_control.hpp" + +#include "rosbag2_storage/bag_metadata.hpp" +#include "rosbag2_storage/serialized_bag_message.hpp" +#include "rosbag2_storage/storage_filter.hpp" +#include "rosbag2_storage/storage_options.hpp" +#include "rosbag2_storage/topic_metadata.hpp" + +namespace rosbag2_cpp +{ +namespace reindexer_interfaces +{ + +class ROSBAG2_CPP_PUBLIC BaseReindexerInterface +{ +public: + virtual ~BaseReindexerInterface() {} + + virtual void reindex(const rosbag2_storage::StorageOptions & storage_options) = 0; +}; + +} // namespace reindexer_interfaces +} // namespace rosbag2_cpp + +#endif // ROSBAG2_CPP__REINDEXER_INTERFACES__BASE_REINDEXER_INTERFACE_HPP_ diff --git a/rosbag2_cpp/include/rosbag2_cpp/reindexers/sequential_reindexer.hpp b/rosbag2_cpp/include/rosbag2_cpp/reindexers/sequential_reindexer.hpp new file mode 100644 index 0000000000..f07e8ad173 --- /dev/null +++ b/rosbag2_cpp/include/rosbag2_cpp/reindexers/sequential_reindexer.hpp @@ -0,0 +1,124 @@ +// Copyright 2020 DCS Corporation, All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// +// DISTRIBUTION A. Approved for public release; distribution unlimited. +// OPSEC #4584. +// +// Delivered to the U.S. Government with Unlimited Rights, as defined in DFARS +// Part 252.227-7013 or 7014 (Feb 2014). +// +// This notice must appear in all copies of this file and its derivatives. + +#ifndef ROSBAG2_CPP__REINDEXERS__SEQUENTIAL_REINDEXER_HPP_ +#define ROSBAG2_CPP__REINDEXERS__SEQUENTIAL_REINDEXER_HPP_ + +#include +#include +#include +#include + +#include "rcpputils/filesystem_helper.hpp" + +#include "rosbag2_cpp/converter.hpp" +#include "rosbag2_cpp/reindexer_interfaces/base_reindexer_interface.hpp" +#include "rosbag2_cpp/serialization_format_converter_factory.hpp" +#include "rosbag2_cpp/serialization_format_converter_factory_interface.hpp" +#include "rosbag2_cpp/visibility_control.hpp" + +#include "rosbag2_storage/metadata_io.hpp" +#include "rosbag2_storage/storage_factory.hpp" +#include "rosbag2_storage/storage_factory_interface.hpp" +#include "rosbag2_storage/storage_options.hpp" +#include "rosbag2_storage/storage_filter.hpp" +#include "rosbag2_storage/storage_interfaces/read_only_interface.hpp" + +// This is necessary because of using stl types here. It is completely safe, because +// a) the member is not accessible from the outside +// b) there are no inline functions. +#ifdef _WIN32 +# pragma warning(push) +# pragma warning(disable:4251) +#endif + +namespace rosbag2_cpp +{ +namespace reindexers +{ + +class ROSBAG2_CPP_PUBLIC SequentialReindexer + : public ::rosbag2_cpp::reindexer_interfaces::BaseReindexerInterface +{ +public: + SequentialReindexer( + std::unique_ptr storage_factory = + std::make_unique(), + std::shared_ptr converter_factory = + std::make_shared(), + std::unique_ptr metadata_io = + std::make_unique()); + + virtual ~SequentialReindexer(); + + + void reindex(const rosbag2_storage::StorageOptions & storage_options) override; + + void fill_topics_metadata(); + + void reset(); + + void finalize_metadata(); + +protected: + std::unique_ptr storage_factory_{}; + std::shared_ptr storage_{}; + std::unique_ptr converter_{}; + std::unique_ptr metadata_io_{}; + rosbag2_storage::BagMetadata metadata_{}; + std::vector topics_metadata_{}; + std::vector file_paths_{}; // List of database files. + // Index of file to read from + std::vector::iterator current_file_iterator_{}; + +private: + rcpputils::fs::path base_folder_; + std::shared_ptr converter_factory_{}; + + std::vector get_database_files(const rcpputils::fs::path & base_folder); + + void open(const rosbag2_storage::StorageOptions & storage_options); + + // Prepares the metadata by setting initial values. + void init_metadata( + const std::vector & files, + const rosbag2_storage::StorageOptions & storage_options); + + // Attempts to harvest metadata from all bag files, and aggregates the result + void aggregate_metadata( + const std::vector & files, + const rosbag2_storage::StorageOptions & storage_options); + + // Compairson function for std::sort with our filepath convention + static bool comp_rel_file( + const rcpputils::fs::path & first_path, + const rcpputils::fs::path & second_path); +}; + +} // namespace reindexers +} // namespace rosbag2_cpp + +#ifdef _WIN32 +# pragma warning(pop) +#endif + +#endif // ROSBAG2_CPP__REINDEXERS__SEQUENTIAL_REINDEXER_HPP_ diff --git a/rosbag2_cpp/package.xml b/rosbag2_cpp/package.xml index 5c633fa9e8..7b1b5d3779 100644 --- a/rosbag2_cpp/package.xml +++ b/rosbag2_cpp/package.xml @@ -11,6 +11,7 @@ ament_cmake ament_index_cpp + libboost-filesystem pluginlib rcutils rcpputils diff --git a/rosbag2_cpp/src/rosbag2_cpp/reindexer.cpp b/rosbag2_cpp/src/rosbag2_cpp/reindexer.cpp new file mode 100644 index 0000000000..d5cd7fc23c --- /dev/null +++ b/rosbag2_cpp/src/rosbag2_cpp/reindexer.cpp @@ -0,0 +1,49 @@ +// Copyright 2020 DCS Corporation, All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// +// DISTRIBUTION A. Approved for public release; distribution unlimited. +// OPSEC #4584. +// +// Delivered to the U.S. Government with Unlimited Rights, as defined in DFARS +// Part 252.227-7013 or 7014 (Feb 2014). +// +// This notice must appear in all copies of this file and its derivatives. + +#include "rosbag2_cpp/reindexer.hpp" + +#include +#include +#include + +#include "rosbag2_cpp/info.hpp" +#include "rosbag2_cpp/reindexer_interfaces/base_reindexer_interface.hpp" + +#include "rosbag2_storage/storage_options.hpp" + +namespace rosbag2_cpp +{ + +Reindexer::Reindexer(std::unique_ptr reindexer_impl) +: reindexer_impl(std::move(reindexer_impl)) +{} + +Reindexer::~Reindexer() {} + +void Reindexer::reindex(const rosbag2_storage::StorageOptions & storage_options) +{ + reindexer_impl->reindex(storage_options); +} + + +} // namespace rosbag2_cpp diff --git a/rosbag2_cpp/src/rosbag2_cpp/reindexers/sequential_reindexer.cpp b/rosbag2_cpp/src/rosbag2_cpp/reindexers/sequential_reindexer.cpp new file mode 100644 index 0000000000..9c02cb6ed5 --- /dev/null +++ b/rosbag2_cpp/src/rosbag2_cpp/reindexers/sequential_reindexer.cpp @@ -0,0 +1,346 @@ +// Copyright 2020 DCS Corporation, All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// +// DISTRIBUTION A. Approved for public release; distribution unlimited. +// OPSEC #4584. +// +// Delivered to the U.S. Government with Unlimited Rights, as defined in DFARS +// Part 252.227-7013 or 7014 (Feb 2014). +// +// This notice must appear in all copies of this file and its derivatives. + +#include +#include +#include +#include +#include +#include +#include +#include + +#include "rcpputils/asserts.hpp" +#include "rcpputils/filesystem_helper.hpp" + +#include "rosbag2_cpp/logging.hpp" +#include "rosbag2_cpp/reindexers/sequential_reindexer.hpp" + +#include "rosbag2_storage/storage_options.hpp" + +#ifdef WIN32 +// Import windows filesystem functionality +#include +#include +#include +#else +// We're on a UNIX system. Import their filesystem stuff instead +#include +#endif + + +namespace rosbag2_cpp +{ +namespace reindexers +{ +namespace details +{ +std::vector resolve_relative_paths( + const rcpputils::fs::path & base_folder, + std::vector relative_files, + const int version = 4) +{ + auto base_path = rcpputils::fs::path(base_folder); // Preserve folder + if (version < 4) { + // In older rosbags (version <=3) relative files are prefixed with the rosbag folder name + base_path = rcpputils::fs::path(base_folder).parent_path(); + } + + rcpputils::require_true( + base_path.exists(), "base folder does not exist: " + base_folder.string()); + rcpputils::require_true( + base_path.is_directory(), "base folder has to be a directory: " + base_folder.string()); + + for (auto & file : relative_files) { + auto path = rcpputils::fs::path(file); + if (path.is_absolute()) { + continue; + } + file = (base_path / path).string(); + } + + return relative_files; +} +} // namespace details + +std::string strip_parent_path(const rcpputils::fs::path & relative_path) +{ + return relative_path.filename().string(); +} + +SequentialReindexer::SequentialReindexer( + std::unique_ptr storage_factory, + std::shared_ptr converter_factory, + std::unique_ptr metadata_io) +: storage_factory_(std::move(storage_factory)), + converter_(nullptr), + metadata_io_(std::move(metadata_io)), + converter_factory_(std::move(converter_factory)) +{} + +SequentialReindexer::~SequentialReindexer() +{ + reset(); +} + +void SequentialReindexer::reset() +{ + if (storage_) { + storage_.reset(); + } +} + +bool SequentialReindexer::comp_rel_file( + const rcpputils::fs::path & first_path, const rcpputils::fs::path & second_path) +{ + std::regex regex_rule(".*_(\\d+)\\.db3", std::regex_constants::ECMAScript); + + std::smatch first_match; + std::smatch second_match; + + auto first_path_string = first_path.string(); + auto second_path_string = second_path.string(); + + auto first_regex_good = std::regex_match(first_path_string, first_match, regex_rule); + auto second_regex_good = std::regex_match(second_path_string, second_match, regex_rule); + + // Make sure the paths have regex matches + if (!first_regex_good || !second_regex_good) { + throw std::runtime_error("Malformed relative file name. Expected numerical identifier."); + } + + // Convert database numbers to uint + u_int32_t first_db_num = std::stoul(first_match.str(1), nullptr, 10); + u_int32_t second_db_num = std::stoul(second_match.str(1), nullptr, 10); + + return first_db_num < second_db_num; +} + +std::vector SequentialReindexer::get_database_files( + const rcpputils::fs::path & base_folder) +{ + // Look in the uri directory to see what database files are there + std::vector output; + + #ifdef WIN32 + { + // Code placed in scope so variables can't accidentally be used elsewhere + WIN32_FIND_DATA FindFileData; + HANDLE hFind; + + std::string search_dir = base_folder.string() + "\\*"; + hFind = FindFirstFile(search_dir, &FindFileData); + // If an error occurs, we want to abort + if (hFind == INVALID_HANDLE_VALUE) { + DWORD dwError = GetLastError(); + std::error_code ec(dwError, std::system_category()); + throw(std::system_error(ec)); + } + + // Loop through the directory, collecting file names as we go + do { + if (ffd.dwFileAttributes & FILE_ATTRIBUTE_DIRECTORY) { + // I guess it's a directory in the bag file? + // Still, not interested in it. + continue; + } else { + auto temp_path = rcpputils::fs::path(ffd.cFileName); + + // We are ONLY interested in database files + if (temp_path.extension().string() != ".db3") { + continue; + } + output.emplace_back(temp_path); + } + } while (FindNextFile(hFind, &ffd) != 0); + + FindClose(hFind); + } + #else + { + // Code placed in scope so variables can't accidentally be used elsewhere + auto dirp = opendir(base_folder.string().c_str()); + // If an error occurs, we want to abort + if (dirp == NULL) { + throw std::system_error(errno, std::generic_category()); + } + dirent * dp; + while ((dp = readdir(dirp)) != NULL) { + auto non_const_folder = rcpputils::fs::path(base_folder); + auto temp_path = non_const_folder /= rcpputils::fs::path(dp->d_name); + + // We are ONLY interested in database files + if (temp_path.extension().string() != ".db3") { + continue; + } + + output.emplace_back(temp_path); + } + closedir(dirp); + } + #endif + + // Sort relative file path by database number + std::sort( + output.begin(), output.end(), + [](rcpputils::fs::path a, rcpputils::fs::path b) {return comp_rel_file(a, b);}); + + return output; +} + +void SequentialReindexer::open(const rosbag2_storage::StorageOptions & storage_options) +{ + // Since this is a reindexing operation, assume that there is no metadata.yaml file. + // As such, ask the storage with the given URI for its metadata. + storage_ = storage_factory_->open_read_only(storage_options); + if (!storage_) { + throw std::runtime_error{"No storage could be initialized. Abort"}; + } +} + +void SequentialReindexer::fill_topics_metadata() +{ + rcpputils::check_true(storage_ != nullptr, "Bag is not open. Call open() before reading."); + topics_metadata_.clear(); + topics_metadata_.reserve(metadata_.topics_with_message_count.size()); + for (const auto & topic_information : metadata_.topics_with_message_count) { + topics_metadata_.push_back(topic_information.topic_metadata); + } +} + +void SequentialReindexer::init_metadata( + const std::vector & files, + const rosbag2_storage::StorageOptions & storage_options) +{ + metadata_ = rosbag2_storage::BagMetadata{}; + + // This reindexer will only work on SQLite files, so this can't change + metadata_.storage_identifier = storage_options.storage_id; + metadata_.starting_time = std::chrono::time_point( + std::chrono::nanoseconds::max()); + + // Record the relative paths to the metadata + for (const auto & path : files) { + auto cleaned_path = strip_parent_path(path); + metadata_.relative_file_paths.push_back(cleaned_path); + } +} + +void SequentialReindexer::aggregate_metadata( + const std::vector & files, + const rosbag2_storage::StorageOptions & storage_options) +{ + // In order to most accurately reconstruct the metadata, we need to + // visit each of the contained relative database files in the bag, + // open them, slurp up the info, and stuff it into the master + // metadata object. + ROSBAG2_CPP_LOG_INFO_STREAM("Extracting metadata from database(s)"); + for (const auto & f_ : files) { + rosbag2_storage::StorageOptions temp_so = { + f_.string(), // uri + storage_options.storage_id, // storage_id + storage_options.max_bagfile_size, // max_bagfile_size + storage_options.max_bagfile_duration, // max_bagfile_duration + storage_options.max_cache_size, // max_cache_size + storage_options.storage_config_uri // storage_config_uri + }; + open(temp_so); // Class storage_ is now full + + auto temp_metadata = storage_->get_metadata(); + + // Last opened file will have our starting time + metadata_.starting_time = temp_metadata.starting_time; + metadata_.duration += temp_metadata.duration; + metadata_.message_count += temp_metadata.message_count; + + // Add the topic metadata + for (const auto & topic : temp_metadata.topics_with_message_count) { + auto found_topic = std::find_if( + metadata_.topics_with_message_count.begin(), + metadata_.topics_with_message_count.end(), + [&topic](const rosbag2_storage::TopicInformation & agg_topic) + {return topic.topic_metadata.name == agg_topic.topic_metadata.name;}); + if (found_topic == metadata_.topics_with_message_count.end()) { + // It's a new topic. Add it. + metadata_.topics_with_message_count.emplace_back(topic); + } else { + // Merge in the new information + found_topic->message_count += topic.message_count; + if (topic.topic_metadata.offered_qos_profiles != "") { + found_topic->topic_metadata.offered_qos_profiles = + topic.topic_metadata.offered_qos_profiles; + } + if (topic.topic_metadata.serialization_format != "") { + found_topic->topic_metadata.serialization_format = + topic.topic_metadata.serialization_format; + } + if (topic.topic_metadata.type != "") { + found_topic->topic_metadata.type = topic.topic_metadata.type; + } + } + } + + ROSBAG2_CPP_LOG_INFO("Closing database"); + storage_.reset(); // Class storage_ is now empty + } +} + +void SequentialReindexer::reindex(const rosbag2_storage::StorageOptions & storage_options) +{ + ROSBAG2_CPP_LOG_INFO("Beginning Reindex Operation."); + + // Identify all database files + base_folder_ = storage_options.uri; + auto files = get_database_files(base_folder_); + std::cout << "Finished getting database files\n"; + if (files.empty()) { + ROSBAG2_CPP_LOG_ERROR("No database files found for reindexing. Abort"); + return; + } + + // Create initial metadata + init_metadata(files, storage_options); + + // Collect all metadata from database files + aggregate_metadata(files, storage_options); + + // Perform final touch-up + finalize_metadata(); + + metadata_io_->write_metadata(base_folder_.string(), metadata_); + ROSBAG2_CPP_LOG_INFO("Reindexing operation completed."); +} + +void SequentialReindexer::finalize_metadata() +{ + metadata_.bag_size = 0; + + for (const auto & path : metadata_.relative_file_paths) { + const auto bag_path = rcpputils::fs::path{path}; + + if (bag_path.exists()) { + metadata_.bag_size += bag_path.file_size(); + } + } +} +} // namespace reindexers +} // namespace rosbag2_cpp diff --git a/rosbag2_storage_default_plugins/src/rosbag2_storage_default_plugins/sqlite/sqlite_storage.cpp b/rosbag2_storage_default_plugins/src/rosbag2_storage_default_plugins/sqlite/sqlite_storage.cpp index 1dd6df1168..ab72332b27 100644 --- a/rosbag2_storage_default_plugins/src/rosbag2_storage_default_plugins/sqlite/sqlite_storage.cpp +++ b/rosbag2_storage_default_plugins/src/rosbag2_storage_default_plugins/sqlite/sqlite_storage.cpp @@ -430,19 +430,19 @@ rosbag2_storage::BagMetadata SqliteStorage::get_metadata() auto statement = database_->prepare_statement( "SELECT name, type, serialization_format, COUNT(messages.id), MIN(messages.timestamp), " - "MAX(messages.timestamp) " + "MAX(messages.timestamp), offered_qos_profiles " "FROM messages JOIN topics on topics.id = messages.topic_id " "GROUP BY topics.name;"); auto query_results = statement->execute_query< std::string, std::string, std::string, int, rcutils_time_point_value_t, - rcutils_time_point_value_t>(); + rcutils_time_point_value_t, std::string>(); rcutils_time_point_value_t min_time = INT64_MAX; rcutils_time_point_value_t max_time = 0; for (auto result : query_results) { metadata.topics_with_message_count.push_back( { - {std::get<0>(result), std::get<1>(result), std::get<2>(result), ""}, + {std::get<0>(result), std::get<1>(result), std::get<2>(result), std::get<6>(result)}, static_cast(std::get<3>(result)) }); diff --git a/rosbag2_tests/CMakeLists.txt b/rosbag2_tests/CMakeLists.txt index da7029eb4c..eeb6ee4574 100644 --- a/rosbag2_tests/CMakeLists.txt +++ b/rosbag2_tests/CMakeLists.txt @@ -42,6 +42,7 @@ if(BUILD_TESTING) find_package(rosbag2_test_common REQUIRED) find_package(std_msgs REQUIRED) find_package(test_msgs REQUIRED) + find_package(yaml-cpp REQUIRED) ament_add_gmock(test_rosbag2_record_end_to_end test/rosbag2_tests/test_rosbag2_record_end_to_end.cpp @@ -80,6 +81,16 @@ if(BUILD_TESTING) ament_add_test_label(test_rosbag2_info_end_to_end xfail) endif() + ament_add_gmock(test_rosbag2_reindex_end_to_end + test/rosbag2_tests/test_rosbag2_reindex_end_to_end.cpp + WORKING_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}) + if(TARGET test_rosbag2_reindex_end_to_end) + ament_target_dependencies(test_rosbag2_reindex_end_to_end + rosbag2_storage + rosbag2_storage_default_plugins + rosbag2_test_common) + endif() + ament_add_gmock(test_converter test/rosbag2_tests/test_converter.cpp WORKING_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}) diff --git a/rosbag2_tests/test/rosbag2_tests/test_rosbag2_reindex_end_to_end.cpp b/rosbag2_tests/test/rosbag2_tests/test_rosbag2_reindex_end_to_end.cpp new file mode 100644 index 0000000000..3f550d6051 --- /dev/null +++ b/rosbag2_tests/test/rosbag2_tests/test_rosbag2_reindex_end_to_end.cpp @@ -0,0 +1,407 @@ +// Copyright 2020 DCS Corporation, All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// +// DISTRIBUTION A. Approved for public release; distribution unlimited. +// OPSEC #4584. +// +// Delivered to the U.S. Government with Unlimited Rights, as defined in DFARS +// Part 252.227-7013 or 7014 (Feb 2014). +// +// This notice must appear in all copies of this file and its derivatives. + +#include + +#include + +#include +#include +#include +#include +#include +#include + +// rclcpp must be included before process_execution_helpers.hpp +#include "rclcpp/rclcpp.hpp" + +#include "rosbag2_storage_default_plugins/sqlite/sqlite_storage.hpp" + +#include "rosbag2_test_common/subscription_manager.hpp" +#include "rosbag2_test_common/process_execution_helpers.hpp" + +// #include "test_msgs/msg/arrays.hpp" +// #include "test_msgs/msg/basic_types.hpp" +// #include "test_msgs/message_fixtures.hpp" + +#include "yaml-cpp/yaml.h" + +using namespace ::testing; // NOLINT +using namespace rosbag2_test_common; // NOLINT + +class ReindexEndToEndTestFixture : public Test +{ +public: + ReindexEndToEndTestFixture() + { + database_path_ = _SRC_RESOURCES_DIR_PATH; // variable defined in CMakeLists.txt + } + + static void SetUpTestCase() + { + rclcpp::init(0, nullptr); + } + + static void TearDownTestCase() + { + rclcpp::shutdown(); + } + + std::string database_path_; + std::unique_ptr sub_; + + // Declare keys to avoid potential typos + const char * const ROOT_KEY = "rosbag2_bagfile_information"; + const char * const VERSION_KEY = "version"; + const char * const SI_KEY = "storage_identifier"; + const char * const RFP_KEY = "relative_file_paths"; + const char * const DURATION_KEY = "duration"; + const char * const NS_KEY = "nanoseconds"; + const char * const ST_KEY = "starting_time"; + const char * const NSSE_KEY = "nanoseconds_since_epoch"; + const char * const COUNT_KEY = "message_count"; + + // Topics With Message Count section + const char * const TWMC_KEY = "topics_with_message_count"; + const char * const TOPIC_KEY = "topic_metadata"; + const char * const NAME_KEY = "name"; + const char * const TYPE_KEY = "type"; + const char * const SF_KEY = "serialization_format"; + const char * const QOS_KEY = "offered_qos_profiles"; + + // Compression section + const char * const CFORMAT_KEY = "compression_format"; + const char * const MODE_KEY = "compression_mode"; +}; + +/** FOR REFERENCE, A MAP OF A METADATA.YAML FILE + * + * rosbag2_bagfile_information: + * version: + * storage_identifier: + * relative_file_paths: + * - + * duration: + * nanoseconds: + * starting_time: + * nanoseconds_since_epoch: + * message_count: + * topics_with_message_count: + * - topic_metadata: + * name: + * type: + * serialization_format: + * offered_qos_profiles: + * message_count: + * compression_format: + * compression_mode: + * + * */ +#ifndef _WIN32 +TEST_F(ReindexEndToEndTestFixture, reindex_end_to_end_test) { + // We will be creating a new metadata file (hopefully), so preserve the old one + std::string old_name = database_path_ + "/cdr_test/metadata.yaml"; + std::string new_name = database_path_ + "/cdr_test/metadata_old.yaml"; + + bool rename_success = std::rename(old_name.c_str(), new_name.c_str()) == 0; + EXPECT_EQ(rename_success, true); + if (!rename_success) { + std::perror("Error occurred during rename: "); + return; + } + + // Try to create the metadata file + auto exit_code = execute_and_wait_until_completion("ros2 bag reindex cdr_test", database_path_); + + // First check to see if we could create a file at all + EXPECT_EQ(exit_code, EXIT_SUCCESS); + if (exit_code != EXIT_SUCCESS) { + // Delete any created metadata file (just in case) + std::remove(old_name.c_str()); + std::rename(new_name.c_str(), old_name.c_str()); + return; + } + + // If we're at this point, that means a metadata file was cretaed. + // Read both the old and new metadata files in. + YAML::Node original_metadata = YAML::LoadFile(new_name.c_str())[ROOT_KEY]; + YAML::Node new_metadata = YAML::LoadFile(old_name.c_str()); + + // Easy test first - check for the root node + /** + * + * rosbag2_bagfile_information: <- WE ARE HERE + * version: + * storage_identifier: + * ... + * */ + EXPECT_TRUE(new_metadata[ROOT_KEY]); + if (!new_metadata[ROOT_KEY]) { + // Something went wrong. Abort. + std::remove(old_name.c_str()); + std::rename(new_name.c_str(), old_name.c_str()); + return; + } + + // All further tests use the root node + YAML::Node root_node = new_metadata[ROOT_KEY]; + + // Check the version + /** + * > + * rosbag2_bagfile_information: + * version: <- WE ARE HERE + * storage_identifier: + * relative_file_paths: + * ... + * */ + std::cout << "Checking version" << std::endl; + EXPECT_TRUE(root_node[VERSION_KEY]); + if (root_node[VERSION_KEY]) { + int new_version = root_node[VERSION_KEY].as(); + int original_version = original_metadata[VERSION_KEY].as(); + EXPECT_EQ(new_version, original_version); + } + + // Test storage identifier + /** + * ... + * rosbag2_bagfile_information: + * version: + * storage_identifier: <- WE ARE HERE + * relative_file_paths: + * - + * ... + * */ + std::cout << "Checking storage identifier" << std::endl; + EXPECT_TRUE(root_node[SI_KEY]); + if (root_node[SI_KEY]) { + std::string new_si = root_node[SI_KEY].as(); + std::string old_si = original_metadata[SI_KEY].as(); + EXPECT_EQ(new_si, old_si); + } + + // Test relative file paths + /** + * ... + * version: + * storage_identifier: + * relative_file_paths: <- WE ARE HERE + * - <- AND HERE TOO + * duration: + * ... + * */ + std::cout << "Checking relative file paths" << std::endl; + EXPECT_TRUE(root_node[RFP_KEY]); + if (root_node[RFP_KEY]) { + std::vector new_filepaths = + root_node[RFP_KEY].as>(); + std::vector original_filepaths = + original_metadata[RFP_KEY].as>(); + EXPECT_EQ(new_filepaths, original_filepaths); + } + + // Test duration + /** + * ... + * relative_file_paths: + * - + * duration: <- WE ARE HERE + * nanoseconds: <- AND HERE TOO + * starting_time: + * nanoseconds_since_epoch: + * ... + * */ + std::cout << "Checking duration" << std::endl; + EXPECT_TRUE(root_node[DURATION_KEY]); + if (root_node[DURATION_KEY]) { + YAML::Node duration_node = root_node[DURATION_KEY]; + YAML::Node original_duration_node = original_metadata[DURATION_KEY]; + EXPECT_TRUE(duration_node[NS_KEY]); + if (duration_node[NS_KEY]) { + int64_t new_duration = duration_node[NS_KEY].as(); + int64_t original_duration = original_duration_node[NS_KEY].as(); + EXPECT_EQ(new_duration, original_duration); + } + } + + // Test starting time + /** + * ... + * duration: + * nanoseconds: + * starting_time: <- WE ARE HERE + * nanoseconds_since_epoch: <- AND HERE TOO + * message_count: + * topics_with_message_count: + * ... + * */ + std::cout << "Checking starting_time" << std::endl; + EXPECT_TRUE(root_node[ST_KEY]); + if (root_node[ST_KEY]) { + YAML::Node starting_time = root_node[ST_KEY]; + YAML::Node original_starting_time_node = original_metadata[ST_KEY]; + EXPECT_TRUE(starting_time[NSSE_KEY]); + if (starting_time[NSSE_KEY]) { + int64_t new_start = starting_time[NSSE_KEY].as(); + int64_t original_start = original_starting_time_node[NSSE_KEY].as(); + EXPECT_EQ(new_start, original_start); + } + } + + // Test message count + /** + * ... + * starting_time: + * nanoseconds_since_epoch: + * message_count: <- WE ARE HERE + * topics_with_message_count: + *... + * */ + std::cout << "Checking message count" << std::endl; + EXPECT_TRUE(root_node[COUNT_KEY]); + if (root_node[COUNT_KEY]) { + int new_count = root_node[COUNT_KEY].as(); + int original_count = original_metadata[COUNT_KEY].as(); + EXPECT_EQ(new_count, original_count); + } + + // Test topics + /** + * ... + * message_count: + * topics_with_message_count: <- WE ARE HERE + * - topic_metadata: <- AND HERE TOO + * name: <- AND HERE AS WELL + * type: <- AND ALSO HERE + * serialization_format: <- AND EVEN HERE + * offered_qos_profiles: <- AND HERE + * message_count: <- AND HERE + * compression_format: + * ... + * */ + std::cout << "Checking topics" << std::endl; + EXPECT_TRUE(root_node[TWMC_KEY]); + if (root_node[TWMC_KEY]) { + YAML::Node new_twmc_node = root_node[TWMC_KEY]; + YAML::Node old_twmc_node = original_metadata[TWMC_KEY]; + + // The ordering of topics is not guaranteed to be preserved during + // the reindexing process. So we have to iterate over each topic + // in the new metadata, and see if it exists in the original + for (YAML::const_iterator new_metadata_iter = new_twmc_node.begin(); + new_metadata_iter != new_twmc_node.end(); + ++new_metadata_iter) + { + bool found_match = false; + YAML::Node new_topic = *new_metadata_iter; + YAML::Node new_topic_metadata = new_topic[TOPIC_KEY]; + std::string new_topic_name = new_topic_metadata[NAME_KEY].as(); + + for (YAML::const_iterator orig_metadata_iter = old_twmc_node.begin(); + orig_metadata_iter != old_twmc_node.end(); + ++orig_metadata_iter) + { + YAML::Node old_topic = *orig_metadata_iter; + YAML::Node old_topic_metadata = old_topic[TOPIC_KEY]; + std::string old_topic_name = old_topic_metadata[NAME_KEY].as(); + + if (new_topic_name == old_topic_name) { + // A match! The testing can now continue + found_match = true; + // Check that the other topic metadata matches... + // Check type: + std::string new_topic_type = new_topic_metadata[TYPE_KEY].as(); + std::string old_topic_type = old_topic_metadata[TYPE_KEY].as(); + EXPECT_EQ(new_topic_type, old_topic_type); + + // Check serialization format + std::string new_s_fmt = new_topic_metadata[SF_KEY].as(); + std::string old_s_fmt = old_topic_metadata[SF_KEY].as(); + EXPECT_EQ(new_s_fmt, old_s_fmt); + + // Check qos profiles + std::string new_qos = new_topic_metadata[QOS_KEY].as(); + std::string old_qos = old_topic_metadata[QOS_KEY].as(); + EXPECT_EQ(new_qos, old_qos); + + // Check message count + int64_t new_count = new_topic[COUNT_KEY].as(); + int64_t old_count = old_topic[COUNT_KEY].as(); + EXPECT_EQ(new_count, old_count); + break; + } + } + + EXPECT_TRUE(found_match); // We expected to have a name match + } + } + + // Test compression format + /** + * ... + * serialization_format: + * offered_qos_profiles: + * message_count: + * compression_format: <- WE ARE HERE + * compression_mode: + * + * */ + std::cout << "Checking compression format" << std::endl; + EXPECT_TRUE(root_node[CFORMAT_KEY]); + if (root_node[CFORMAT_KEY]) { + std::string new_format = root_node[CFORMAT_KEY].as(); + std::string old_format = original_metadata[CFORMAT_KEY].as(); + EXPECT_EQ(new_format, old_format); + } + + // Test compression mode + /** + * ... + * compression_format: + * compression_mode: <- WE ARE HERE + * + * */ + std::cout << "Checking compression mode" << std::endl; + EXPECT_TRUE(root_node[MODE_KEY]); + if (root_node[MODE_KEY]) { + std::string new_mode = root_node[MODE_KEY].as(); + std::string old_mode = original_metadata[MODE_KEY].as(); + EXPECT_EQ(new_mode, old_mode); + } + + // Cleanup + std::remove(old_name.c_str()); + std::rename(new_name.c_str(), old_name.c_str()); +} +#endif + +TEST_F(ReindexEndToEndTestFixture, reindex_fails_gracefully_if_bag_does_not_exist) { + internal::CaptureStderr(); + auto exit_code = + execute_and_wait_until_completion("ros2 bag reindex does_not_exist", database_path_); + auto error_output = internal::GetCapturedStderr(); + + // Exit code could be EXIT_FAILURE (1) or 2 (no such file or directory) + EXPECT_NE(exit_code, EXIT_SUCCESS); + EXPECT_THAT(error_output, HasSubstr("'does_not_exist' does not exist")); +} diff --git a/rosbag2_transport/include/rosbag2_transport/rosbag2_transport.hpp b/rosbag2_transport/include/rosbag2_transport/rosbag2_transport.hpp index 2d7f2af575..bf53478d3d 100644 --- a/rosbag2_transport/include/rosbag2_transport/rosbag2_transport.hpp +++ b/rosbag2_transport/include/rosbag2_transport/rosbag2_transport.hpp @@ -31,6 +31,7 @@ namespace rosbag2_cpp class Info; class Reader; class Writer; +class Reindexer; } // namespace rosbag2_cpp namespace rosbag2_transport @@ -50,7 +51,8 @@ class Rosbag2Transport Rosbag2Transport( std::shared_ptr reader, std::shared_ptr writer, - std::shared_ptr info); + std::shared_ptr info, + std::shared_ptr reindexer); ROSBAG2_TRANSPORT_PUBLIC void init(); @@ -89,6 +91,16 @@ class Rosbag2Transport ROSBAG2_TRANSPORT_PUBLIC void print_bag_info(const std::string & uri, const std::string & storage_id); + /** + * Reconstruct the metadata yaml file from the stored data in the bag. + * + * \param uri path to the bag + * \param storage_options Options regarding the storage (e.g. bag file name) + * \param record_options Options regarding how the file was recorded (e.g. compression format) + */ + ROSBAG2_TRANSPORT_PUBLIC + void reindex(const rosbag2_storage::StorageOptions & storage_options); + private: std::shared_ptr setup_node( std::string node_prefix = "", @@ -97,6 +109,7 @@ class Rosbag2Transport std::shared_ptr reader_; std::shared_ptr writer_; std::shared_ptr info_; + std::shared_ptr reindexer_; std::shared_ptr transport_node_; }; diff --git a/rosbag2_transport/src/rosbag2_transport/rosbag2_transport.cpp b/rosbag2_transport/src/rosbag2_transport/rosbag2_transport.cpp index 57c866361d..b168744077 100644 --- a/rosbag2_transport/src/rosbag2_transport/rosbag2_transport.cpp +++ b/rosbag2_transport/src/rosbag2_transport/rosbag2_transport.cpp @@ -31,6 +31,10 @@ #include "rosbag2_cpp/typesupport_helpers.hpp" #include "rosbag2_cpp/writer.hpp" #include "rosbag2_cpp/writers/sequential_writer.hpp" +#include "rosbag2_cpp/reindexer.hpp" +#include "rosbag2_cpp/reindexers/sequential_reindexer.hpp" + +#include "rosbag2_storage/storage_options.hpp" #include "rosbag2_transport/logging.hpp" @@ -47,14 +51,20 @@ Rosbag2Transport::Rosbag2Transport() std::make_unique())), writer_(std::make_shared( std::make_unique())), - info_(std::make_shared()) + info_(std::make_shared()), + reindexer_(std::make_shared( + std::make_unique())) {} Rosbag2Transport::Rosbag2Transport( std::shared_ptr reader, std::shared_ptr writer, - std::shared_ptr info) -: reader_(std::move(reader)), writer_(std::move(writer)), info_(std::move(info)) {} + std::shared_ptr info, + std::shared_ptr reindexer) +: reader_(std::move(reader)), + writer_(std::move(writer)), + info_(std::move(info)), + reindexer_(std::move(reindexer)) {} void Rosbag2Transport::init() { @@ -109,6 +119,16 @@ void Rosbag2Transport::play( } } +void Rosbag2Transport::reindex( + const rosbag2_storage::StorageOptions & storage_options) +{ + try { + reindexer_->reindex(storage_options); + } catch (std::runtime_error & e) { + ROSBAG2_TRANSPORT_LOG_ERROR("Failed to reindex: %s", e.what()); + } +} + void Rosbag2Transport::print_bag_info(const std::string & uri, const std::string & storage_id) { rosbag2_storage::BagMetadata metadata; diff --git a/rosbag2_transport/src/rosbag2_transport/rosbag2_transport_python.cpp b/rosbag2_transport/src/rosbag2_transport/rosbag2_transport_python.cpp index 024038a64d..a8f8f766bf 100644 --- a/rosbag2_transport/src/rosbag2_transport/rosbag2_transport_python.cpp +++ b/rosbag2_transport/src/rosbag2_transport/rosbag2_transport_python.cpp @@ -31,6 +31,8 @@ #include "rosbag2_cpp/readers/sequential_reader.hpp" #include "rosbag2_cpp/writer.hpp" #include "rosbag2_cpp/writers/sequential_writer.hpp" +#include "rosbag2_cpp/reindexer.hpp" +#include "rosbag2_cpp/reindexers/sequential_reindexer.hpp" #include "rosbag2_storage/metadata_io.hpp" #include "rosbag2_storage/storage_options.hpp" #include "rosbag2_transport/rosbag2_transport.hpp" @@ -205,6 +207,8 @@ rosbag2_transport_record(PyObject * Py_UNUSED(self), PyObject * args, PyObject * auto info = std::make_shared(); auto reader = std::make_shared( std::make_unique()); + auto reindexer = std::make_shared( + std::make_unique()); std::shared_ptr writer; // Change writer based on recording options if (record_options.compression_format == "zstd") { @@ -215,7 +219,7 @@ rosbag2_transport_record(PyObject * Py_UNUSED(self), PyObject * args, PyObject * std::make_unique()); } - rosbag2_transport::Rosbag2Transport transport(reader, writer, info); + rosbag2_transport::Rosbag2Transport transport(reader, writer, info, reindexer); transport.init(); transport.record(storage_options, record_options); transport.shutdown(); @@ -319,6 +323,8 @@ rosbag2_transport_play(PyObject * Py_UNUSED(self), PyObject * args, PyObject * k rosbag2_storage::BagMetadata metadata{}; // Specify defaults auto info = std::make_shared(); + auto reindexer = std::make_shared( + std::make_unique()); std::shared_ptr reader; auto writer = std::make_shared( std::make_unique()); @@ -337,7 +343,7 @@ rosbag2_transport_play(PyObject * Py_UNUSED(self), PyObject * args, PyObject * k std::make_unique()); } - rosbag2_transport::Rosbag2Transport transport(reader, writer, info); + rosbag2_transport::Rosbag2Transport transport(reader, writer, info, reindexer); transport.init(); transport.play(storage_options, play_options); transport.shutdown(); @@ -367,6 +373,62 @@ rosbag2_transport_info(PyObject * Py_UNUSED(self), PyObject * args, PyObject * k Py_RETURN_NONE; } +static PyObject * +rosbag2_transport_reindex(PyObject * Py_UNUSED(self), PyObject * args, PyObject * kwargs) +{ + rosbag2_storage::StorageOptions storage_options{}; + rosbag2_transport::RecordOptions record_options{}; + static const char * kwlist[] = + {"uri", "storage_id", "serialization_format", "compression_format", nullptr}; + + char * char_uri = nullptr; + char * char_storage_id = nullptr; + char * char_serialization_fmt = nullptr; + char * char_compression_fmt = nullptr; + if (!PyArg_ParseTupleAndKeywords( + args, kwargs, "ssss", const_cast(kwlist), + &char_uri, + &char_storage_id, + &char_serialization_fmt, + &char_compression_fmt)) + { + return nullptr; + } + + storage_options.uri = std::string(char_uri); + storage_options.storage_id = std::string(char_storage_id); + record_options.compression_format = std::string(char_compression_fmt); + + record_options.rmw_serialization_format = std::string(char_serialization_fmt).empty() ? + rmw_get_serialization_format() : + char_serialization_fmt; + + // Specify defaults + auto info = std::make_shared(); + auto reader = std::make_shared( + std::make_unique()); + auto reindexer = std::make_shared( + std::make_unique()); + // std::shared_ptr writer; + // // Change writer based on recording options + // if (record_options.compression_format == "zstd") { + // writer = std::make_shared( + // std::make_unique(compression_options)); + // } else { + // writer = std::make_shared( + // std::make_unique()); + // } + auto writer = std::make_shared( + std::make_unique()); + + rosbag2_transport::Rosbag2Transport transport(reader, writer, info, reindexer); + transport.init(); + transport.reindex(storage_options); + transport.shutdown(); + + Py_RETURN_NONE; +} + /// Define the public methods of this module #if __GNUC__ >= 8 # pragma GCC diagnostic push @@ -385,6 +447,12 @@ static PyMethodDef rosbag2_transport_methods[] = { "info", reinterpret_cast(rosbag2_transport_info), METH_VARARGS | METH_KEYWORDS, "Print bag info" }, + { + "reindex", + reinterpret_cast(rosbag2_transport_reindex), + METH_VARARGS | METH_KEYWORDS, + "Reindex bag" + }, {nullptr, nullptr, 0, nullptr} /* sentinel */ }; #if __GNUC__ >= 8 diff --git a/rosbag2_transport/test/rosbag2_transport/record_integration_fixture.hpp b/rosbag2_transport/test/rosbag2_transport/record_integration_fixture.hpp index 1e794e878c..8eb2d98d8d 100644 --- a/rosbag2_transport/test/rosbag2_transport/record_integration_fixture.hpp +++ b/rosbag2_transport/test/rosbag2_transport/record_integration_fixture.hpp @@ -51,7 +51,7 @@ class RecordIntegrationTestFixture : public Rosbag2TransportTestFixture // the future object returned from std::async needs to be stored not to block the execution future_ = std::async( std::launch::async, [this, options]() { - rosbag2_transport::Rosbag2Transport rosbag2_transport(reader_, writer_, info_); + rosbag2_transport::Rosbag2Transport rosbag2_transport(reader_, writer_, info_, reindexer_); rosbag2_transport.record(storage_options_, options); }); } diff --git a/rosbag2_transport/test/rosbag2_transport/rosbag2_transport_test_fixture.hpp b/rosbag2_transport/test/rosbag2_transport/rosbag2_transport_test_fixture.hpp index f224683a3a..a7737a27f6 100644 --- a/rosbag2_transport/test/rosbag2_transport/rosbag2_transport_test_fixture.hpp +++ b/rosbag2_transport/test/rosbag2_transport/rosbag2_transport_test_fixture.hpp @@ -87,6 +87,7 @@ class Rosbag2TransportTestFixture : public Test std::shared_ptr reader_; std::shared_ptr writer_; std::shared_ptr info_; + std::shared_ptr reindexer_; }; #endif // ROSBAG2_TRANSPORT__ROSBAG2_TRANSPORT_TEST_FIXTURE_HPP_ diff --git a/rosbag2_transport/test/rosbag2_transport/test_info.cpp b/rosbag2_transport/test/rosbag2_transport/test_info.cpp index 564079e67f..f2485cfdd7 100644 --- a/rosbag2_transport/test/rosbag2_transport/test_info.cpp +++ b/rosbag2_transport/test/rosbag2_transport/test_info.cpp @@ -41,7 +41,7 @@ TEST_F(Rosbag2TransportTestFixture, info_pretty_prints_information_from_bagfile) EXPECT_CALL(*info_, read_metadata(_, _)).WillOnce(Return(bagfile)); // the expected output uses a regex to handle different time zones. - rosbag2_transport::Rosbag2Transport transport(reader_, writer_, info_); + rosbag2_transport::Rosbag2Transport transport(reader_, writer_, info_, reindexer_); transport.print_bag_info("test", "sqlite3"); std::string output = internal::GetCapturedStdout(); diff --git a/rosbag2_transport/test/rosbag2_transport/test_play.cpp b/rosbag2_transport/test/rosbag2_transport/test_play.cpp index 6b1652e8dc..4be231d756 100644 --- a/rosbag2_transport/test/rosbag2_transport/test_play.cpp +++ b/rosbag2_transport/test/rosbag2_transport/test_play.cpp @@ -75,7 +75,7 @@ TEST_F(RosBag2PlayTestFixture, recorded_messages_are_played_for_all_topics) auto await_received_messages = sub_->spin_subscriptions(); - Rosbag2Transport rosbag2_transport(reader_, writer_, info_); + Rosbag2Transport rosbag2_transport(reader_, writer_, info_, reindexer_); rosbag2_transport.play(storage_options_, play_options_); await_received_messages.get(); @@ -144,7 +144,7 @@ TEST_F(RosBag2PlayTestFixture, recorded_messages_are_played_for_all_topics_with_ auto await_received_messages = sub_->spin_subscriptions(); - Rosbag2Transport rosbag2_transport(reader_, writer_, info_); + Rosbag2Transport rosbag2_transport(reader_, writer_, info_, reindexer_); rosbag2_transport.play(storage_options_, play_options_); await_received_messages.get(); @@ -213,7 +213,7 @@ TEST_F(RosBag2PlayTestFixture, recorded_messages_are_played_for_filtered_topics) auto await_received_messages = sub_->spin_subscriptions(); - Rosbag2Transport rosbag2_transport(reader_, writer_, info_); + Rosbag2Transport rosbag2_transport(reader_, writer_, info_, reindexer_); rosbag2_transport.play(storage_options_, play_options_); await_received_messages.get(); @@ -236,7 +236,7 @@ TEST_F(RosBag2PlayTestFixture, recorded_messages_are_played_for_filtered_topics) await_received_messages = sub_->spin_subscriptions(); - rosbag2_transport = Rosbag2Transport(reader_, writer_, info_); + rosbag2_transport = Rosbag2Transport(reader_, writer_, info_, reindexer_); rosbag2_transport.play(storage_options_, play_options_); await_received_messages.get(); @@ -257,7 +257,7 @@ TEST_F(RosBag2PlayTestFixture, recorded_messages_are_played_for_filtered_topics) await_received_messages = sub_->spin_subscriptions(); - rosbag2_transport = Rosbag2Transport(reader_, writer_, info_); + rosbag2_transport = Rosbag2Transport(reader_, writer_, info_, reindexer_); rosbag2_transport.play(storage_options_, play_options_); await_received_messages.get(); @@ -309,7 +309,7 @@ TEST_F(RosBag2PlayTestFixture, recorded_messages_are_played_for_filtered_topics_ auto await_received_messages = sub_->spin_subscriptions(); - Rosbag2Transport rosbag2_transport(reader_, writer_, info_); + Rosbag2Transport rosbag2_transport(reader_, writer_, info_, reindexer_); play_options_.topics_to_filter = {"topic2"}; rosbag2_transport.play(storage_options_, play_options_); @@ -335,7 +335,7 @@ TEST_F(RosBag2PlayTestFixture, recorded_messages_are_played_for_filtered_topics_ await_received_messages = sub_->spin_subscriptions(); - rosbag2_transport = Rosbag2Transport(reader_, writer_, info_); + rosbag2_transport = Rosbag2Transport(reader_, writer_, info_, reindexer_); play_options_.topics_to_filter = {"topic1"}; rosbag2_transport.play(storage_options_, play_options_); @@ -362,7 +362,7 @@ TEST_F(RosBag2PlayTestFixture, recorded_messages_are_played_for_filtered_topics_ await_received_messages = sub_->spin_subscriptions(); - rosbag2_transport = Rosbag2Transport(reader_, writer_, info_); + rosbag2_transport = Rosbag2Transport(reader_, writer_, info_, reindexer_); play_options_.topics_to_filter = {"topic1", "topic2"}; rosbag2_transport.play(storage_options_, play_options_); @@ -417,7 +417,7 @@ class RosBag2PlayQosOverrideTestFixture : public RosBag2PlayTestFixture void play_and_wait(Duration timeout, bool expect_timeout = false) { auto await_received_messages = sub_->spin_subscriptions(); - Rosbag2Transport transport{reader_, writer_, info_}; + Rosbag2Transport transport{reader_, writer_, info_, reindexer_}; transport.play(storage_options_, play_options_); const auto result = await_received_messages.wait_for(timeout); // Must EXPECT, can't ASSERT because transport needs to be shutdown if timed out diff --git a/rosbag2_transport/test/rosbag2_transport/test_play_loop.cpp b/rosbag2_transport/test/rosbag2_transport/test_play_loop.cpp index aaa5c1bfc4..afb3ec0d87 100644 --- a/rosbag2_transport/test/rosbag2_transport/test_play_loop.cpp +++ b/rosbag2_transport/test/rosbag2_transport/test_play_loop.cpp @@ -67,7 +67,8 @@ TEST_F(RosBag2PlayTestFixture, messages_played_in_loop) { auto rosbag2_transport_ptr = std::make_shared( reader_, writer_, - info_); + info_, + reindexer_); std::thread loop_thread(&rosbag2_transport::Rosbag2Transport::play, rosbag2_transport_ptr, storage_options_, rosbag2_transport::PlayOptions{read_ahead_queue_size, "", rate, {}, {}, loop_playback, {}}); diff --git a/rosbag2_transport/test/rosbag2_transport/test_play_timing.cpp b/rosbag2_transport/test/rosbag2_transport/test_play_timing.cpp index 695830eebf..9b7e801a9e 100644 --- a/rosbag2_transport/test/rosbag2_transport/test_play_timing.cpp +++ b/rosbag2_transport/test/rosbag2_transport/test_play_timing.cpp @@ -60,7 +60,7 @@ TEST_F(Rosbag2TransportTestFixture, playing_respects_relative_timing_of_stored_m // we check that time elapsed during playing is at least the time difference between the two // messages auto start = std::chrono::steady_clock::now(); - Rosbag2Transport rosbag2_transport(reader_, writer_, info_); + Rosbag2Transport rosbag2_transport(reader_, writer_, info_, reindexer_); rosbag2_transport.play(storage_options_, play_options_); auto replay_time = std::chrono::steady_clock::now() - start; @@ -95,7 +95,7 @@ TEST_F(Rosbag2TransportTestFixture, playing_respects_rate) play_options_.rate = 2.0; auto start = std::chrono::steady_clock::now(); - Rosbag2Transport rosbag2_transport(reader_, writer_, info_); + Rosbag2Transport rosbag2_transport(reader_, writer_, info_, reindexer_); rosbag2_transport.play(storage_options_, play_options_); auto replay_time = std::chrono::steady_clock::now() - start; @@ -109,7 +109,7 @@ TEST_F(Rosbag2TransportTestFixture, playing_respects_rate) play_options_.rate = 1.0; start = std::chrono::steady_clock::now(); - rosbag2_transport = Rosbag2Transport(reader_, writer_, info_); + rosbag2_transport = Rosbag2Transport(reader_, writer_, info_, reindexer_); rosbag2_transport.play(storage_options_, play_options_); replay_time = std::chrono::steady_clock::now() - start; @@ -122,7 +122,7 @@ TEST_F(Rosbag2TransportTestFixture, playing_respects_rate) play_options_.rate = 0.5; start = std::chrono::steady_clock::now(); - rosbag2_transport = Rosbag2Transport(reader_, writer_, info_); + rosbag2_transport = Rosbag2Transport(reader_, writer_, info_, reindexer_); rosbag2_transport.play(storage_options_, play_options_); replay_time = std::chrono::steady_clock::now() - start; @@ -135,7 +135,7 @@ TEST_F(Rosbag2TransportTestFixture, playing_respects_rate) play_options_.rate = 0.0; start = std::chrono::steady_clock::now(); - rosbag2_transport = Rosbag2Transport(reader_, writer_, info_); + rosbag2_transport = Rosbag2Transport(reader_, writer_, info_, reindexer_); rosbag2_transport.play(storage_options_, play_options_); replay_time = std::chrono::steady_clock::now() - start; @@ -148,7 +148,7 @@ TEST_F(Rosbag2TransportTestFixture, playing_respects_rate) play_options_.rate = -1.23f; start = std::chrono::steady_clock::now(); - rosbag2_transport = Rosbag2Transport(reader_, writer_, info_); + rosbag2_transport = Rosbag2Transport(reader_, writer_, info_, reindexer_); rosbag2_transport.play(storage_options_, play_options_); replay_time = std::chrono::steady_clock::now() - start; diff --git a/rosbag2_transport/test/rosbag2_transport/test_play_topic_remap.cpp b/rosbag2_transport/test/rosbag2_transport/test_play_topic_remap.cpp index 4db645099a..36d82d999a 100644 --- a/rosbag2_transport/test/rosbag2_transport/test_play_topic_remap.cpp +++ b/rosbag2_transport/test/rosbag2_transport/test_play_topic_remap.cpp @@ -63,7 +63,7 @@ TEST_F(RosBag2PlayTestFixture, recorded_message_is_played_on_remapped_topic) { remapped_topic, 1u); auto await_received_messages = sub_->spin_subscriptions(); - rosbag2_transport::Rosbag2Transport rosbag2_transport(reader_, writer_, info_); + rosbag2_transport::Rosbag2Transport rosbag2_transport(reader_, writer_, info_, reindexer_); rosbag2_transport.play(storage_options_, play_options_);