16 add message syncronisation

README.md

@@ -641,6 +641,82 @@ subscribers:
 
 QoS is required for all subscribers. See [QoS Configuration](#qos-configuration) for details. Topic names support `${param:name}` substitution — see [Dynamic Topic/Service/Action Names](#dynamic-topicserviceaction-names).
 
+### Synchronised Subscribers (Sync Groups)
+
+When multiple topics need to be processed together based on matching timestamps, use a **sync group**. Jig wraps `message_filters` to pair messages by time and deliver them through a single callback.
+
+```yaml
+subscribers:
+    # Regular subscriber — works as normal
+    - topic: odom
+      type: nav_msgs/msg/Odometry
+      qos:
+        history: 1
+        reliability: RELIABLE
+
+    # Sync group — delivers paired messages via a single callback
+    - name: dual_fix
+      policy: approximate       # "approximate" or "exact"
+      queue_size: 10
+      max_interval: 0.05        # seconds (required for approximate, forbidden for exact)
+      topics:
+        - topic: fix_left
+          type: sensor_msgs/msg/NavSatFix
+          qos:
+            history: 10
+            reliability: BEST_EFFORT
+        - topic: fix_right
+          type: sensor_msgs/msg/NavSatFix
+          qos:
+            history: 10
+            reliability: BEST_EFFORT
+```
+
+Sync groups live in the `subscribers` list alongside regular subscribers. They are distinguished by having `name`, `policy`, and `topics` fields instead of `topic` and `type`. Both appear under `sn.subscribers.*` in generated code.
+
+**Fields:**
+
+| Field | Required | Description |
+|-------|----------|-------------|
+| `name` | Yes | Field name in generated code (valid C++ identifier) |
+| `policy` | Yes | `approximate` (time-based matching) or `exact` (exact timestamp match) |
+| `queue_size` | Yes | Per-topic message queue depth |
+| `max_interval` | Conditional | Max time difference in seconds. Required for `approximate`, forbidden for `exact` |
+| `topics` | Yes | 2-9 topics to synchronise, each with `topic`, `type`, and `qos` |
+
+**Constraints:**
+- A sync group must contain between 2 and 9 topics (the `message_filters` template arity limit)
+- `${for_each_param:...}` is not supported in sync group topic names (subscriber count must be known at compile time)
+- `${param:...}` substitution works in both topic names and QoS fields
+- Sync group names must not collide with regular subscriber field names
+
+#### C++ Usage
+
+```cpp
+CallbackReturn on_configure(std::shared_ptr<Session> sn) {
+    sn->subscribers.dual_fix->set_callback(
+        [](auto sn, auto left_msg, auto right_msg) {
+            // Both messages are from the same epoch
+            // Compute heading, odometry, etc.
+        });
+    return CallbackReturn::SUCCESS;
+}
+```
+
+#### Python Usage
+
+```python
+def on_configure(sn: Session) -> TransitionCallbackReturn:
+    sn.subscribers.dual_fix.set_callback(on_dual_fix)
+    return TransitionCallbackReturn.SUCCESS
+
+def on_dual_fix(sn, left_msg, right_msg):
+    # Both messages are from the same epoch
+    pass
+```
+
+The sync callback follows the same lifecycle guard as regular subscribers — it only fires when the node is in the ACTIVE state.
+
 ### Services
 
 ```yaml

jig/include/jig/sync_group.hpp

@@ -0,0 +1,142 @@
+#pragma once
+
+#include <array>
+#include <functional>
+#include <memory>
+#include <string>
+#include <tuple>
+#include <type_traits>
+#include <utility>
+
+#include <lifecycle_msgs/msg/state.hpp>
+#include <rclcpp/rclcpp.hpp>
+#include <rclcpp_lifecycle/lifecycle_node.hpp>
+
+// Kilted+ moved to .hpp headers and deprecated the old .h shims.
+// Use __has_include so we keep building on Humble/Jazzy where only .h exists.
+#if __has_include(<message_filters/sync_policies/approximate_time.hpp>)
+#include <message_filters/subscriber.hpp>
+#include <message_filters/sync_policies/approximate_time.hpp>
+#include <message_filters/sync_policies/exact_time.hpp>
+#include <message_filters/synchronizer.hpp>
+#else
+#include <message_filters/subscriber.h>
+#include <message_filters/sync_policies/approximate_time.h>
+#include <message_filters/sync_policies/exact_time.h>
+#include <message_filters/synchronizer.h>
+#endif
+
+#include "session.hpp"
+
+namespace jig {
+
+namespace detail {
+// Custom placeholder type for variadic std::bind expansion.
+// message_filters::registerCallback needs std::bind with explicit placeholders
+// to deduce callback arity — std::function alone is not sufficient.
+template <int N> struct Placeholder {};
+} // namespace detail
+
+} // namespace jig
+
+// Enable our custom placeholder for std::bind
+namespace std {
+template <int N> struct is_placeholder<jig::detail::Placeholder<N>> : std::integral_constant<int, N> {};
+} // namespace std
+
+namespace jig {
+
+struct SyncTopicConfig {
+    std::string name;
+    rclcpp::QoS qos;
+    SyncTopicConfig(std::string name, rclcpp::QoS qos) : name(std::move(name)), qos(std::move(qos)) {}
+};
+
+template <typename SessionType, typename PolicyT, typename... MessageTs> class SyncGroup {
+    static_assert(std::is_base_of_v<Session, SessionType>, "SessionType must derive from jig::Session");
+    static constexpr size_t NumTopics = sizeof...(MessageTs);
+
+  public:
+    using Callback = std::function<void(std::shared_ptr<SessionType>, typename MessageTs::ConstSharedPtr...)>;
+
+    explicit SyncGroup(
+        std::shared_ptr<SessionType> sn, size_t queue_size, std::array<SyncTopicConfig, NumTopics> configs
+    )
+        : session_(sn) {
+        subscribe_all(&sn->node, configs, std::index_sequence_for<MessageTs...>{});
+        create_sync(queue_size, std::index_sequence_for<MessageTs...>{});
+        register_callback(std::index_sequence_for<MessageTs...>{});
+    }
+
+    void set_max_interval(double seconds) {
+        if (sync_) {
+            sync_->setMaxIntervalDuration(rclcpp::Duration::from_seconds(seconds));
+        }
+    }
+
+    void set_callback(Callback cb) { callback_ = std::move(cb); }
+
+  private:
+    std::weak_ptr<SessionType> session_;
+    std::tuple<message_filters::Subscriber<MessageTs, rclcpp_lifecycle::LifecycleNode>...> subscribers_;
+    std::shared_ptr<message_filters::Synchronizer<PolicyT>> sync_;
+    Callback callback_;
+
+    template <std::size_t... Is>
+    void
+    subscribe_all(rclcpp_lifecycle::LifecycleNode *node, const std::array<SyncTopicConfig, NumTopics> &configs, std::index_sequence<Is...>) {
+        (std::get<Is>(subscribers_).subscribe(node, configs[Is].name, configs[Is].qos.get_rmw_qos_profile()), ...);
+    }
+
+    template <std::size_t... Is> void create_sync(size_t queue_size, std::index_sequence<Is...>) {
+        sync_ = std::make_shared<message_filters::Synchronizer<PolicyT>>(
+            PolicyT(queue_size), std::get<Is>(subscribers_)...
+        );
+    }
+
+    template <std::size_t... Is> void register_callback(std::index_sequence<Is...>) {
+        sync_->registerCallback(std::bind(
+            [this](const typename MessageTs::ConstSharedPtr &...msgs) {
+                auto sn = session_.lock();
+                if (!sn)
+                    return;
+                if (sn->node.get_current_state().id() != lifecycle_msgs::msg::State::PRIMARY_STATE_ACTIVE)
+                    return;
+                if (callback_) {
+                    callback_(sn, msgs...);
+                }
+            },
+            detail::Placeholder<static_cast<int>(Is) + 1>{}...
+        ));
+    }
+};
+
+// Convenience type aliases
+template <typename SessionType, typename... MessageTs>
+using ApproximateSync =
+    SyncGroup<SessionType, message_filters::sync_policies::ApproximateTime<MessageTs...>, MessageTs...>;
+
+template <typename SessionType, typename... MessageTs>
+using ExactSync = SyncGroup<SessionType, message_filters::sync_policies::ExactTime<MessageTs...>, MessageTs...>;
+
+// Factory functions matching the jig::create_subscriber / jig::create_publisher pattern
+template <typename SessionType, typename... MessageTs>
+std::shared_ptr<ApproximateSync<SessionType, MessageTs...>> create_approximate_sync_group(
+    std::shared_ptr<SessionType> sn,
+    size_t queue_size,
+    double max_interval,
+    std::array<SyncTopicConfig, sizeof...(MessageTs)> configs
+) {
+    auto group = std::make_shared<ApproximateSync<SessionType, MessageTs...>>(sn, queue_size, std::move(configs));
+    group->set_max_interval(max_interval);
+    return group;
+}
+
+template <typename SessionType, typename... MessageTs>
+std::shared_ptr<ExactSync<SessionType, MessageTs...>> create_exact_sync_group(
+    std::shared_ptr<SessionType> sn, size_t queue_size, std::array<SyncTopicConfig, sizeof...(MessageTs)> configs
+) {
+    return std::make_shared<ExactSync<SessionType, MessageTs...>>(sn, queue_size, std::move(configs));
+}
+
+} // namespace jig

jig/jig/__init__.py

@@ -5,5 +5,6 @@
 from .service import Service
 from .session import Session
 from .subscriber import Subscriber
+from .sync_group import SyncGroup2, SyncGroup3, SyncGroup4, SyncGroup5, SyncGroup6, SyncGroup7, SyncGroup8, SyncGroup9
 from .timer import create_timer, create_wall_timer
 from .transition import TransitionCallbackReturn

jig/jig/sync_group.py

@@ -0,0 +1,155 @@
+import message_filters
+
+from lifecycle_msgs.msg import State
+
+from .session import Session
+
+from typing import Any, Callable, Generic, Optional, TypeVar
+
+SessionT = TypeVar("SessionT", bound=Session)
+
+Msg1T = TypeVar("Msg1T")
+Msg2T = TypeVar("Msg2T")
+Msg3T = TypeVar("Msg3T")
+Msg4T = TypeVar("Msg4T")
+Msg5T = TypeVar("Msg5T")
+Msg6T = TypeVar("Msg6T")
+Msg7T = TypeVar("Msg7T")
+Msg8T = TypeVar("Msg8T")
+Msg9T = TypeVar("Msg9T")
+
+
+class _SyncGroupBase(Generic[SessionT]):
+    """Base implementation for time-synchronised subscriber groups.
+
+    Owns and creates message_filters.Subscriber instances internally.
+    Provides lifecycle-guarded callback dispatch matching the jig entity pattern.
+    """
+
+    _subscribers: list[message_filters.Subscriber]
+    _sync: Any  # TimeSynchronizer or ApproximateTimeSynchronizer
+    _callback: Optional[Callable] = None
+
+    def _initialise(
+        self,
+        session: SessionT,
+        sync_class: type,
+        queue_size: int,
+        topics: list[tuple[type, str, Any]],
+        *,
+        slop: float | None = None,
+    ) -> None:
+        """Initialise the sync group.
+
+        Args:
+            session: The session instance (provides node reference).
+            sync_class: message_filters.TimeSynchronizer or ApproximateTimeSynchronizer.
+            queue_size: Size of the message queue per topic.
+            topics: List of (msg_class, topic_name, qos_profile) tuples.
+            slop: Maximum time difference in seconds (ApproximateTimeSynchronizer only).
+        """
+        self._subscribers = []
+        for msg_class, topic_name, qos in topics:
+            sub = message_filters.Subscriber(session.node, msg_class, topic_name, qos_profile=qos)
+            self._subscribers.append(sub)
+
+        sync_kwargs = {"queue_size": queue_size}
+        if slop is not None:
+            sync_kwargs["slop"] = slop
+        self._sync = sync_class(self._subscribers, **sync_kwargs)
+
+        def _guarded_callback(*msgs, _sn=session):
+            if _sn.node.current_state != State.PRIMARY_STATE_ACTIVE:
+                return
+            if self._callback:
+                self._callback(_sn, *msgs)
+
+        self._sync.registerCallback(_guarded_callback)
+
+    def _destroy(self, node) -> None:
+        for sub in self._subscribers:
+            sub.unregister()
+        self._subscribers = []
+        self._sync = None
+
+    def set_callback(self, callback: Callable) -> None:
+        self._callback = callback
+
+
+class SyncGroup2(Generic[SessionT, Msg1T, Msg2T], _SyncGroupBase[SessionT]):
+    _callback: Optional[Callable[[SessionT, Msg1T, Msg2T], None]] = None
+
+    def set_callback(self, callback: Callable[[SessionT, Msg1T, Msg2T], None]) -> None:
+        super().set_callback(callback)
+
+
+class SyncGroup3(Generic[SessionT, Msg1T, Msg2T, Msg3T], _SyncGroupBase[SessionT]):
+    _callback: Optional[Callable[[SessionT, Msg1T, Msg2T, Msg3T], None]] = None
+
+    def set_callback(self, callback: Callable[[SessionT, Msg1T, Msg2T, Msg3T], None]) -> None:
+        super().set_callback(callback)
+
+
+class SyncGroup4(Generic[SessionT, Msg1T, Msg2T, Msg3T, Msg4T], _SyncGroupBase[SessionT]):
+    _callback: Optional[Callable[[SessionT, Msg1T, Msg2T, Msg3T, Msg4T], None]] = None
+
+    def set_callback(self, callback: Callable[[SessionT, Msg1T, Msg2T, Msg3T, Msg4T], None]) -> None:
+        super().set_callback(callback)
+
+
+class SyncGroup5(Generic[SessionT, Msg1T, Msg2T, Msg3T, Msg4T, Msg5T], _SyncGroupBase[SessionT]):
+    _callback: Optional[Callable[[SessionT, Msg1T, Msg2T, Msg3T, Msg4T, Msg5T], None]] = None
+
+    def set_callback(self, callback: Callable[[SessionT, Msg1T, Msg2T, Msg3T, Msg4T, Msg5T], None]) -> None:
+        super().set_callback(callback)
+
+
+class SyncGroup6(Generic[SessionT, Msg1T, Msg2T, Msg3T, Msg4T, Msg5T, Msg6T], _SyncGroupBase[SessionT]):
+    _callback: Optional[Callable[[SessionT, Msg1T, Msg2T, Msg3T, Msg4T, Msg5T, Msg6T], None]] = None
+
+    def set_callback(self, callback: Callable[[SessionT, Msg1T, Msg2T, Msg3T, Msg4T, Msg5T, Msg6T], None]) -> None:
+        super().set_callback(callback)
+
+
+class SyncGroup7(Generic[SessionT, Msg1T, Msg2T, Msg3T, Msg4T, Msg5T, Msg6T, Msg7T], _SyncGroupBase[SessionT]):
+    _callback: Optional[Callable[[SessionT, Msg1T, Msg2T, Msg3T, Msg4T, Msg5T, Msg6T, Msg7T], None]] = None
+
+    def set_callback(
+        self, callback: Callable[[SessionT, Msg1T, Msg2T, Msg3T, Msg4T, Msg5T, Msg6T, Msg7T], None]
+    ) -> None:
+        super().set_callback(callback)
+
+
+class SyncGroup8(Generic[SessionT, Msg1T, Msg2T, Msg3T, Msg4T, Msg5T, Msg6T, Msg7T, Msg8T], _SyncGroupBase[SessionT]):
+    _callback: Optional[Callable[[SessionT, Msg1T, Msg2T, Msg3T, Msg4T, Msg5T, Msg6T, Msg7T, Msg8T], None]] = None
+
+    def set_callback(
+        self, callback: Callable[[SessionT, Msg1T, Msg2T, Msg3T, Msg4T, Msg5T, Msg6T, Msg7T, Msg8T], None]
+    ) -> None:
+        super().set_callback(callback)
+
+
+class SyncGroup9(
+    Generic[SessionT, Msg1T, Msg2T, Msg3T, Msg4T, Msg5T, Msg6T, Msg7T, Msg8T, Msg9T], _SyncGroupBase[SessionT]
+):
+    _callback: Optional[Callable[[SessionT, Msg1T, Msg2T, Msg3T, Msg4T, Msg5T, Msg6T, Msg7T, Msg8T, Msg9T], None]] = (
+        None
+    )
+
+    def set_callback(
+        self,
+        callback: Callable[[SessionT, Msg1T, Msg2T, Msg3T, Msg4T, Msg5T, Msg6T, Msg7T, Msg8T, Msg9T], None],
+    ) -> None:
+        super().set_callback(callback)
+
+
+__all__ = [
+    "SyncGroup2",
+    "SyncGroup3",
+    "SyncGroup4",
+    "SyncGroup5",
+    "SyncGroup6",
+    "SyncGroup7",
+    "SyncGroup8",
+    "SyncGroup9",
+]

jig/package.xml

@@ -27,6 +27,7 @@
   <depend>python3-jinja2</depend>
   <depend>python3-jsonschema</depend>
   <depend>tf2_ros</depend>
+  <depend>message_filters</depend>
 
   <exec_depend>ament_index_python</exec_depend>