Script show topics dependency tree

[xuyuan]

import rclpy
from rclpy.node import Node
from ros2node.api import get_node_names, get_node_info
from ros2topic.api import get_topic_names_and_types

class TopicGraphExplorer(Node):
def init(self, start_topic, max_depth=3):
super().init('topic_graph_explorer')
self.visited_nodes = set()
self.max_depth = max_depth
self.explore_topic(start_topic, 0)

def explore_topic(self, topic, depth):
    if depth > self.max_depth:
        return

    self.get_logger().info(f"{'  ' * depth}Exploring topic: {topic}")

    # Get all nodes publishing to this topic
    publishers = self.get_publishers_for_topic(topic)
    self.get_logger().info(f"{'  ' * depth}Publishers: {publishers}")

    for node in publishers:
        if node in self.visited_nodes:
            continue
        self.visited_nodes.add(node)

        # Get all topics subscribed by this node
        subscribed_topics = self.get_subscribed_topics(node)
        self.get_logger().info(f"{'  ' * depth}Node '{node}' subscribes to: {subscribed_topics}")

        # Recursively explore these topics
        for sub_topic in subscribed_topics:
            self.explore_topic(sub_topic, depth + 1)

def get_publishers_for_topic(self, topic):
    """ Return a list of nodes publishing to a given topic """
    topic_info = self.get_topic_names_and_types()
    publishers = []

    for topic_name, _ in topic_info:
        if topic_name == topic:
            node_info = get_node_names(self)
            for node in node_info:
                details = get_node_info(self, node)
                if topic in details.publisher_names:
                    publishers.append(node)

    return publishers

def get_subscribed_topics(self, node_name):
    """ Return a list of topics a node is subscribed to """
    node_info = get_node_info(self, node_name)
    return list(node_info.subscription_names) if node_info else []

def main():
rclpy.init()
start_topic = '/example_topic' # Change this to your desired topic
explorer = TopicGraphExplorer(start_topic)
rclpy.shutdown()

if name == 'main':
main()

[xuyuan]

#!/bin/bash

Max depth for recursion

MAX_DEPTH=3

Function to recursively explore topics

explore_topic() {
local topic="$1"
local depth="$2"
local indent=$(printf ' %.0s' $(seq 1 $depth))

if [ "$depth" -ge "$MAX_DEPTH" ]; then
    return
fi

echo "${indent}Exploring topic: $topic"

# Get all nodes publishing to this topic
publishers=$(ros2 topic info "$topic" | grep "Publisher count" | awk '{print $3}')
if [ "$publishers" -eq 0 ]; then
    echo "${indent}No publishers found."
    return
fi

nodes=($(ros2 topic info "$topic" | grep -A100 "Publisher count" | grep "Node name" | awk '{print $3}'))

echo "${indent}Publishers: ${nodes[*]}"

for node in "${nodes[@]}"; do
    echo "${indent}Inspecting node: $node"

    # Get topics this node subscribes to
    subscribed_topics=($(ros2 node info "$node" | grep "Subscribed topics" -A100 | grep -v "Published topics" | awk '{print $1}' | grep "^/"))

    if [ ${#subscribed_topics[@]} -eq 0 ]; then
        echo "${indent}  No subscribed topics."
    else
        echo "${indent}  Subscribed topics: ${subscribed_topics[*]}"
        for sub_topic in "${subscribed_topics[@]}"; do
            explore_topic "$sub_topic" $((depth + 1))
        done
    fi
done

}

Starting topic (change this as needed)

START_TOPIC="/example_topic"

echo "Starting topic exploration from: $START_TOPIC"
explore_topic "$START_TOPIC" 0