MediaPipe 与 ROS 集成：机器人动作交互系统部署实战

服务默认开启 WebUI 页面，可直接用于测试。

3.2 WebUI 功能操作指南

进入页面后执行以下步骤：

1. 上传图片：点击'Choose File'选择一张包含人物的 JPG/PNG 图像
2. 提交处理：点击 "Upload & Detect" 按钮
3. 查看结果：
   • 原图上会叠加绘制火柴人骨架
   • 红点 表示检测到的关键点
   • 白线 连接相邻关节点，构成身体结构
4. 获取数据（可选）：
   • 页面下方可查看 JSON 格式的原始关键点数据
   • 支持下载带骨架标注的图像

提示：建议使用正面站立、光照良好、背景简洁的照片以获得最佳检测效果

4. ROS 系统集成实践

4.1 架构设计思路

为了将 MediaPipe 的姿态识别能力融入机器人控制系统，我们设计如下集成架构：

[Camera] ↓ (raw image) [MediaPipe Node] → 检测关键点 → 发布 /skeleton_pose Topic ↓ (visualization) [WebUI Server] ↓ (pose data) [ROS Master] ← 其他 Nodes（如 Motion Planner）

核心思想是将 MediaPipe 封装为一个 ROS Node，持续订阅摄像头图像流，完成姿态估计后发布标准化的消息。

4.2 创建 ROS Package

cd ~/catkin_ws/src
catkin_create_pkg mediapipe_ros rospy cv_bridge sensor_msgs geometry_msgs
cd ..
catkin_make

4.3 关键代码实现

核心节点：pose_estimator.py

这里需要仔细处理缩进，确保是合法的 Python 代码。

#!/usr/bin/env python
import rospy
import cv2
import numpy as np
from cv_bridge import CvBridge
from sensor_msgs.msg import Image
from geometry_msgs.msg import PointStamped, Vector3
from std_msgs.msg import Header, String
import mediapipe as mp
import json

class PoseEstimator:
    def __init__(self):
        self.bridge = CvBridge()
        self.mp_pose = mp.solutions.pose
        self.pose = self.mp_pose.Pose(
            static_image_mode=False,
            model_complexity=1,  # 平衡速度与精度
            enable_segmentation=False,
            min_detection_confidence=0.5,
            min_tracking_confidence=0.5
        )
        # Subscribers
        self.image_sub = rospy.Subscriber("/camera/image_raw", Image, self.image_callback)
        # Publishers
        self.keypoint_pub = rospy.Publisher("/skeleton_pose", PointStamped, queue_size=10)
        self.skeleton_pub = rospy.Publisher("/full_skeleton", String, queue_size=10)
        rospy.loginfo("Pose Estimator Node Initialized")

    def image_callback(self, msg):
        try:
            cv_image = self.bridge.imgmsg_to_cv2(msg, "bgr8")
        except Exception as e:
            rospy.logerr(f"Image conversion error: {e}")
            return

        rgb_image = cv2.cvtColor(cv_image, cv2.COLOR_BGR2RGB)
        results = self.pose.process(rgb_image)

        if results.pose_landmarks:
            h, w, _ = cv_image.shape
            keypoints = []
            for i, landmark in enumerate(results.pose_landmarks.landmark):
                px = int(landmark.x * w)
                py = int(landmark.y * h)
                pz = landmark.z * w  # 相对深度
                vis = landmark.visibility

                # 发布每个关键点（示例：仅发布鼻子）
                if i == 0:  # 鼻子索引为 0
                    point_msg = PointStamped()
                    point_msg.header = Header(stamp=rospy.Time.now(), frame_id="camera_link")
                    point_msg.point.x = px
                    point_msg.point.y = py
                    point_msg.point.z = pz
                    self.keypoint_pub.publish(point_msg)

                keypoints.append({
                    "id": i,
                    "x": px,
                    "y": py,
                    "z": pz,
                    "visibility": float(vis)
                })

            # 发布完整骨架（JSON 字符串格式）
            skeleton_str = json.dumps(keypoints)
            str_msg = String()
            str_msg.data = skeleton_str
            self.skeleton_pub.publish(str_msg)

            # 绘制骨架（可选：回传给 WebUI）
            mp.solutions.drawing_utils.draw_landmarks(
                cv_image, results.pose_landmarks, self.mp_pose.POSE_CONNECTIONS)

            # 可视化反馈（调试用）
            cv2.imshow("MediaPipe Pose", cv_image)
            cv2.waitKey(1)

if __name__ == '__main__':
    rospy.init_node('mediapipe_pose_node', anonymous=True)
    node = PoseEstimator()
    rospy.spin()

4.4 编译与运行

确保安装依赖：

pip install mediapipe opencv-python rospy cv_bridge

赋予脚本可执行权限并运行：

chmod +x pose_estimator.py
roscore &
rosrun mediapipe_ros pose_estimator.py

4.5 数据订阅测试

新开终端运行：

rostopic echo /skeleton_pose

你应该能看到类似以下输出：

header:
  seq: 123
  stamp:
    secs: 1712345678
    nsecs: 901234567
  frame_id: "camera_link"
point:
  x: 320.0
  y: 180.0
  z: 45.2

5. 动作识别与机器人控制联动

5.1 简单动作判断逻辑

我们可以基于关键点之间的几何关系实现基本动作识别。例如判断'举手'动作：

def is_hand_raised(landmarks):
    """判断右手是否举起"""
    shoulder_y = landmarks[12]['y']  # 右肩
    wrist_y = landmarks[16]['y']     # 右手腕
    return wrist_y < shoulder_y       # 手腕高于肩膀即视为举起

5.2 控制机器人移动示例

假设你的机器人支持 /cmd_vel 消息控制：

from geometry_msgs.msg import Twist

class RobotController:
    def __init__(self):
        self.cmd_pub = rospy.Publisher('/cmd_vel', Twist, queue_size=10)

    def move_forward(self):
        twist = Twist()
        twist.linear.x = 0.2  # 前进速度
        self.cmd_pub.publish(twist)

在主循环中加入条件判断：

if is_hand_raised(keypoints):
    controller.move_forward()

即可实现'抬手前进'的自然交互体验。

6. 总结

6.1 实践经验总结

本文完整展示了如何将 MediaPipe 与 ROS 集成，打造一套可用于真实机器人系统的动作交互框架。核心收获包括：

• ✅ 成功部署了一个稳定、高效的本地化姿态检测服务
• ✅ 实现了 WebUI 与 ROS 双通道输出，兼顾可视化与工程集成
• ✅ 掌握了将 AI 视觉能力接入机器人控制系统的标准模式
• ✅ 构建了'感知→决策→执行'的闭环人机交互原型

6.2 最佳实践建议

1. 性能优化：对于嵌入式设备，建议降低输入图像分辨率至 640x480 并启用 model_complexity=0
2. 坐标转换：若需精确空间定位，请标定相机内参并使用 solvePnP 计算世界坐标
3. 异常处理：增加超时机制和重连逻辑，防止长时间无响应
4. 安全策略：避免直接映射复杂动作指令，应加入确认机制或限速保护