使用mediapip 检测pose 并作为一个服务

代码

import uvicorn
from fastapi import FastAPI, HTTPException
import cv2
import mediapipe as mp
from pydantic import BaseModel

app = FastAPI()


# 创建一个模型来序列化姿态数据
class PoseData(BaseModel):
    landmarks: list



# 初始化MediaPipe的姿态估计模型
mp_pose = mp.solutions.pose
pose = mp_pose.Pose()

# 初始化摄像头
cap = cv2.VideoCapture("q888.mp4")


@app.get("/pose")
def get_pose():
    # 从摄像头读取一帧
    success, image = cap.read()
    if not success:
        raise HTTPException(status_code=500, detail="Failed to capture image")

    # 转换为RGB格式，因为MediaPipe需要RGB图像
    image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)

    # 处理图像以获取姿态数据
    results = pose.process(image)

    # 如果没有检测到姿态，返回错误
    if not results.pose_landmarks:
        raise HTTPException(status_code=404, detail="No pose detected")

    # 获取所有地标的位置和ID
    landmarks = []

    for landmark in results.pose_landmarks.landmark:
        landmarks.append({
            'x': landmark.x,
            'y': landmark.y,
            'z': landmark.z,
            'visibility': landmark.visibility
        })


    # 返回姿态数据
    return PoseData(landmarks=landmarks)




if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

    # 释放资源
    cap.release()
    pose.close()

解析

当然，以下是该代码的逐行解析：

import uvicorn
from fastapi import FastAPI, HTTPException
import cv2
import mediapipe as mp
from pydantic import BaseModel

• 导入必要的库：uvicorn 用于运行 FastAPI 应用，FastAPI 是一个现代、快速（高性能）的 Web 框架，用于构建 API，HTTPException 用于处理 HTTP 错误，cv2 是 OpenCV 的 Python 绑定库，用于图像处理，mediapipe 是一个跨平台的机器学习解决方案库，用于构建各种感知增强应用，BaseModel 用于数据验证和设置类型提示。

app = FastAPI()

• 创建一个 FastAPI 实例。

class PoseData(BaseModel):
    landmarks: list

• 定义一个 Pydantic 模型 PoseData，用于序列化和反序列化姿态数据。该模型包含一个名为 landmarks 的列表字段。

mp_pose = mp.solutions.pose
pose = mp_pose.Pose()

• 初始化 MediaPipe 的姿态估计模型。

cap = cv2.VideoCapture("q888.mp4")

• 初始化摄像头，这里使用的是视频文件 q888.mp4 而不是实际的摄像头。

@app.get("/pose")
def get_pose():

• 定义一个路由 /pose，这是一个 GET 请求，返回姿态数据。

    success, image = cap.read()
    if not success:
        raise HTTPException(status_code=500, detail="Failed to capture image")

• 从摄像头读取一帧图像，如果读取失败，则抛出一个 HTTP 500 内部服务器错误。

    image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)

• 将图像从 BGR 格式转换为 RGB 格式，因为 MediaPipe 需要 RGB 图像。

    results = pose.process(image)

• 使用 MediaPipe 的姿态估计模型处理图像以获取姿态数据。

    if not results.pose_landmarks:
        raise HTTPException(status_code=404, detail="No pose detected")

• 如果没有检测到姿态，则抛出一个 HTTP 404 未找到错误。

    landmarks = []
    for landmark in results.pose_landmarks.landmark:
        landmarks.append({
            'x': landmark.x,
            'y': landmark.y,
            'z': landmark.z,
            'visibility': landmark.visibility
        })

• 遍历检测结果中的所有地标，并将它们的位置和可见性添加到 landmarks 列表中。

    return PoseData(landmarks=landmarks)

• 返回序列化的姿态数据。

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

• 如果该脚本作为主程序运行，则启动 FastAPI 应用，监听所有网络接口的 8000 端口。

    cap.release()
    pose.close()

• 释放摄像头资源和关闭 MediaPipe 的姿态估计模型。这段代码实际上不会执行，因为 uvicorn.run 是一个阻塞调用，它不会继续执行之后的代码。正确的方法是在一个事件循环中处理资源的释放。