feat: allow selecting topics by message type in ring buffer config

- topic_filter.py: extend find_matching_config with optional type_name
  parameter; entries with a 'type' key are matched by exact message type,
  entries with a 'pattern' key by name glob, both keys AND-ed together
- ringbuffer_node.py: pass type_names[0] to find_matching_config so
  type-based config entries are evaluated during topic discovery
- example_config.yaml: add geometry_msgs/msg/Twist type-selector example
- test_topic_filter.py: 11 new tests covering type-only, combined, and
  edge-case matching scenarios

Co-authored-by: marc-hanheide <1153084+marc-hanheide@users.noreply.github.com>

Author: Copilot

src/rosbag_blackbox/config/example_config.yaml

@@ -7,6 +7,11 @@
 # Pattern syntax (zenoh-style):
 #   *   matches exactly ONE path segment  (e.g. /camera/* → /camera/left)
 #   **  matches ANY depth of namespace   (e.g. /camera/** → /camera/left/image_raw)
+#
+# Topic selectors (at least one must be provided per entry):
+#   pattern: <glob>   – match by topic NAME using the syntax above
+#   type:    <string> – match by message TYPE (exact, e.g. geometry_msgs/msg/Twist)
+# Both keys may be combined; in that case BOTH conditions must hold.
 
 # ── Ring-buffer settings ────────────────────────────────────────────────────
 buffer:
@@ -32,6 +37,14 @@ topics:
       durability: volatile
       depth: 1
 
+  # ── All Twist messages, whatever topic they appear on ──────────────────
+  - type: "geometry_msgs/msg/Twist"
+    max_frequency: 10.0
+    qos:
+      reliability: best_effort
+      durability: volatile
+      depth: 10
+
   # ── TF / TF-static (reliable, higher depth) ────────────────────────────
   - pattern: "/tf"
     qos:

src/rosbag_blackbox/rosbag_blackbox/ringbuffer_node.py

@@ -243,7 +243,9 @@ def _discover_topics(self) -> None:
             # If topic configs are defined, only subscribe to matches;
             # otherwise record everything.
             if self._topic_cfgs:
-                topic_cfg = find_matching_config(topic_name, self._topic_cfgs)
+                topic_cfg = find_matching_config(
+                    topic_name, self._topic_cfgs, type_names[0]
+                )
                 if topic_cfg is None:
                     continue
             else:

src/rosbag_blackbox/rosbag_blackbox/topic_filter.py

@@ -24,6 +24,14 @@
     matches_pattern('/camera/left/image', '/camera/*')  -> False
     matches_pattern('/camera/left/image', '/camera/**') -> True
     matches_pattern('/tf', '/**')                       -> True
+
+Topic config entries support two selectors that can be used independently or in combination:
+    ``pattern`` – Zenoh-style glob matched against the topic *name*.
+    ``type``    – Exact message type string matched against the topic *type*
+                  (e.g. ``'geometry_msgs/msg/Twist'``).
+
+At least one of ``pattern`` or ``type`` must be present in each entry.
+When both are present both conditions must hold simultaneously.
 """
 
 from typing import List, Optional
@@ -79,16 +87,45 @@ def matches_pattern(topic: str, pattern: str) -> bool:
     return _match_parts(topic_parts, pattern_parts)
 
 
-def find_matching_config(topic: str, topic_configs: list) -> Optional[dict]:
-    """Return the first config entry whose ``pattern`` matches *topic*.
+def find_matching_config(
+    topic: str,
+    topic_configs: list,
+    type_name: str = '',
+) -> Optional[dict]:
+    """Return the first config entry whose selector matches *topic*.
+
+    Entries are evaluated in list order; the first match wins.
+
+    Each config entry must supply **at least one** of:
+
+    * ``pattern`` – Zenoh-style glob matched against the topic *name*.
+    * ``type``    – Exact message type string matched against *type_name*
+      (e.g. ``'geometry_msgs/msg/Twist'``).
 
-    Patterns are evaluated in list order; the first match wins.
+    When both keys are present **both** conditions must hold simultaneously.
+    Entries that contain neither key are skipped.
 
-    :param topic:         ROS2 topic name.
-    :param topic_configs: List of dicts, each containing a ``'pattern'`` key.
+    :param topic:         ROS2 topic name (e.g. ``'/cmd_vel'``).
+    :param topic_configs: List of dicts, each containing ``'pattern'``
+                          and/or ``'type'`` keys.
+    :param type_name:     ROS2 message type of the topic
+                          (e.g. ``'geometry_msgs/msg/Twist'``).
+                          When empty, entries that require type matching
+                          are skipped.
     :returns: Matched config dict, or ``None`` if nothing matches.
     """
     for cfg in topic_configs:
-        if matches_pattern(topic, cfg.get('pattern', '')):
-            return cfg
+        has_pattern = 'pattern' in cfg
+        has_type = 'type' in cfg
+
+        if not has_pattern and not has_type:
+            continue
+
+        if has_pattern and not matches_pattern(topic, cfg['pattern']):
+            continue
+
+        if has_type and (not type_name or type_name != cfg['type']):
+            continue
+
+        return cfg
     return None

src/rosbag_blackbox/test/test_topic_filter.py

@@ -103,3 +103,74 @@ def test_no_match(self):
 
     def test_empty_configs(self):
         assert find_matching_config('/tf', []) is None
+
+
+# ---------------------------------------------------------------------------
+# find_matching_config – type-based and combined selectors
+# ---------------------------------------------------------------------------
+
+class TestFindMatchingConfigByType:
+
+    _CONFIGS = [
+        {'type': 'geometry_msgs/msg/Twist'},
+        {'pattern': '/tf', 'qos': {'depth': 100}},
+        {'type': 'std_msgs/msg/String', 'pattern': '/chatter'},
+    ]
+
+    def test_type_only_match(self):
+        # Any topic that publishes Twist should match the first entry.
+        cfg = find_matching_config('/cmd_vel', self._CONFIGS, 'geometry_msgs/msg/Twist')
+        assert cfg is not None
+        assert cfg['type'] == 'geometry_msgs/msg/Twist'
+
+    def test_type_only_different_topic_name(self):
+        # A different topic name publishing Twist also matches.
+        cfg = find_matching_config('/robot/cmd_vel', self._CONFIGS, 'geometry_msgs/msg/Twist')
+        assert cfg is not None
+        assert cfg['type'] == 'geometry_msgs/msg/Twist'
+
+    def test_type_only_wrong_type(self):
+        # The Twist-only entry should NOT match a String topic.
+        # The pattern-only /tf entry should NOT match /cmd_vel.
+        # The combined entry requires /chatter, so /cmd_vel also misses.
+        cfg = find_matching_config('/cmd_vel', self._CONFIGS, 'std_msgs/msg/String')
+        assert cfg is None
+
+    def test_type_no_type_name_provided(self):
+        # When type_name is not provided, type-only configs are skipped.
+        cfg = find_matching_config('/cmd_vel', self._CONFIGS)
+        assert cfg is None
+
+    def test_pattern_config_unaffected_by_type(self):
+        # Pattern-only configs still match regardless of type_name.
+        cfg = find_matching_config('/tf', self._CONFIGS, 'tf2_msgs/msg/TFMessage')
+        assert cfg is not None
+        assert cfg['qos']['depth'] == 100
+
+    def test_pattern_config_with_no_type_arg(self):
+        # Pattern-only configs still work when type_name is omitted.
+        cfg = find_matching_config('/tf', self._CONFIGS)
+        assert cfg is not None
+        assert cfg['qos']['depth'] == 100
+
+    def test_combined_both_match(self):
+        # Entry requires BOTH /chatter pattern AND String type.
+        cfg = find_matching_config('/chatter', self._CONFIGS, 'std_msgs/msg/String')
+        assert cfg is not None
+        assert cfg['type'] == 'std_msgs/msg/String'
+
+    def test_combined_pattern_no_match(self):
+        # Type matches but topic name doesn't match the pattern.
+        cfg = find_matching_config('/wrong_topic', self._CONFIGS, 'std_msgs/msg/String')
+        assert cfg is None
+
+    def test_combined_type_no_match(self):
+        # Pattern matches (/chatter) but type is wrong.
+        cfg = find_matching_config('/chatter', self._CONFIGS, 'geometry_msgs/msg/Pose')
+        assert cfg is None
+
+    def test_entry_without_either_key_is_skipped(self):
+        configs = [{'max_frequency': 5.0}, {'pattern': '/tf'}]
+        cfg = find_matching_config('/tf', configs)
+        assert cfg is not None
+        assert 'pattern' in cfg