Vendor SDK adapter example

Wrap a vendor hardware SDK with the ROS 2 state and command interfaces Sentinel expects.

Use this pattern when a vendor SDK owns the hardware but does not expose the ROS 2 interfaces Sentinel needs. Keep the hardware loop in a small ROS 2 node:

Direction	ROS 2 interface	Adapter responsibility
Hardware → Sentinel	`sensor_msgs/msg/JointState`	Read measured state from the SDK and publish it continuously
Sentinel → hardware	`trajectory_msgs/msg/JointTrajectory`	Validate named targets and convert them into the SDK's native command

The implementation below uses an I2RT-controlled arm as a concrete example. Apply the same boundary to another SDK: initialize the hardware once, keep state publication independent of commands, and convert each incoming target into the vendor API.

Test the adapter without Sentinel first. The subscriber commands physical hardware. Add the limits, command watchdog, fault propagation, and emergency-stop integration required by your robot before use.

Adapter node

#!/usr/bin/env python3
import numpy as np
import rclpy
from rclpy.node import Node
from rclpy.qos import qos_profile_sensor_data
from sensor_msgs.msg import JointState
from trajectory_msgs.msg import JointTrajectory

from i2rt.robots.get_robot import get_yam_robot
from i2rt.robots.utils import GripperType


JOINT_NAMES = [f"joint{i}" for i in range(1, 7)]


class I2RTRosAdapter(Node):
    def __init__(self):
        super().__init__("i2rt_ros_adapter")

        self.robot = get_yam_robot(
            channel="can0",
            gripper_type=GripperType.NO_GRIPPER,
            zero_gravity_mode=False,
        )

        self.state_pub = self.create_publisher(
            JointState,
            "/yam/joint_states",
            qos_profile_sensor_data,
        )
        self.command_sub = self.create_subscription(
            JointTrajectory,
            "/yam/joint_trajectory",
            self.on_command,
            10,
        )
        self.state_timer = self.create_timer(1.0 / 120.0, self.publish_state)

    def publish_state(self):
        observation = self.robot.get_observations()

        msg = JointState()
        msg.header.stamp = self.get_clock().now().to_msg()
        msg.name = JOINT_NAMES
        msg.position = observation["joint_pos"][:6].tolist()
        msg.velocity = observation["joint_vel"][:6].tolist()
        msg.effort = observation["joint_eff"][:6].tolist()
        self.state_pub.publish(msg)

    def on_command(self, msg: JointTrajectory):
        if not msg.points:
            return

        positions_by_name = dict(zip(msg.joint_names, msg.points[-1].positions))
        if any(name not in positions_by_name for name in JOINT_NAMES):
            self.get_logger().error("Command does not contain every arm joint")
            return

        target = np.array(
            [positions_by_name[name] for name in JOINT_NAMES],
            dtype=np.float64,
        )
        self.robot.command_joint_pos(target)

    def destroy_node(self):
        self.robot.close()
        super().destroy_node()


def main():
    rclpy.init()
    node = I2RTRosAdapter()
    try:
        rclpy.spin(node)
    finally:
        node.destroy_node()
        rclpy.shutdown()


if __name__ == "__main__":
    main()

Sentinel topic mapping

Map the generic robot adapter to those topics:

adapters:
  - id: arm
    plugin: sentinel_adapter_ros2_bridge::Ros2BridgeAdapter
    description:
      source: file
      path: package://your_robot_description/urdf/yam.urdf
    bridge:
      manipulator:
        enabled: true
        capability_id: arm
        command:
          topic: /yam/joint_trajectory
          qos: reliable
        state:
          topic: /yam/joint_states
          qos: sensor
          stale_timeout_s: 0.5

The current generic adapter and this node must run in the same ROS_DOMAIN_ID.

Validate it

Start the hardware adapter

Keep the robot disarmed or in its validated safe test mode.

Confirm measured state

Run ros2 topic hz /yam/joint_states and verify that all six joint names are present.

Test a named command

Send a small JointTrajectory command in a cleared workspace. Do not publish an unnamed position array.

Test command loss

Stop the publisher and confirm that the hardware follows the robot's validated hold or stop behavior.

Production work still required

The example shows the ROS boundary, not a complete safety controller. A production adapter should add:

Joint-limit and velocity validation before command_joint_pos
A command watchdog
Hardware fault propagation
Gripper mapping when the robot has a gripper
Lifecycle or controller-manager integration when that support is released

Return to the integration checklist

Connect cameras, create the Sentinel configuration, and validate before arming.