ROS2实操可供大模型检索流程与参考,亲测上机试验运行
运行环境是ubuntu24    

首先我们在之前要要求ros2环境,这些安装好的环境的方式就是我也经在虚拟机中装好了,大家直接下载虚拟机运行就好,我们直接来开发ros2,以及了解操作流程

一、工作空间基础概念 📌

1. 定义

工作空间(workspace)是存放 ROS2 项目全部开发文件的顶层文件夹,是机器人项目开发的大本营,自定义命名示例:dev_ws。其实意思是每个个功能都 是一个节点, 因为机器人有很多功能,比如行走,说话,每个功能就是一个节点,因为开发机器人不是一家公司可以完成的,所以为了好管理,ros2 他有节点的概念,大家管自己的节点,通过ros来调用,所以每个节点是什么语言开发的无关得要。

2. 四大子目录作用详解

表格

文件夹空间名称核心用途
src代码空间唯一需要手动编写代码的目录,所有功能包、源码、自定义节点都存放在此处
build编译空间colcon 编译时生成的中间缓存文件,存放编译过程临时产物
install安装空间编译完成后的可执行程序、库文件、启动文件安装目录,后续运行节点必须 source 该目录内脚本
log日志空间编译、运行过程产生的系统日志、报错日志,排查程序异常时查看

💡 补充:build/install/log 三个目录全部由编译命令自动生成,不需要手动创建、修改。


二、工作空间完整实操分步命令(可逐条复制执行)
说明,我们其实只需要新建src文件,安装所有需要的依赖,其它文件是编译自动生成,只要你按ros2的规范来发
 

步骤 1:创建工作空间与 src 源码目录

bash

# 递归创建工作空间及src子目录
mkdir -p ~/dev_ws/src
# 进入src目录
cd ~/dev_ws/src
# 克隆示例代码仓库到src内
git clone https://gitee.com/guyuehome/ros2_21_tutorials.git

步骤 2:一键自动安装项目所有依赖

bash

# 安装pip工具
sudo apt install -y python3-pip
# 安装rosdep依赖管理工具
sudo pip3 install rosdepc
# 初始化+更新rosdep数据源
sudo rosdepc init && rosdepc update
# 退回工作空间根目录
cd ..
# 自动检索src内所有功能包缺失依赖并批量安装(humble版本)
rosdepc install -i --from-path src --rosdistro humble -y

步骤 3:编译整个工作空间

bash

# 安装colcon编译工具
sudo apt install python3-colcon-ros
# 进入工作空间根目录
cd ~/dev_ws/
# 执行编译,自动生成build、install、log文件夹
colcon build

⚠️ 注意:编译命令必须在工作空间根目录执行,不能进入 src 内部编译。

步骤 4:配置环境变量(让系统识别编译后的程序)

目录

ROS2 工作空间 完整学习笔记

一、工作空间基础概念 📌

1. 定义

2. 四大子目录作用详解

二、工作空间完整实操分步命令(可逐条复制执行)说明,我们其实只需要新建src文件,安装所有需要的依赖,其它文件是编译自动生成,只要你按ros2的规范来发

步骤 1:创建工作空间与 src 源码目录

步骤 2:一键自动安装项目所有依赖

步骤 3:编译整个工作空间

步骤 4:配置环境变量(让系统识别编译后的程序)

临时生效(当前终端有效,新开终端需重新执行)

永久生效(所有终端自动加载,推荐配置一次即可)


bash

source install/local_setup.sh
永久生效(所有终端自动加载,推荐配置一次即可)

bash

echo "source ~/dev_ws/install/local_setup.sh" >> ~/.bashrc
# 刷新终端配置,立即生效
source ~/.bashrc

表格步聚

序号图片中的流程步骤对应代码核心语句
1编程接口初始化rclpy.init(args=args)
2创建节点并初始化super().__init__("节点名")、实例化节点
3创建发布者对象create_publisher(String,"chatter",10)
4创建并填充话题消息msg=String()msg.data="Hello World"
5发布话题消息self.publisher.publish(msg)
6销毁节点并关闭接口destroy_node()rclpy.shutdown()

1. 功能需求

编写一个 ROS2 发布节点,周期性向外发送字符串文本消息;配套订阅节点可以接收并打印这条消息。

2. 消息与话题规划

  1. 话题名称:固定为 chatter(订阅端必须同名才能接收)
  2. 消息类型:标准字符串消息 std_msgs/msg/String
  3. 发送周期:0.5 秒发送 1 条消息
  4. 消息内容:固定文本 Hello World
  5. 节点名称:topic_helloworld_pub

python 详细代码

import rclpy
from rclpy.node import Node
from std_msgs.msg import String

class PublisherNode(Node):
    def __init__(self):
        # 步骤2:初始化节点,指定节点名
        super().__init__("topic_helloworld_pub")

        # 步骤3:创建发布者对象
        self.publisher = self.create_publisher(String, "chatter", 10)

        # 设置定时周期,周期性触发发布逻辑
        self.timer_period = 0.5
        self.timer = self.create_timer(self.timer_period, self.timer_callback)

    def timer_callback(self):
        # 步骤4:新建消息对象,填充消息内容
        msg = String()
        msg.data = "Hello World"

        # 步骤5:执行发布操作
        self.publisher.publish(msg)
        self.get_logger().info(f"Publishing: {msg.data}")

def main(args=None):
    # 步骤1:ROS2编程接口初始化
    rclpy.init(args=args)

    pub_node = PublisherNode()

    # 持续自旋,节点保持运行
    rclpy.spin(pub_node)

    # 步骤6:销毁节点,关闭ROS2接口,释放资源
    pub_node.destroy_node()
    rclpy.shutdown()

if __name__ == "__main__":
    main()

四、配套订阅节点(用来验证发布效果)

python

import rclpy
from rclpy.node import Node
from std_msgs.msg import String

class SubscriberNode(Node):
    def __init__(self):
        super().__init__("topic_helloworld_sub")
        # 订阅同名话题chatter
        self.subscription = self.create_subscription(
            String,
            "chatter",
            self.listener_callback,
            10
        )

    def listener_callback(self, msg):
        self.get_logger().info(f"收到消息:{msg.data}")

def main(args=None):
    rclpy.init(args=args)
    sub_node = SubscriberNode()
    rclpy.spin(sub_node)
    sub_node.destroy_node()
    rclpy.shutdown()

if __name__ == "__main__":
    main()

谢双元总结:

逐步骤代码对应拆解(对标 6 条流程)

前置导入依赖

python

import rclpy
from rclpy.node import Node
# 导入标准字符串消息类型
from std_msgs.msg import String

步骤 1:编程接口初始化

对应代码:rclpy.init(args=args)作用:初始化 ROS2 的 Python 客户端接口,整个 ROS2 程序运行前必须执行,否则无法创建节点、发布订阅。

步骤 2:创建节点并初始化

  1. 自定义节点类继承 ROS2 基类Node
  2. 构造函数调用父类构造,绑定节点名称;
  3. 实例化节点对象。

python

class PublisherNode(Node):
    def __init__(self):
        # 2、创建节点,节点命名 topic_helloworld_pub
        super().__init__("topic_helloworld_pub")

步骤 3:创建发布者对象

调用节点内置create_publisher方法,绑定消息类型、话题名、消息队列长度

python

        # 3、创建发布者对象
        self.publisher = self.create_publisher(
            msg_type=String,        # 消息类型
            topic="chatter",        # 话题名称
            qos_profile=10          # 消息队列长度
        )

配套定时器(实现周期性重复发布):

python

        # 0.5s触发一次回调函数,重复执行发消息逻辑
        self.timer_period = 0.5
        self.timer = self.create_timer(self.timer_period, self.timer_callback)

步骤 4:创建并填充话题消息

在定时器回调函数里,实例化消息对象,给消息成员赋值:

python

    def timer_callback(self):
        # 4、创建消息实例 + 填充消息内容
        msg = String()
        msg.data = "Hello World"

步骤 5:发布话题消息

调用发布者对象的publish()方法,把封装好的消息发送出去

python

        # 5、执行消息发布
        self.publisher.publish(msg)
        # 控制台打印日志,查看发布内容
        self.get_logger().info(f"已发布消息:{msg.data}")

步骤 6:销毁节点并关闭接口

程序正常退出前执行:销毁节点、关闭 ROS2 接口,释放资源

python

def main(args=None):
    # 步骤1:ROS2 Python接口初始化
    rclpy.init(args=args)

    # 步骤2:创建节点实例
    pub_node = PublisherNode()

    # 节点自旋,持续运行,不断执行定时器回调发布消息
    rclpy.spin(pub_node)

    # 步骤6:销毁节点 + 关闭ROS2接口
    pub_node.destroy_node()
    rclpy.shutdown()

if __name__ == "__main__":
    main()

示例三:ROS2 OpenCV 摄像头图像发布 + 订阅全套完整梳理

一、整体需求与规划

1. 项目需求

  1. 发布端:调用本地摄像头实时采集画面,把 OpenCV 图像转换成 ROS2 标准图像消息,通过话题持续发布;
  2. 订阅端:订阅图像话题,把 ROS 图像消息转回 OpenCV 格式,做红色目标轮廓检测、框选、标记中心点,弹窗显示处理后的画面。

2. 核心参数规划

表格

配置项设定值说明
话题名称image_raw发布、订阅必须完全一致
ROS 图像消息类型sensor_msgs/msg/ImageROS 标准图像消息
图像转换工具CvBridge打通 OpenCV Mat ↔ ROS Image 消息
发布帧率周期0.1s10 帧 / 秒推送图像
发布节点名topic_webcam_pubROS 节点唯一标识
订阅节点名topic_webcam_subROS 节点唯一标识
图像处理逻辑HSV 红色阈值二值化 + 轮廓检测 + 过滤小噪声 + 绘制轮廓 + 标记中心订阅端内置视觉算法

3. 整体执行流程

  1. 发布端:ROS 接口初始化→创建节点→创建发布者 + 定时器→打开摄像头→定时读取帧→CvBridge 转 ROS 消息→发布图像;
  2. 订阅端:ROS 接口初始化→创建节点→创建订阅器→收到消息自动回调→转回 OpenCV 图像→红色目标检测绘图→窗口显示。

因为我们要用摄像头,所以要装一下ros 发布的读usb包

# 1. 安装ROS Humble官方usb_cam功能包
sudo apt install ros-humble-usb-cam

# 2. 启动usb相机发布节点,自动读取USB摄像头,向外发布图像话题
ros2 run usb_cam usb_cam_node_exe

# 3. 运行你自己写的图像订阅节点,接收相机发出来的图像做处理
ros2 run learning_topic topic_webcam_sub

二、是不是所有 USB 摄像头都支持?结论先说

不是全部 USB 摄像头都能直接即用,绝大多数标准 UVC 协议 USB 摄像头免驱支持;非 UVC 私有协议摄像头无法直接使用 usb_cam 包。

1. 能直接支持的摄像头(95% 普通电脑 USB 摄像头)

usb_cam 底层依赖 Linux 内核的v4l2视频子系统,只兼容 UVC(USB Video Class)标准协议 摄像头:

  • 笔记本内置摄像头、普通台式 USB 免驱摄像头、常用工业 USB 相机(罗技、雷柏、普通百元监控 USB 相机);
  • Ubuntu 系统插上就能被ls /dev/video*识别,不需要额外装厂商驱动;
  • 执行第二条命令直接就能启动,默认发布/image_raw图像话题,和你之前写的订阅节点完美对接。

2. 无法直接使用的 USB 相机

  1. 老旧特殊工业相机、定制私有协议 USB 相机,不遵循 UVC 标准;
  2. 部分高清高速 USB 工业相机,厂商自研驱动,不走 v4l2;这类设备无法被usb_cam包识别,不能直接用这条命令启动,需要用厂商提供的 ROS 驱动或者 SDK 单独开发采集程序。

三、快速自测你的 USB 相机能不能被识别

步骤 1:插上 USB 摄像头,终端查看设备

bash

ls /dev/video*

正常会列出/dev/video0/dev/video1等设备号,代表系统识别到硬件。

步骤 2:测试 v4l2 读取

bash

v4l2-ctl --list-formats

能打印分辨率、像素格式,就说明 UVC 兼容,可以直接用usb_cam

四、补充 usb_cam 常用配置细节

  1. 多摄像头时,可以指定设备号:

bash

ros2 run usb_cam usb_cam_node_exe --ros-args -p device_name:=/dev/video1
  1. 默认发布话题就是image_raw,和你之前写的topic_webcam_sub订阅话题名一致,无需修改代码就能直接接收画面;
  2. 不用自己再写 OpenCV 采集发布节点,官方 usb_cam 已经封装好了采集、转 ROS 消息、话题发布全套逻辑。

二、发布节点:摄像头图像发布代码(topic_webcam_pub.py)

完整可运行代码

python

import rclpy
from rclpy.node import Node
from sensor_msgs.msg import Image
from cv_bridge import CvBridge
import cv2

class ImagePublisher(Node):
    def __init__(self, name):
        super().__init__(name)
        # 1. 创建图像发布者
        self.publisher_ = self.create_publisher(Image, 'image_raw', 10)
        # 2. 定时器:0.1s执行一次回调,持续推流
        self.timer = self.create_timer(0.1, self.timer_callback)
        # 3. 打开默认摄像头(设备号0)
        self.cap = cv2.VideoCapture(0)
        # 4. 实例化图像转换器
        self.cv_bridge = CvBridge()

    def timer_callback(self):
        # 读取一帧摄像头画面
        ret, frame = self.cap.read()
        if ret is True:
            # OpenCV图像 → ROS Image消息并发布
            self.publisher_.publish(
                self.cv_bridge.cv2_to_imgmsg(frame, 'bgr8')
            )
            self.get_logger().info('Publishing video frame')

def main(args=None):
    rclpy.init(args=args)
    node = ImagePublisher("topic_webcam_pub")
    rclpy.spin(node)
    # 释放摄像头资源
    node.cap.release()
    node.destroy_node()
    rclpy.shutdown()

if __name__ == '__main__':
    main()

发布端关键代码逐行详解

1)核心导入库说明

python

from sensor_msgs.msg import Image    # ROS标准图像消息载体
from cv_bridge import CvBridge       # OpenCV与ROS图像互转核心类
import cv2                           # 摄像头采集+图像原生操作

ROS 本身不能直接传输 OpenCV 的矩阵图像,必须依靠CvBridge做格式封装和解封。

2)发布者创建语句

python

self.publisher_ = self.create_publisher(Image, 'image_raw', 10)
  • 参数 1:消息类型Image,ROS 标准图像消息;
  • 参数 2:话题名image_raw,订阅端必须一字不差;
  • 参数 3:QoS 队列长度 10,缓存未及时接收的图像帧。
3)定时器周期推流

python

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

0.1 秒触发一次回调,连续读取摄像头画面并发布,控制帧率。

4)摄像头初始化

python

self.cap = cv2.VideoCapture(0)

设备编号0代表电脑默认内置摄像头;外接 USB 摄像头可尝试改成12

5)图像格式转换 + 发布(最核心)

python

self.cv_bridge.cv2_to_imgmsg(frame, 'bgr8')
  • cv2_to_imgmsg:OpenCV Mat 图像 → ROS Image 消息;
  • bgr8:OpenCV 默认存储格式:BGR 三通道、8 位深度,必须严格对应;
  • 转换完成后直接调用publish()发送到话题。
6)资源释放补充

节点退出时必须执行node.cap.release(),否则摄像头会被占用,下次运行打不开。


三、订阅节点:接收图像 + 红色目标检测(topic_webcam_sub.py)

完整可运行代码

python

import rclpy
from rclpy.node import Node
from sensor_msgs.msg import Image
from cv_bridge import CvBridge
import cv2
import numpy as np

# HSV红色阈值范围(必须提前定义,原图缺失)
lower_red = np.array([0, 120, 70])
upper_red = np.array([10, 255, 255])

class ImageSubscriber(Node):
    def __init__(self, name):
        super().__init__(name)
        # 创建订阅器
        self.sub = self.create_subscription(
            Image,
            'image_raw',
            self.listener_callback,
            10
        )
        self.cv_bridge = CvBridge()

    # 红色目标检测+绘图函数
    def object_detect(self, image):
        # BGR图像转HSV色彩空间,颜色分割更稳定
        hsv_img = cv2.cvtColor(image, cv2.COLOR_BGR2HSV)
        # 红色区域二值掩码提取
        mask_red = cv2.inRange(hsv_img, lower_red, upper_red)
        # 查找轮廓
        contours, hierarchy = cv2.findContours(
            mask_red, cv2.RETR_LIST, cv2.CHAIN_APPROX_NONE
        )
        # 遍历所有轮廓,过滤小噪声轮廓
        for cnt in contours:
            if cnt.shape[0] < 150:
                continue
            # 获取轮廓外接矩形
            (x, y, w, h) = cv2.boundingRect(cnt)
            # 绘制轮廓(绿色)
            cv2.drawContours(image, [cnt], -1, (0, 255, 0), 2)
            # 绘制目标中心点(红色实心圆)
            center_x = int(x + w/2)
            center_y = int(y + h/2)
            cv2.circle(image, (center_x, center_y), 5, (0, 255, 0), -1)
        # 弹窗显示处理后图像
        cv2.imshow("object", image)
        cv2.waitKey(10)

    # 话题接收回调函数
    def listener_callback(self, data):
        self.get_logger().info('Receiving video frame')
        # ROS图像消息转回OpenCV图像
        image = self.cv_bridge.imgmsg_to_cv2(data, 'bgr8')
        # 调用检测算法
        self.object_detect(image)

def main(args=None):
    rclpy.init(args=args)
    node = ImageSubscriber("topic_webcam_sub")
    rclpy.spin(node)
    # 关闭OpenCV窗口
    cv2.destroyAllWindows()
    node.destroy_node()
    rclpy.shutdown()

if __name__ == '__main__':
    main()

订阅端关键代码逐行详解

1)订阅器创建

python

self.sub = self.create_subscription(Image,'image_raw',self.listener_callback,10)
  • 话题名image_raw、消息类型Image必须和发布端完全一致;
  • listener_callback:收到图像消息后自动触发的回调函数。
2)ROS 消息转回 OpenCV 图像(双向转换反向操作)

python

image = self.cv_bridge.imgmsg_to_cv2(data, 'bgr8')

imgmsg_to_cv2:把 ROS 封装好的图像消息还原成 OpenCV 可直接操作的像素矩阵,通道格式依旧是bgr8

3)HSV 颜色空间转换(颜色检测核心)

python

hsv_img = cv2.cvtColor(image, cv2.COLOR_BGR2HSV)

BGR 直接做颜色分割容易受光照影响,HSV 分离色相、饱和度、亮度,红色物体提取鲁棒性更强。

4)阈值掩码二值化

python

mask_red = cv2.inRange(hsv_img, lower_red, upper_red)

在 HSV 区间内,红色像素置为白色 (255),其余全部黑色 (0),得到红色区域掩码图。

5)轮廓检测 + 噪声过滤

python

contours, hierarchy = cv2.findContours(mask_red, cv2.RETR_LIST, cv2.CHAIN_APPROX_NONE)
if cnt.shape[0] < 150: continue
  • 查找所有连通轮廓;
  • 像素点数小于 150 的轮廓判定为噪点直接跳过,避免微小干扰。
6)轮廓绘制 + 中心点标记

python

cv2.drawContours(image, [cnt], -1, (0, 255, 0), 2)
cv2.circle(image, (center_x, center_y), 5, (0, 255, 0), -1)
  • drawContours:绿色线条框出红色目标整体轮廓;
  • circle:在目标几何中心画实心圆点,标记目标位置。
7)OpenCV 窗口持续刷新

python

cv2.imshow("object", image)
cv2.waitKey(10)

waitKey(10)必须保留,否则图像窗口卡死不刷新。


四、功能包配置 & 运行步骤

1. 功能包创建

bash

ros2 pkg create --build-type ament_python webcam_topic_demo --dependencies rclpy sensor_msgs cv_bridge opencv-python

2. 修改 setup.py 入口点

python

entry_points={
    'console_scripts': [
        'webcam_pub = webcam_topic_demo.topic_webcam_pub:main',
        'webcam_sub = webcam_topic_demo.topic_webcam_sub:main',
    ],
},

3. 编译刷新环境

bash

colcon build --packages-select webcam_topic_demo
source install/setup.bash

4. 双终端分别启动

终端 1(启动摄像头发布):

bash

ros2 run webcam_topic_demo webcam_pub

会自动弹出摄像头采集画面,终端持续打印Publishing video frame

终端 2(启动图像订阅 + 目标检测):

bash

ros2 run webcam_topic_demo webcam_sub

弹出object窗口,画面里红色物体会自动被绿色轮廓包围、中心点标记。


五、高频易错点总结

  1. 话题名大小写严格一致:发布订阅都是image_raw,写错收不到图像;
  2. 通道格式配对cv2_to_imgmsgimgmsg_to_cv2都必须写bgr8,不能写成 rgb8,画面会颜色颠倒;
  3. CvBridge 不能省略:ROS 无法直接传输 OpenCV 图像,缺少转换代码会直接运行报错;
  4. 摄像头占用释放:发布节点退出务必cap.release(),否则下次运行摄像头打不开;
  5. 订阅端 HSV 阈值必须定义:原图代码省略了lower_red/upper_red,我已补充完整,否则代码无法运行。

最后给出一个查看节点的工具    rqt_graph

属于 ROS 官方配套 GUI 工具(rqt 插件),但分安装版本决定是否预装:

  1. 如果你当初装的是 ros-humble-desktop / desktop-full 桌面完整版 ROS2自动一并安装,不用你手动单独下载,装 ROS 本体的时候就顺带装好了。
  2. 如果你装的是极简版(ros-humble-ros-base,无 GUI、纯后台运行版)❌ 不会自带,必须手动敲命令单独安装。
Logo

脑启社区是一个专注类脑智能领域的开发者社区。欢迎加入社区,共建类脑智能生态。社区为开发者提供了丰富的开源类脑工具软件、类脑算法模型及数据集、类脑知识库、类脑技术培训课程以及类脑应用案例等资源。

更多推荐