# py_vision_demo — Python 图像话题(cv_bridge + OpenCV) > ROS2 图像数据流:`fake_camera` 模拟相机 → 发布 Image → `image_processor` 用 OpenCV 处理。 > > 预计学习时间:1-2 小时。 --- ## 这是什么? ROS2 图像话题跟普通 Topic 一样,**只是消息类型用 `sensor_msgs/Image`**。本包演示: 1. **fake_camera**:周期性发布合成图像(640x480,带渐变背景 + 帧号文本 + 中心圆) 2. **image_processor**:订阅图像 → 转 numpy → OpenCV 处理 → 再发回 **关键工具**: `cv_bridge`(ROS Image ↔ OpenCV numpy array)。 --- ## 🎯 学完之后你能做什么? 1. ✅ 理解 `sensor_msgs/Image` 消息结构(height/width/encoding/data) 2. ✅ 用 `cv_bridge.imgmsg_to_cv2()` 把 ROS Image 转 OpenCV 数组 3. ✅ 用 OpenCV 处理图像(灰度、画框、滤波) 4. ✅ 用 `cv_bridge.cv2_to_imgmsg()` 转回 ROS Image 发布 5. ✅ 用 `image_transport` 压缩传输(可选) --- ## 📁 文件结构 ``` src/py_vision_demo/ ├── py_vision_demo/ │ ├── fake_camera.py # 模拟相机 Publisher │ └── image_processor.py # 图像处理 Subscriber ├── launch/vision_launch.py # 一键启动 ├── test/ │ ├── conftest.py │ ├── test_fake_camera.py │ ├── test_image_processor.py │ └── test_vision.py # 端到端测试 └── setup.py ``` --- ## 🚀 跑起来 ### 终端 1:启动 fake_camera + image_processor ```bash source /opt/ros/humble/setup.bash source /root/ros2_ws/install/setup.bash ros2 launch py_vision_demo vision_launch.py ``` **预期输出**: ``` [INFO] [fake_camera]: FakeCamera started: 640x480 @ 1Hz, topic="/image_raw" [INFO] [image_processor]: ImageProcessor started: topic="/image_processed" [INFO] [image_processor]: processing frame=0, mean_brightness=127.5 ... ``` ### 终端 2:看图像话题列表 ```bash ros2 topic list ``` **预期输出**: ``` /image_raw /image_processed /parameter_events /rosout ``` ### 终端 3:运行时改参数 ```bash # 改分辨率 ros2 param set fake_camera image_width 320 ros2 param set fake_camera image_height 240 # 改帧率 ros2 param set fake_camera publish_rate_hz 5.0 # 改处理模式(下游 image_processor) ros2 param set image_processor mode 'edges' # 或 'gray' 或 'raw' ``` --- ## 📖 核心代码解读 ### fake_camera.py(发布图像) ```python import cv2 import numpy as np from sensor_msgs.msg import Image class FakeCamera(Node): def __init__(self): super().__init__('fake_camera') # 1) 声明参数 self.declare_parameter('image_width', 640, ...) self.declare_parameter('image_height', 480, ...) self.declare_parameter('publish_rate_hz', 1.0, ...) # 2) 读参数 self._width = self.get_parameter('image_width').value self._height = self.get_parameter('image_height').value # 3) 创建 Image Publisher self.publisher_ = self.create_publisher(Image, '/image_raw', 10) # 4) 定时器 self.timer_ = self.create_timer(1.0/rate, self._publish_frame) def _publish_frame(self): # 1) 用 numpy + OpenCV 生成合成图像 frame = np.zeros((self._height, self._width, 3), dtype=np.uint8) frame[:, :, 0] = np.linspace(0, 255, self._width, dtype=np.uint8) # B frame[:, :, 1] = np.linspace(0, 255, self._height, dtype=np.uint8) # G cv2.putText(frame, f'frame={self._frame_count}', (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1.0, (255,255,255), 2) # 2) 转 ROS Image 消息 msg = Image() msg.height = self._height msg.width = self._width msg.encoding = 'bgr8' # OpenCV 默认 BGR! msg.step = self._width * 3 # 每行字节数(width * 3 channels) msg.data = frame.tobytes() # numpy → bytes # 3) 发布 self.publisher_.publish(msg) ``` **关键**:`encoding='bgr8'`(OpenCV 用 BGR 而非 RGB!),`step = width * 3`(每行字节数)。 ### image_processor.py(订阅 + 处理 + 再发布) ```python import cv2 from cv_bridge import CvBridge class ImageProcessor(Node): def __init__(self): super().__init__('image_processor') self._bridge = CvBridge() # 关键:cv_bridge 实例 # 订阅图像 self.subscription = self.create_subscription( Image, '/image_raw', self._on_image, 10) # 发布处理后的图像 self.publisher_ = self.create_publisher(Image, '/image_processed', 10) def _on_image(self, msg): # 1) ROS Image → OpenCV numpy cv_image = self._bridge.imgmsg_to_cv2(msg, desired_encoding='bgr8') # 2) OpenCV 处理 mode = self.get_parameter('mode').value if mode == 'gray': processed = cv2.cvtColor(cv_image, cv2.COLOR_BGR2GRAY) processed = cv2.cvtColor(processed, cv2.COLOR_GRAY2BGR) # 转回 3 通道 elif mode == 'edges': gray = cv2.cvtColor(cv_image, cv2.COLOR_BGR2GRAY) processed = cv2.cvtColor(cv2.Canny(gray, 50, 150), cv2.COLOR_GRAY2BGR) else: processed = cv_image # 3) numpy → ROS Image out_msg = self._bridge.cv2_to_imgmsg(processed, encoding='bgr8') out_msg.header = msg.header # 保留 timestamp + frame_id # 4) 发布 self.publisher_.publish(out_msg) ``` --- ## 🧪 跑测试 ```bash colcon test --packages-select py_vision_demo colcon test-result --all --verbose ``` **预期**:`py_vision_demo: pytest 13/13 ✓` 全部通过。 --- ## 🔧 实战:接真实相机 把 `fake_camera` 换成 `usb_cam` 或 `realsense2_camera`: ```bash sudo apt install ros-humble-usb-cam ros2 launch usb_cam camera.launch.py ``` 下游 `image_processor` 完全不用改 — 它只关心 `/image_raw` 是不是 `sensor_msgs/Image`。 --- ## 📚 深入学习 - [sensor_msgs/Image 文档](https://docs.ros.org/en/humble/p/sensor_msgs/msg/Image.html) - [cv_bridge 教程](https://wiki.ros.org/cv_bridge/Tutorials/ConvertingBetweenROSImagesAndOpenCVImagesPython) --- ## ⏭️ 下一个包 继续学 **[cpp_robot_tf2](../cpp_robot_tf2/README.md)** — TF2 坐标变换 + URDF 机械臂模型(机器人入门必学)。 --- ## 📍 学习路径导航 | ⏮ 上一个 | 🏠 当前位置 | ⏭ 下一个 | |---|---|---| | [cpp_custom_interface — C++ 自定义接口](../cpp_custom_interface/README.md) | **py_vision_demo — Python 图像** | [cpp_robot_tf2 — C++ TF2 + URDF](../cpp_robot_tf2/README.md) | 📍 完整 12 包学习顺序见 [主 README](../../README.md#-12-包推荐学习顺序)