代码
import uvicorn
from fastapi import FastAPI, HTTPException
import cv2
import mediapipe as mp
from pydantic import BaseModelapp = 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")# 获取所有地标的位置和IDlandmarks = []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
是一个阻塞调用,它不会继续执行之后的代码。正确的方法是在一个事件循环中处理资源的释放。