Skip to main content

ROS2 Integration

Bot Pulse works alongside ROS2. This tutorial shows how to instrument a ROS2 node with traces, logs, and metrics using the Bot Pulse Python API.

Setup​

Install Bot Pulse alongside your ROS2 workspace:

pip install botpulse

No ROS2-specific wrapper is needed — Bot Pulse is a pure Python library that works in any process.

Example: Instrumented navigation node​

Here's a ROS2 node that uses Bot Pulse for full observability:

navigation_node.py
import rclpy
from rclpy.node import Node
from geometry_msgs.msg import PoseStamped
import botpulse
import time


class NavigationNode(Node):
    def __init__(self):
        super().__init__("navigation_node")
        self.pulse = botpulse.init(
            service_name="navigation",
            robot_id="robot-01",
        )
        self.waypoint_sub = self.create_subscription(
            PoseStamped, "/goal_pose", self.on_waypoint, 10
        )
        self.get_logger().info("Navigation node started")

    def on_waypoint(self, msg: PoseStamped):
        x = msg.pose.position.x
        y = msg.pose.position.y

        with self.pulse.trace("navigate_to_waypoint") as span:
            span.set_attribute("waypoint.x", x)
            span.set_attribute("waypoint.y", y)

            self.pulse.log(f"Received waypoint ({x}, {y})")

            with self.pulse.child_span("plan_path") as plan_span:
                path = self.plan_path(x, y)
                plan_span.set_attribute("path_length_m", path.length)

            with self.pulse.child_span("follow_path") as follow_span:
                self.execute_path(path)
                follow_span.set_attribute("completed", True)

            self.pulse.metric("waypoint_latency_ms", span.duration_ms)
            self.pulse.log("Waypoint reached")

    def plan_path(self, x, y):
        # Your path planning logic
        time.sleep(0.02)
        return Path(length=((x ** 2 + y ** 2) ** 0.5))

    def execute_path(self, path):
        # Your trajectory execution logic
        time.sleep(0.1)


def main():
    rclpy.init()
    node = NavigationNode()
    rclpy.spin(node)
    rclpy.shutdown()


if __name__ == "__main__":
    main()

Example: Battery monitor node​

battery_node.py
import rclpy
from rclpy.node import Node
from sensor_msgs.msg import BatteryState
import botpulse


class BatteryNode(Node):
    def __init__(self):
        super().__init__("battery_node")
        self.pulse = botpulse.init(
            service_name="battery-monitor",
            robot_id="robot-01",
        )
        self.battery_pub = self.create_publisher(BatteryState, "/battery", 10)
        self.timer = self.create_timer(1.0, self.publish_battery)
        self.cycle_count = 0

    def publish_battery(self):
        with self.pulse.trace("read_battery") as span:
            voltage = self.read_voltage()
            current = self.read_current()
            soc = self.compute_soc(voltage)

            self.pulse.gauge("battery_voltage", value=voltage)
            self.pulse.gauge("battery_soc_percent", value=soc)
            self.pulse.counter("battery_read_cycles", value=1)

            span.set_attribute("voltage", voltage)
            span.set_attribute("soc_percent", soc)

            if soc < 20.0:
                self.pulse.log("Battery critically low", level="warning")

            msg = BatteryState()
            msg.voltage = voltage
            msg.current = current
            self.battery_pub.publish(msg)

        self.cycle_count += 1

    def read_voltage(self):
        return 22.4

    def read_current(self):
        return 3.2

    def compute_soc(self, voltage):
        return max(0.0, min(100.0, (voltage - 18.0) / (25.2 - 18.0) * 100))


def main():
    rclpy.init()
    node = BatteryNode()
    rclpy.spin(node)
    rclpy.shutdown()


if __name__ == "__main__":
    main()

Using one BotPulse instance per process​

Create the botpulse.init() instance once (in __init__) and reuse it. This ensures:

  • Trace context is shared across all operations
  • Batching and flushing are handled efficiently
  • No duplicate connections to the collector
# Good — single instance, reused
class MyNode(Node):
    def __init__(self):
        self.pulse = botpulse.init(service_name="my_node", robot_id="robot-01")

# Bad — creates a new connection every time
def callback(self, msg):
    pulse = botpulse.init(service_name="my_node")  # don't do this

Using with launch files​

Bot Pulse needs no special launch configuration. Just start your node normally:

launch.py
from launch import LaunchDescription
from launch_ros.actions import Node

def generate_launch_description():
    return LaunchDescription([
        Node(package="my_pkg", executable="navigation_node"),
        Node(package="my_pkg", executable="battery_node"),
    ])

Environment variables​

You can also configure Bot Pulse via environment variables, which is useful in containerized or multi-robot deployments:

export BOTPULSE_SERVICE_NAME=navigation-node
export BOTPULSE_ROBOT_ID=robot-01
export BOTPULSE_ENDPOINT=http://collector:4317
# Picks up configuration from environment variables
pulse = botpulse.init()

Next steps​