Exclude topics

ros2bag/ros2bag/verb/play.py

@@ -33,6 +33,9 @@ def add_arguments(self, parser, cli_name):  # noqa: D102
             help='size of message queue rosbag tries to hold in memory to help deterministic '
                  'playback. Larger size will result in larger memory needs but might prevent '
                  'delay of message playback.')
+        parser.add_argument(
+            '-e', '--exclude-topics', nargs='*', default=[], help='topics which are not played from the bag file')
+
 
     def main(self, *, args):  # noqa: D102
         bag_file = args.bag_file
@@ -43,4 +46,5 @@ def main(self, *, args):  # noqa: D102
             uri=bag_file,
             storage_id=args.storage,
             node_prefix=NODE_NAME_PREFIX,
-            read_ahead_queue_size=args.read_ahead_queue_size)
+            read_ahead_queue_size=args.read_ahead_queue_size,
+            exclude_topics=args.exclude_topics)

rosbag2_transport/include/rosbag2_transport/play_options.hpp

@@ -24,8 +24,11 @@ namespace rosbag2_transport
 struct PlayOptions
 {
 public:
+  using TopicFilter = std::function<bool(const std::string& topic)>;
+
   size_t read_ahead_queue_size;
   std::string node_prefix = "";
+  TopicFilter topic_filter = [](const std::string&){return false;};
 };
 
 }  // namespace rosbag2_transport

rosbag2_transport/src/rosbag2_transport/player.cpp

@@ -54,14 +54,14 @@ bool Player::is_storage_completely_loaded() const
 
 void Player::play(const PlayOptions & options)
 {
-  prepare_publishers();
+  prepare_publishers(options);
 
   storage_loading_future_ = std::async(std::launch::async,
       [this, options]() {load_storage_content(options);});
 
   wait_for_filled_queue(options);
 
-  play_messages_from_queue();
+  play_messages_from_queue(options);
 }
 
 void Player::wait_for_filled_queue(const PlayOptions & options) const
@@ -114,36 +114,43 @@ void Player::enqueue_up_to_boundary(const TimePoint & time_first_message, uint64
   }
 }
 
-void Player::play_messages_from_queue()
+void Player::play_messages_from_queue(const PlayOptions & options)
 {
   start_time_ = std::chrono::system_clock::now();
   do {
-    play_messages_until_queue_empty();
+    play_messages_until_queue_empty(options);
     if (!is_storage_completely_loaded() && rclcpp::ok()) {
       ROSBAG2_TRANSPORT_LOG_WARN("Message queue starved. Messages will be delayed. Consider "
         "increasing the --read-ahead-queue-size option.");
     }
   } while (!is_storage_completely_loaded() && rclcpp::ok());
 }
 
-void Player::play_messages_until_queue_empty()
+void Player::play_messages_until_queue_empty(const PlayOptions & options)
 {
   ReplayableMessage message;
   while (message_queue_.try_dequeue(message) && rclcpp::ok()) {
+    if (options.topic_filter(message.message->topic_name))
+      continue;
     std::this_thread::sleep_until(start_time_ + message.time_since_start);
     if (rclcpp::ok()) {
       publishers_[message.message->topic_name]->publish(message.message->serialized_data);
     }
   }
 }
 
-void Player::prepare_publishers()
+void Player::prepare_publishers(const PlayOptions & options)
 {
   auto topics = reader_->get_all_topics_and_types();
+
   for (const auto & topic : topics) {
+    if (options.topic_filter(topic.name))
+    {
+      ROSBAG2_TRANSPORT_LOG_INFO("Excluding topic %s", topic.name.c_str());
+      continue;
+    }
     publishers_.insert(std::make_pair(
         topic.name, rosbag2_transport_->create_generic_publisher(topic.name, topic.type)));
   }
 }
-
 }  // namespace rosbag2_transport

rosbag2_transport/src/rosbag2_transport/player.hpp

@@ -54,9 +54,9 @@ class Player
   bool is_storage_completely_loaded() const;
   void enqueue_up_to_boundary(const TimePoint & time_first_message, uint64_t boundary);
   void wait_for_filled_queue(const PlayOptions & options) const;
-  void play_messages_from_queue();
-  void play_messages_until_queue_empty();
-  void prepare_publishers();
+  void play_messages_from_queue(const PlayOptions & options);
+  void play_messages_until_queue_empty(const PlayOptions & options);
+  void prepare_publishers(const PlayOptions & options);
 
   static constexpr double read_ahead_lower_bound_percentage_ = 0.9;
   static const std::chrono::milliseconds queue_read_wait_period_;

rosbag2_transport/src/rosbag2_transport/rosbag2_transport_python.cpp

@@ -16,6 +16,7 @@
 #include <chrono>
 #include <string>
 #include <vector>
+#include <set>
 
 #include "rosbag2_transport/rosbag2_transport.hpp"
 #include "rosbag2_transport/record_options.hpp"
@@ -102,18 +103,21 @@ rosbag2_transport_play(PyObject * Py_UNUSED(self), PyObject * args, PyObject * k
     "storage_id",
     "node_prefix",
     "read_ahead_queue_size",
+    "exclude_topics",
     nullptr
   };
 
   char * uri;
   char * storage_id;
   char * node_prefix;
+  PyObject * exclude_topics = nullptr;
   size_t read_ahead_queue_size;
-  if (!PyArg_ParseTupleAndKeywords(args, kwargs, "sss|k", const_cast<char **>(kwlist),
+  if (!PyArg_ParseTupleAndKeywords(args, kwargs, "sss|kO", const_cast<char **>(kwlist),
     &uri,
     &storage_id,
     &node_prefix,
-    &read_ahead_queue_size))
+    &read_ahead_queue_size,
+    &exclude_topics))
   {
     return nullptr;
   }
@@ -124,6 +128,32 @@ rosbag2_transport_play(PyObject * Py_UNUSED(self), PyObject * args, PyObject * k
   play_options.node_prefix = std::string(node_prefix);
   play_options.read_ahead_queue_size = read_ahead_queue_size;
 
+  std::set<std::string> exclude_topics_list;
+  if (exclude_topics) {
+    PyObject * topic_exclude_iterator = PyObject_GetIter(exclude_topics);
+    if (topic_exclude_iterator != nullptr) {
+      PyObject * topic;
+      while ((topic = PyIter_Next(topic_exclude_iterator))) {
+        exclude_topics_list.insert(PyUnicode_AsUTF8(topic));
+
+        Py_DECREF(topic);
+      }
+      Py_DECREF(topic_exclude_iterator);
+    }
+
+    if (exclude_topics_list.empty() == false) {
+      auto topic_filter_function = [exclude_topics_list](const std::string& topic)
+      {
+        auto entry = exclude_topics_list.find(topic);
+        if (entry == exclude_topics_list.end())
+          return false;
+        return true;
+      };
+
+      play_options.topic_filter = topic_filter_function;
+    }
+  }
+
   rosbag2_transport::Rosbag2Transport transport;
   transport.init();
   transport.play(storage_options, play_options);