Add --content-filter support to ros2 topic echo|hz|bw

Add DDS content filter expression support to `ros2 topic echo`,
`ros2 topic hz`, and `ros2 topic bw` commands via new
`--content-filter` and `--content-filter-params` CLI arguments.

Content filters are applied at the middleware level, so only matching
messages are delivered to the subscriber — reducing bandwidth and CPU
usage compared to client-side Python filtering.

Closes #1126

Signed-off-by: Pavel Guzenfeld <me@pavelguzenfeld.com>
Signed-off-by: Pavel Guzenfeld <pavelguzenfeld@gmail.com>

Author: PavelGuzenfeld

ros2topic/ros2topic/verb/bw.py

@@ -40,6 +40,7 @@
 import rclpy
 
 from rclpy.executors import ExternalShutdownException
+from rclpy.subscription_content_filter_options import ContentFilterOptions
 
 from ros2cli.helpers import interactive_select
 from ros2cli.node.direct import add_arguments as add_direct_node_arguments
@@ -96,6 +97,12 @@ def add_arguments(self, parser, cli_name):
             '--window', '-w', dest='window_size', type=positive_int, default=DEFAULT_WINDOW_SIZE,
             help='maximum window size, in # of messages, for calculating rate '
                  f'(default: {DEFAULT_WINDOW_SIZE})', metavar='WINDOW')
+        parser.add_argument(
+            '--content-filter', dest='content_filter_expr', default=None,
+            help='DDS content filter expression applied at the middleware level')
+        parser.add_argument(
+            '--content-filter-params', dest='content_filter_params', nargs='*', default=[],
+            help='Parameters for the content filter expression')
         add_direct_node_arguments(parser)
 
     def main(self, *, args):
@@ -127,6 +134,12 @@ def main(args):
 
     topics = args.topic_name
 
+    content_filter_options = None
+    if args.content_filter_expr:
+        content_filter_options = ContentFilterOptions(
+            filter_expression=args.content_filter_expr,
+            expression_parameters=args.content_filter_params)
+
     with DirectNode(args) as node:
         # Get all available topics at this moment
         if args.all_topics:
@@ -139,7 +152,8 @@ def main(args):
                 return
             print(f'Subscribing to all {len(topics)} available topics...')
         return _rostopic_bw(node.node, topics, qos_args=args, window_size=args.window_size,
-                            all_topics=args.all_topics)
+                            all_topics=args.all_topics,
+                            content_filter_options=content_filter_options)
 
 
 class ROSTopicBandwidth(object):
@@ -293,7 +307,8 @@ def _get_ascii_table(header, cols):
     return table
 
 
-def _rostopic_bw(node, topics, qos_args, window_size=DEFAULT_WINDOW_SIZE, all_topics=False):
+def _rostopic_bw(node, topics, qos_args, window_size=DEFAULT_WINDOW_SIZE, all_topics=False,
+                 content_filter_options=None):
     """Periodically print the received bandwidth of topics to console until shutdown."""
     # pause bw until topic is published
     rt = ROSTopicBandwidth(node, window_size)
@@ -314,7 +329,8 @@ def _rostopic_bw(node, topics, qos_args, window_size=DEFAULT_WINDOW_SIZE, all_to
             topic,
             functools.partial(rt.callback, topic=topic),
             qos_profile,
-            raw=True
+            raw=True,
+            content_filter_options=content_filter_options
         )
         print('Subscribed to [%s]' % topic)
 

ros2topic/ros2topic/verb/echo.py

@@ -21,6 +21,7 @@
 from rclpy.event_handler import UnsupportedEventTypeError
 from rclpy.node import Node
 from rclpy.qos import QoSProfile
+from rclpy.subscription_content_filter_options import ContentFilterOptions
 from rclpy.task import Future
 
 from ros2cli.helpers import interactive_select
@@ -107,6 +108,15 @@ def add_arguments(self, parser, cli_name):
             '--filter', dest='filter_expr', help='Python expression to filter messages that '
                                                  'are printed. Expression can use Python builtins '
                                                  'as well as m (the message).')
+        parser.add_argument(
+            '--content-filter', dest='content_filter_expr', default=None,
+            help='DDS content filter expression applied at the middleware level '
+                 '(e.g. "data > 100"). Only messages matching the filter are '
+                 'delivered to the subscriber. Use %%0, %%1, ... as parameter '
+                 'placeholders with --content-filter-params.')
+        parser.add_argument(
+            '--content-filter-params', dest='content_filter_params', nargs='*', default=[],
+            help='Parameters for the content filter expression')
         parser.add_argument(
             '--once', action='store_true', help='Print the first message received and then exit.')
         parser.add_argument(
@@ -192,13 +202,20 @@ def main(self, *, args):
             if args.timeout is not None:
                 self.timer = node.create_timer(args.timeout, self._timed_out)
 
+            content_filter_options = None
+            if args.content_filter_expr:
+                content_filter_options = ContentFilterOptions(
+                    filter_expression=args.content_filter_expr,
+                    expression_parameters=args.content_filter_params)
+
             self.subscribe_and_spin(
                 node,
                 args.topic_name,
                 message_type,
                 qos_profile,
                 args.no_lost_messages,
-                args.raw)
+                args.raw,
+                content_filter_options=content_filter_options)
 
     def subscribe_and_spin(
         self,
@@ -208,6 +225,7 @@ def subscribe_and_spin(
         qos_profile: QoSProfile,
         no_report_lost_messages: bool,
         raw: bool,
+        content_filter_options: Optional[ContentFilterOptions] = None,
     ) -> Optional[str]:
         """Initialize a node with a single subscription and spin."""
         event_callbacks = None
@@ -221,7 +239,8 @@ def subscribe_and_spin(
                 self._subscriber_callback,
                 qos_profile,
                 event_callbacks=event_callbacks,
-                raw=raw)
+                raw=raw,
+                content_filter_options=content_filter_options)
         except UnsupportedEventTypeError:
             assert not no_report_lost_messages
             node.create_subscription(
@@ -230,7 +249,8 @@ def subscribe_and_spin(
                 self._subscriber_callback,
                 qos_profile,
                 event_callbacks=None,
-                raw=raw)
+                raw=raw,
+                content_filter_options=content_filter_options)
 
         if self.future is not None:
             rclpy.spin_until_future_complete(node, self.future)

ros2topic/ros2topic/verb/hz.py

@@ -41,6 +41,7 @@
 from rclpy.clock import Clock
 from rclpy.clock import ClockType
 from rclpy.executors import ExternalShutdownException
+from rclpy.subscription_content_filter_options import ContentFilterOptions
 
 from ros2cli.helpers import interactive_select
 from ros2cli.node.direct import add_arguments as add_direct_node_arguments
@@ -95,6 +96,12 @@ def add_arguments(self, parser, cli_name):
             dest='use_wtime', default=False, action='store_true',
             help='calculates rate using wall time which can be helpful'
                  ' when clock is not published during simulation')
+        parser.add_argument(
+            '--content-filter', dest='content_filter_expr', default=None,
+            help='DDS content filter expression applied at the middleware level')
+        parser.add_argument(
+            '--content-filter-params', dest='content_filter_params', nargs='*', default=[],
+            help='Parameters for the content filter expression')
         add_direct_node_arguments(parser)
 
     def main(self, *, args):
@@ -134,6 +141,12 @@ def eval_fn(m):
     else:
         filter_expr = None
 
+    content_filter_options = None
+    if args.content_filter_expr:
+        content_filter_options = ContentFilterOptions(
+            filter_expression=args.content_filter_expr,
+            expression_parameters=args.content_filter_params)
+
     with DirectNode(args) as node:
         # Get all available topics at this moment
         if args.all_topics:
@@ -149,7 +162,8 @@ def eval_fn(m):
         return _rostopic_hz(
             node.node, topics, qos_args=args, window_size=args.window_size,
             filter_expr=filter_expr, use_wtime=args.use_wtime,
-            all_topics=args.all_topics)
+            all_topics=args.all_topics,
+            content_filter_options=content_filter_options)
 
 
 class ROSTopicHz(object):
@@ -337,7 +351,7 @@ def _get_ascii_table(header, cols):
 
 
 def _rostopic_hz(node, topics, qos_args, window_size=DEFAULT_WINDOW_SIZE, filter_expr=None,
-                 use_wtime=False, all_topics=False):
+                 use_wtime=False, all_topics=False, content_filter_options=None):
     """
     Periodically print the publishing rate of a topic to console until shutdown.
 
@@ -346,6 +360,7 @@ def _rostopic_hz(node, topics, qos_args, window_size=DEFAULT_WINDOW_SIZE, filter
     :param window_size: number of messages to average over, -1 for infinite, ``int``
     :param filter_expr: Python filter expression that is called with m, the message instance
     :param all_topics: whether all topics are being monitored, ``bool``
+    :param content_filter_options: DDS content filter options, ``ContentFilterOptions``
     """
     # pause hz until topic is published
     rt = ROSTopicHz(node, window_size, filter_expr=filter_expr, use_wtime=use_wtime)
@@ -367,7 +382,8 @@ def _rostopic_hz(node, topics, qos_args, window_size=DEFAULT_WINDOW_SIZE, filter
             topic,
             functools.partial(rt.callback_hz, topic=topic),
             qos_profile,
-            raw=filter_expr is None)
+            raw=filter_expr is None,
+            content_filter_options=content_filter_options)
         if topics_len > 1:
             print('Subscribed to [%s]' % topic)