# 查看话题数据内容
ros2 topic echo /cmd_vel

# 只看一条消息（避免刷屏）
ros2 topic echo /cmd_vel --once

# 查看发布频率
ros2 topic hz /scan
# 输出: average rate: 10.001

# 查看带宽占用
ros2 topic bw /camera/image_raw
# 输出: average: 27.3MB/s

# 查看话题延迟（消息头时间 vs 当前时间）
ros2 topic delay /scan
# 输出: average delay: 0.003s

# 列出所有活跃节点
ros2 node list
# /lidar_driver
# /slam_toolbox
# /nav2_controller

# 查看节点详细信息
ros2 node info /slam_toolbox
# 输出:
# Subscribers:
#   /scan: sensor_msgs/msg/LaserScan
# Publishers:
#   /map: nav_msgs/msg/OccupancyGrid
# Services:
#   /slam_toolbox/serialize_map
# Action Servers:
#   /slam_toolbox

# 查看服务类型
ros2 service type /spawn_entity
# 输出: gazebo_msgs/srv/SpawnEntity

# 手动调用服务（注意YAML格式）
ros2 service call /spawn_entity gazebo_msgs/srv/SpawnEntity \
  "{name: 'my_robot', xml: '<robot>...</robot>'}"

# 调用空参数服务
ros2 service call /reset_simulation std_srvs/srv/Empty

# 查看动作类型
ros2 action type /navigate_to_position
# 输出: nav2_msgs/action/NavigateToPose

# 发送动作目标并接收反馈
ros2 action send_goal /navigate_to_position nav2_msgs/action/NavigateToPose \
  "{pose: {header: {frame_id: 'map'}, pose: {position: {x: 2.0, y: 1.0, z: 0.0}}}}" \
  --feedback
# 输出:
# Feedback: current_pose: {x: 0.5, y: 0.2}
# Feedback: current_pose: {x: 1.2, y: 0.6}
# Result: result: true

# 启动rqt_graph
rqt_graph

# 命令行方式查看计算图（无需GUI）
ros2 run rqt_graph rqt_graph

# 启动rqt_plot
rqt_plot

# 命令行直接指定绘图话题
rqt_plot /cmd_vel/linear/x /odom/twist/twist/linear/x

# 启动rviz2
rviz2

# 指定配置文件启动
rviz2 -d my_config.rviz

# 录制指定话题
ros2 bag record /scan /tf /odom

# 录制所有话题
ros2 bag record -a

# 限制录制时长（秒）
ros2 bag record -a --duration 60

# 限制录制大小（MB）
ros2 bag record -a --max-bag-size 500

# 查看bag信息
ros2 bag info rosbag2_2026_06_07/

# 回放bag（发布clock话题供其他节点同步时间）
ros2 bag play rosbag2_2026_06_07/ --clock

# 2倍速回放
ros2 bag play rosbag2_2026_06_07/ --rate 2.0

// minimal_logger_node.cpp
#include <rclcpp/rclcpp.hpp>

class LoggerNode : public rclcpp::Node {
public:
    LoggerNode() : Node("logger_node") {
        // 设置本节点日志级别
        this->get_logger().set_level(rclcpp::Logger::Level::Debug);

        timer_ = this->create_wall_timer(
            std::chrono::seconds(1),
            [this]() {
                RCLCPP_DEBUG(this->get_logger(), "调试信息：定时器触发");
                RCLCPP_INFO(this->get_logger(), "常规信息：系统运行正常");
                RCLCPP_WARN(this->get_logger(), "警告信息：内存使用率80%%");
                RCLCPP_ERROR(this->get_logger(), "错误信息：传感器读取失败");
            });
    }

private:
    rclcpp::TimerBase::SharedPtr timer_;
};

int main(int argc, char** argv) {
    rclcpp::init(argc, argv);
    rclcpp::spin(std::make_shared<LoggerNode>());
    rclcpp::shutdown();
    return 0;
}

# 启动时设置全局日志级别
ros2 run my_package my_node --ros-args --log-level debug

# 设置特定节点日志级别
ros2 run my_package my_node --ros-args --log-level logger_node:=debug

# 运行时动态修改日志级别
ros2 service call /logger_node/set_logger_level rcl_interfaces/srv/SetLoggerLevel \
  "{logger_name: 'logger_node', level: 1}"
# level: 0=UNSET, 1=DEBUG, 2=INFO, 3=WARN, 4=ERROR, 5=FATAL

# 日志输出到文件
ros2 run my_package my_node --ros-args --log-file-name /tmp/my_node.log

# 同时保留终端输出
ros2 run my_package my_node --ros-args --log-file-name /tmp/my_node.log --log-level debug

# logger_node.py
import rclpy
from rclpy.node import Node
from rclpy.logging import get_logger

class LoggerNode(Node):
    def __init__(self):
        super().__init__('logger_node')

        # 设置日志级别
        self.get_logger().set_level(rclpy.logging.LoggingSeverity.DEBUG)

        self.timer = self.create_timer(1.0, self.timer_callback)

    def timer_callback(self):
        self.get_logger().debug('调试信息：定时器触发')
        self.get_logger().info('常规信息：系统运行正常')
        self.get_logger().warn('警告信息：内存使用率80%')
        self.get_logger().error('错误信息：传感器读取失败')

def main(args=None):
    rclpy.init(args=args)
    node = LoggerNode()
    rclpy.spin(node)
    node.destroy_node()
    rclpy.shutdown()

if __name__ == '__main__':
    main()

# 发布频率
ros2 topic hz /scan
# average rate: 10.001, min: 0.099s, max: 0.101s

# 带宽占用
ros2 topic bw /camera/image_raw
# average: 27.3MB/s

# 消息延迟
ros2 topic delay /scan
# average delay: 0.003s

# QoS详情（Jazzy增强）
ros2 topic info /scan --verbose
# 输出:
# QoS profile:
#   Reliability: BEST_EFFORT
#   Durability: VOLATILE
#   Deadline: 0ns
#   Lifespan: 0ns
#   Liveliness: AUTOMATIC
#   Subscription count: 2
#   Publication count: 1

// callback_group_demo.cpp
#include <rclcpp/rclcpp.hpp>
#include <chrono>

class CallbackGroupDemo : public rclcpp::Node {
public:
    CallbackGroupDemo() : Node("cb_group_demo") {
        // 串行组：回调之间互斥，同一时刻只执行一个
        serial_group_ = this->create_callback_group(
            rclcpp::CallbackGroupType::MutuallyExclusive);

        // 并行组：回调之间可并行
        parallel_group_ = this->create_callback_group(
            rclcpp::CallbackGroupType::Reentrant);

        // 串行订阅
        sub_serial_ = this->create_subscription<std_msgs::msg::String>(
            "/topic_a", 10,
            [this](const std_msgs::msg::String::SharedPtr msg) {
                RCLCPP_INFO(this->get_logger(), "串行回调A: %s", msg->data.c_str());
                std::this_thread::sleep_for(std::chrono::milliseconds(500));
            }, serial_group_);

        // 并行订阅
        sub_parallel_ = this->create_subscription<std_msgs::msg::String>(
            "/topic_b", 10,
            [this](const std_msgs::msg::String::SharedPtr msg) {
                RCLCPP_INFO(this->get_logger(), "并行回调B: %s", msg->data.c_str());
                std::this_thread::sleep_for(std::chrono::milliseconds(500));
            }, parallel_group_);
    }

private:
    rclcpp::CallbackGroup::SharedPtr serial_group_;
    rclcpp::CallbackGroup::SharedPtr parallel_group_;
    rclcpp::Subscription<std_msgs::msg::String>::SharedPtr sub_serial_;
    rclcpp::Subscription<std_msgs::msg::String>::SharedPtr sub_parallel_;
};

# executor_demo.py
import rclpy
from rclpy.executors import MultiThreadedExecutor

def main():
    rclpy.init()
    node1 = MyNode('node1')
    node2 = MyNode('node2')

    # 多线程执行器，线程数=CPU核心数
    executor = MultiThreadedExecutor(num_threads=4)
    executor.add_node(node1)
    executor.add_node(node2)

    try:
        executor.spin()
    finally:
        executor.shutdown()
        node1.destroy_node()
        node2.destroy_node()
        rclpy.shutdown()

if __name__ == '__main__':
    main()

// multi_executor.cpp
#include <rclcpp/rclcpp.hpp>

int main(int argc, char** argv) {
    rclcpp::init(argc, argv);

    auto node1 = std::make_shared<MyNode>("node1");
    auto node2 = std::make_shared<MyNode>("node2");

    // 多线程执行器
    rclcpp::executors::MultiThreadedExecutor executor(
        rclcpp::ExecutorOptions(), 4);  // 4个线程
    executor.add_node(node1);
    executor.add_node(node2);

    executor.spin();
    rclcpp::shutdown();
    return 0;
}

# 安装valgrind
sudo apt install valgrind

# 检测ROS2 C++节点内存泄漏
# 先source环境
source /opt/ros/jazzy/setup.bash

# 运行valgrind（注意：启动较慢，耐心等待）
valgrind --leak-check=full --show-leak-kinds=all \
  --track-origins=yes --verbose \
  --log-file=/tmp/valgrind_report.txt \
  ros2 run my_package my_node

# 查看报告
cat /tmp/valgrind_report.txt | grep "LEAK SUMMARY" -A 5
# 期望输出:
# LEAK SUMMARY:
#    definitely lost: 0 bytes in 0 blocks
#    indirectly lost: 0 bytes in 0 blocks

// 错误：裸指针未释放
class BadNode : public rclcpp::Node {
public:
    BadNode() : Node("bad_node") {
        data_ = new float[1024];  // 泄漏：析构时未delete
    }
private:
    float* data_;  // 应该用std::vector<float>
};

// 正确：使用智能指针或STL容器
class GoodNode : public rclcpp::Node {
public:
    GoodNode() : Node("good_node") {
        data_.resize(1024);  // 自动管理内存
    }
private:
    std::vector<float> data_;
};

# memory_tracker_node.py
import tracemalloc
import rclpy
from rclpy.node import Node

class MemoryTrackerNode(Node):
    def __init__(self):
        super().__init__('memory_tracker')

        # 启动内存追踪
        tracemalloc.start()

        self.timer = self.create_timer(5.0, self.check_memory)

    def check_memory(self):
        # 获取内存快照
        snapshot = tracemalloc.take_snapshot()
        top_stats = snapshot.statistics('lineno')

        self.get_logger().info('=== 内存占用TOP5 ===')
        for stat in top_stats[:5]:
            self.get_logger().info(f'  {stat}')

def main(args=None):
    rclpy.init(args=args)
    node = MemoryTrackerNode()
    rclpy.spin(node)
    node.destroy_node()
    tracemalloc.stop()
    rclpy.shutdown()

if __name__ == '__main__':
    main()

# 优化前：高频发布 + 重复计算
class UnoptimizedNode(Node):
    def __init__(self):
        super().__init__('unoptimized')
        # 100Hz发布，大部分消息没人消费
        self.pub = self.create_publisher(String, '/topic', 10)
        self.timer = self.create_timer(0.01, self.callback)  # 100Hz

    def callback(self):
        # 每次都重新计算
        result = self.heavy_computation()
        msg = String()
        msg.data = result
        self.pub.publish(msg)

# 优化后：按需发布 + 缓存结果
class OptimizedNode(Node):
    def __init__(self):
        super().__init__('optimized')
        self.pub = self.create_publisher(String, '/topic', 10)
        # 降低到10Hz
        self.timer = self.create_timer(0.1, self.callback)
        self.cached_result = None

    def callback(self):
        # 只在结果变化时重新计算
        if self.cached_result is None:
            self.cached_result = self.heavy_computation()
        msg = String()
        msg.data = self.cached_result
        self.pub.publish(msg)

# 设置DDS Domain（同一Domain的节点才能通信）
export ROS_DOMAIN_ID=30

# 多机通信时，避免Domain冲突
# 机器人A
export ROS_DOMAIN_ID=10
# 机器人B
export ROS_DOMAIN_ID=20

# qos_optimization.py
from rclpy.node import Node
from rclpy.qos import QoSProfile, ReliabilityPolicy, HistoryPolicy

class QoSOptimizedNode(Node):
    def __init__(self):
        super().__init__('qos_optimized')

        # 传感器数据：允许丢包，追求低延迟
        sensor_qos = QoSProfile(
            reliability=ReliabilityPolicy.BEST_EFFORT,
            history=HistoryPolicy.KEEP_LAST,
            depth=5
        )

        # 控制指令：必须可靠送达
        control_qos = QoSProfile(
            reliability=ReliabilityPolicy.RELIABLE,
            history=HistoryPolicy.KEEP_LAST,
            depth=10
        )

        self.scan_sub = self.create_subscription(
            LaserScan, '/scan', self.scan_callback, sensor_qos)
        self.cmd_pub = self.create_publisher(
            Twist, '/cmd_vel', control_qos)

# 录制指定话题
ros2 bag record /scan /tf /odom -o my_bag

# 录制所有话题（慎用，数据量大）
ros2 bag record -a -o all_topics_bag

# 限制录制时长
ros2 bag record -a --duration 120 -o timed_bag

# 限制录制大小
ros2 bag record -a --max-bag-size 1024 -o limited_bag

# 查看bag信息
ros2 bag info my_bag/
# 输出:
# Duration: 60.5s
# Start: 2026-06-07 10:00:00
# End: 2026-06-07 10:01:00
# Messages: 6050
# Topics with count:
#   /scan: 605 (10Hz)
#   /tf: 6050 (100Hz)
#   /odom: 605 (10Hz)

# 基本回放
ros2 bag play my_bag/

# 发布/clock话题（仿真时间同步）
ros2 bag play my_bag/ --clock

# 调整回放速率
ros2 bag play my_bag/ --rate 0.5   # 半速
ros2 bag play my_bag/ --rate 2.0   # 2倍速

# 只回放部分话题
ros2 bag play my_bag/ --topics /scan /odom

# 延迟启动（等待其他节点就绪）
ros2 bag play my_bag/ --delay 5.0

# 使用SQLite3存储（Jazzy默认）
ros2 bag record -a --storage sqlite3 -o sqlite_bag

# 查看存储配置
ros2 bag info sqlite_bag/

# 录制服务调用
ros2 bag record --services /spawn_entity /reset_simulation -o service_bag

# 回放时自动重放服务调用
ros2 bag play service_bag/

ros2_ws/
├── src/
│   ├── robot_bringup/          # 启动文件
│   │   ├── launch/
│   │   ├── config/
│   │   ├── package.xml
│   │   └── setup.py
│   ├── robot_description/      # 机器人描述包
│   │   ├── urdf/
│   │   ├── meshes/
│   │   ├── package.xml
│   │   └── setup.py
│   ├── robot_drivers/          # 驱动器包
│   │   ├── include/
│   │   ├── src/
│   │   ├── CMakeLists.txt
│   │   └── package.xml
│   ├── robot_navigation/       # 导航器包
│   │   ├── include/
│   │   ├── src/
│   │   ├── CMakeLists.txt
│   │   └── package.xml
│   └── robot_msgs/             # 自定义消息包
│       ├── msg/
│       ├── srv/
│       ├── action/
│       ├── CMakeLists.txt
│       └── package.xml
├── build/
├── install/
└── log/

<!-- package.xml 完整示例 -->
<package format="3">
  <name>robot_navigation</name>
  <version>1.0.0</version>
  <description>机器人导航功能包</description>
  <maintainer email="dev@linuxros.cn">linuxros</maintainer>
  <license>Apache-2.0</license>

  <!-- 构建依赖 -->
  <buildtool_depend>ament_cmake</buildtool_depend>

  <build_depend>rclcpp</build_depend>
  <build_depend>nav_msgs</build_depend>
  <build_depend>geometry_msgs</build_depend>
  <build_depend>tf2_ros</build_depend>

  <!-- 运行依赖 -->
  <exec_depend>rclcpp</exec_depend>
  <exec_depend>nav_msgs</exec_depend>
  <exec_depend>geometry_msgs</exec_depend>
  <exec_depend>tf2_ros</exec_depend>
  <exec_depend>robot_msgs</exec_depend>

  <!-- 测试依赖 -->
  <test_depend>ament_lint_auto</test_depend>
  <test_depend>ament_lint_common</test_depend>
  <test_depend>ament_cmake_gtest</test_depend>

  <export>
    <build_type>ament_cmake</build_type>
  </export>
</package>

# 基本构建
colcon build

# 只构建指定包（节省时间）
colcon build --packages-select robot_navigation

# 构建指定包及其依赖
colcon build --packages-up-to robot_navigation

# 并行构建（默认自动检测CPU核心数）
colcon build --parallel-workers 4

# 传递CMake参数
colcon build --cmake-args -DCMAKE_BUILD_TYPE=Release

# 跳过构建测试（加快构建速度）
colcon build --cmake-args -DBUILD_TESTING=OFF

# 构建完成后自动source
colcon build && source install/setup.bash

# 查看构建依赖
colcon list --topological-order

# C++代码规范检测
sudo apt install ros-jazzy-ament-lint

# 检查C++代码风格
ament_cpplint --filter=-legal/copyright src/

# 检查Python代码风格
ament_flake8 src/

# 一键运行所有lint检测
ament_lint_cmake CMakeLists.txt

# 在包中集成lint（CMakeLists.txt）
# find_package(ament_lint_auto REQUIRED)
# ament_lint_auto_find_test_dependencies()

// test/test_navigation.cpp
#include <gtest/gtest.h>
#include <rclcpp/rclcpp.hpp>
#include "robot_navigation/nav_engine.hpp"

TEST(NavEngineTest, ComputeDistance) {
    NavEngine engine;
    double dist = engine.compute_distance(0.0, 0.0, 3.0, 4.0);
    EXPECT_NEAR(dist, 5.0, 0.001);
}

TEST(NavEngineTest, GoalReached) {
    NavEngine engine;
    bool reached = engine.is_goal_reached(1.0, 1.0, 1.01, 1.01, 0.1);
    EXPECT_TRUE(reached);
}

int main(int argc, char** argv) {
    rclcpp::init(argc, argv);
    testing::InitGoogleTest(&argc, argv);
    auto result = RUN_ALL_TESTS();
    rclcpp::shutdown();
    return result;
}

# CMakeLists.txt 测试部分
if(BUILD_TESTING)
  find_package(ament_lint_auto REQUIRED)
  ament_lint_auto_find_test_dependencies()

  find_package(ament_cmake_gtest REQUIRED)
  ament_add_gtest(nav_test test/test_navigation.cpp)
  target_include_directories(nav_test PRIVATE include)
  target_link_libraries(nav_test navigation_lib)
  ament_target_dependencies(nav_test rclcpp)
endif()

# test/test_navigation.py
import pytest
from robot_navigation.nav_engine import NavEngine

def test_compute_distance():
    engine = NavEngine()
    dist = engine.compute_distance(0.0, 0.0, 3.0, 4.0)
    assert abs(dist - 5.0) < 0.001

def test_goal_reached():
    engine = NavEngine()
    reached = engine.is_goal_reached(1.0, 1.0, 1.01, 1.01, 0.1)
    assert reached is True

# setup.py 测试配置
entry_points={
    'pytest11': ['robot_navigation = pytest'],
},

# 运行所有测试
colcon test

# 运行指定包测试
colcon test --packages-select robot_navigation

# 查看测试结果
colcon test-result --verbose

# 查看测试日志
colcon test-result --all

name: ROS2 CI

on:
  push:
    branches: [main, develop]
  pull_request:
    branches: [main]

jobs:
  build-and-test:
    runs-on: ubuntu-24.04
    container:
      image: ros:jazzy-ros-base

    steps:
      # 检出代码
      - name: Checkout
        uses: actions/checkout@v4

      # 安装依赖
      - name: Install dependencies
        run: |
          apt-get update
          rosdep update
          cd ${{ github.workspace }}
          rosdep install --from-paths src --ignore-src -r -y

      # 构建
      - name: Build
        run: |
          cd ${{ github.workspace }}
          source /opt/ros/jazzy/setup.bash
          colcon build --cmake-args -DCMAKE_BUILD_TYPE=Release

      # 测试
      - name: Test
        run: |
          cd ${{ github.workspace }}
          source /opt/ros/jazzy/setup.bash
          source install/setup.bash
          colcon test
          colcon test-result --verbose

      # 上传测试报告
      - name: Upload test results
        if: always()
        uses: actions/upload-artifact@v4
        with:
          name: test-results
          path: |
            log/latest_test/
            build/*/test_results/

# Dockerfile
FROM ros:jazzy-ros-base

# 安装构建依赖
RUN apt-get update && apt-get install -y \
    python3-colcon-common-extensions \
    ros-jazzy-ros2-control \
    ros-jazzy-ros2-controllers \
    && rm -rf /var/lib/apt/lists/*

# 复制源码
WORKDIR /ros2_ws
COPY src src/

# 构建
RUN /bin/bash -c "source /opt/ros/jazzy/setup.bash && colcon build"

# 设置入口
ENTRYPOINT ["/bin/bash", "-c", "source /opt/ros/jazzy/setup.bash && source install/setup.bash && ros2 launch robot_bringup bringup.launch.py"]

# 构建镜像
docker build -t robot_nav:jazzy .

# 运行容器
docker run --rm --net=host robot_nav:jazzy

# 交互式进入容器调试
docker run -it --rm --net=host robot_nav:jazzy /bin/bash

MAJOR.MINOR.PATCH

MAJOR：不兼容的API变更
MINOR：向后兼容的功能新增
PATCH：向后兼容的Bug修复

<package format="3">
  <version>1.2.3</version>
  <!-- MAJOR=1, MINOR=2, PATCH=3 -->
</package>

# 1. 确保package.xml版本号正确
# 2. 在GitHub创建Release标签
git tag 1.0.0
git push origin 1.0.0

# 3. Fork rosdistro仓库
# https://github.com/ros/rosdistro

# 4. 修改对应ROS版本的distribution.yaml
# 在jazzy/distribution.yaml中添加包信息

# 5. 提交PR到rosdistro
# 等待ROS2维护者Review和合并
# 6. 合并后，用户可通过apt安装
sudo apt install ros-jazzy-my-package

# 1. 检查日志级别（可能被设为WARN以上）
ros2 run my_package my_node --ros-args --log-level debug

# 2. 检查输出是否被重定向
# 某些launch文件会重定向日志到文件
ros2 run my_package my_node  # 直接运行，不用launch

# 3. 检查节点是否真的在运行
ros2 node list | grep my_node

# 4. 检查日志文件（Jazzy特性）
ros2 run my_package my_node --ros-args --log-file-name /tmp/debug.log
cat /tmp/debug.log

# 1. 检查QoS配置
ros2 topic info /scan --verbose

# 2. 检查是否有RELIABLE + 大消息的组合
# 传感器数据应该用BEST_EFFORT

# 3. 检查DDS配置
# FastDDS默认配置可能不适合高频率场景
export FASTRTPS_DEFAULT_PROFILES_FILE=/path/to/fastdds_profile.xml

# 4. 检查网络延迟
ping <remote_host>

# 5. 调整QoS为BEST_EFFORT
# Python示例
sensor_qos = QoSProfile(
    reliability=ReliabilityPolicy.BEST_EFFORT,
    history=HistoryPolicy.KEEP_LAST,
    depth=1  # 只保留最新数据
)

# 1. 检查依赖是否安装
rosdep install --from-paths src --ignore-src -r -y

# 2. 清理构建缓存重新构建
rm -rf build/ install/ log/
colcon build

# 3. 检查package.xml依赖声明
# 编译错误通常是因为build_depend缺失

# 4. 检查CMakeLists.txt
# find_package是否包含所有依赖
# ament_target_dependencies是否完整

# 5. 只构建失败的包查看详细错误
colcon build --packages-select failing_package --event-handlers console_direct+