MediaPipe BlazePose 33关键点实战：Python+OpenCV 实现深蹲计数与角度可视化

MediaPipe BlazePose 33关键点实战：Python+OpenCV 实现深蹲计数与角度可视化

在健身领域，准确的动作计数和姿势分析是提升训练效果的关键。传统方法依赖人工观察或穿戴式设备，而基于计算机视觉的解决方案正逐渐成为更高效的选择。本文将带你从零实现一个基于BlazePose 33个关键点的深蹲分析系统，通过Python和OpenCV实时计算关节角度并可视化训练数据。

1. 环境配置与基础框架搭建

首先需要准备开发环境。推荐使用Python 3.8+版本，并创建虚拟环境隔离依赖：

python -m venv pose-env source pose-env/bin/activate # Linux/Mac pose-env\Scripts\activate # Windows

安装核心依赖库：

pip install mediapipe opencv-python numpy matplotlib

基础代码框架包含三个主要组件：视频输入处理、姿态检测引擎和可视化输出。创建一个squat_analyzer.py文件，初始化基础结构：

import cv2 import mediapipe as mp import numpy as np from matplotlib import pyplot as plt class SquatAnalyzer: def __init__(self): self.mp_pose = mp.solutions.pose self.pose = self.mp_pose.Pose( static_image_mode=False, model_complexity=1, smooth_landmarks=True, min_detection_confidence=0.5, min_tracking_confidence=0.5) self.mp_drawing = mp.solutions.drawing_utils self.cap = None self.angle_history = [] def process_frame(self, frame): # 待实现的核心处理逻辑 pass def run(self, video_source=0): self.cap = cv2.VideoCapture(video_source) while self.cap.isOpened(): ret, frame = self.cap.read() if not ret: break processed_frame = self.process_frame(frame) cv2.imshow('Squat Analysis', processed_frame) if cv2.waitKey(1) & 0xFF == ord('q'): break self.cap.release() cv2.destroyAllWindows() if __name__ == "__main__": analyzer = SquatAnalyzer() analyzer.run() # 默认使用摄像头

提示：测试时可通过传入视频路径替代摄像头输入，如analyzer.run("squat_demo.mp4")

2. 关键点解析与角度计算

BlazePose的33个关键点中，对深蹲分析最重要的是下肢关节。我们需要特别关注以下点位：

角度计算采用向量夹角公式。以下函数计算三个关键点形成的夹角：

def calculate_angle(a, b, c): """ 计算三个点之间的夹角(度) :param a: 点A坐标 (x,y) :param b: 点B坐标 (中心点) :param c: 点C坐标 :return: 角度值(0-180) """ a = np.array(a) b = np.array(b) c = np.array(c) ba = a - b bc = c - b cosine_angle = np.dot(ba, bc) / (np.linalg.norm(ba) * np.linalg.norm(bc)) angle = np.arccos(cosine_angle) return np.degrees(angle)

更新process_frame方法实现实时角度计算：

def process_frame(self, frame): # 转换颜色空间并处理 image = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) results = self.pose.process(image) # 绘制关键点连接 annotated_image = frame.copy() self.mp_drawing.draw_landmarks( annotated_image, results.pose_landmarks, self.mp_pose.POSE_CONNECTIONS) if results.pose_landmarks: landmarks = results.pose_landmarks.landmark # 获取关键点坐标 left_hip = [landmarks[23].x, landmarks[23].y] right_hip = [landmarks[24].x, landmarks[24].y] left_knee = [landmarks[25].x, landmarks[25].y] right_knee = [landmarks[26].x, landmarks[26].y] left_ankle = [landmarks[27].x, landmarks[27].y] right_ankle = [landmarks[28].x, landmarks[28].y] # 计算膝盖角度 left_knee_angle = calculate_angle(left_hip, left_knee, left_ankle) right_knee_angle = calculate_angle(right_hip, right_knee, right_ankle) # 显示角度信息 cv2.putText(annotated_image, f"L Knee: {int(left_knee_angle)}", tuple(np.multiply(left_knee, [640, 480]).astype(int)), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (255,255,255), 2) # 记录角度历史用于分析 self.angle_history.append((left_knee_angle + right_knee_angle)/2) return annotated_image

3. 深蹲计数逻辑实现

准确的深蹲计数需要定义完整的动作周期。我们通过膝关节角度变化来识别：

1. 起始位置：膝盖角度 > 150度
2. 下蹲阶段：角度持续减小至 < 90度
3. 上升阶段：角度恢复至 > 150度
4. 完整计数：完成1→2→3→1的循环

实现计数状态机：

class SquatAnalyzer: def __init__(self): # ...原有初始化... self.squat_count = 0 self.squat_state = "up" # 初始状态 def update_counter(self, knee_angle): if self.squat_state == "up" and knee_angle < 90: self.squat_state = "down" elif self.squat_state == "down" and knee_angle > 150: self.squat_count += 1 self.squat_state = "up" def process_frame(self, frame): # ...原有处理逻辑... if results.pose_landmarks: # ...角度计算... avg_angle = (left_knee_angle + right_knee_angle)/2 self.update_counter(avg_angle) # 显示计数 cv2.putText(annotated_image, f"Squats: {self.squat_count}", (50, 50), cv2.FONT_HERSHEY_SIMPLEX, 1, (0,255,0), 2) return annotated_image

优化计数逻辑，增加动作质量检测：

def update_counter(self, knee_angle): if self.squat_state == "up": if knee_angle < 90: # 进入下蹲 self.squat_state = "down" self.min_angle = knee_angle else: self.min_angle = None elif self.squat_state == "down": if knee_angle < (self.min_angle or 180): self.min_angle = knee_angle if knee_angle > 150: # 完成站立 if self.min_angle and self.min_angle < 80: # 深度达标 self.squat_count += 1 self.squat_state = "up" else: cv2.putText(annotated_image, "Squat too shallow!", (50, 100), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0,0,255), 2)

4. 实时数据可视化

为了更直观地展示训练数据，我们添加实时角度曲线和统计面板：

def __init__(self): # ...原有初始化... self.fig, (self.ax1, self.ax2) = plt.subplots(2, 1, figsize=(8,6)) plt.ion() # 开启交互模式 def update_plots(self): # 清空原有图形 self.ax1.clear() self.ax2.clear() # 角度变化曲线 if len(self.angle_history) > 0: self.ax1.plot(self.angle_history, 'b-') self.ax1.set_title('Knee Angle Variation') self.ax1.set_ylim(0, 180) self.ax1.axhline(y=90, color='r', linestyle='--') # 计数统计 self.ax2.text(0.1, 0.5, f"Total Squats: {self.squat_count}", fontsize=14) self.ax2.axis('off') plt.tight_layout() plt.pause(0.01) def process_frame(self, frame): # ...原有处理... self.update_plots() return annotated_image

优化可视化效果，添加运动指标：

def update_plots(self): # ...角度曲线... # 运动指标面板 stats_text = [ f"Total Squats: {self.squat_count}", f"Current Angle: {self.angle_history[-1]:.1f}°" if self.angle_history else "", f"Depth {'✓' if (self.min_angle and self.min_angle < 80) else '✗'}" ] self.ax2.text(0.1, 0.8, "\n".join(stats_text), fontsize=12, verticalalignment='top') self.ax2.axis('off')

5. 系统优化与高级功能

5.1 多角度同步分析

同时监测髋关节和膝关节角度，提供更全面的动作分析：

def process_frame(self, frame): # ...原有处理... if results.pose_landmarks: # 髋关节角度计算 left_hip_angle = calculate_angle( [landmarks[11].x, landmarks[11].y], # 左肩 left_hip, left_knee) # 添加到可视化 cv2.putText(annotated_image, f"L Hip: {int(left_hip_angle)}", tuple(np.multiply(left_hip, [640, 480]).astype(int)), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (255,255,0), 2) # 更新曲线图 self.ax1.plot(self.angle_history, 'b-', label='Knee') self.ax1.plot(self.hip_angle_history, 'g-', label='Hip') self.ax1.legend()

5.2 姿势矫正提示

检测常见错误姿势并提供实时反馈：

def check_posture(self, landmarks): warnings = [] # 膝盖内扣检测 left_knee = landmarks[25].x right_knee = landmarks[26].x if left_knee > right_knee + 0.05: # 左膝过度内扣 warnings.append("Left knee collapsing inward") # 背部弯曲检测 shoulder = (landmarks[11].y + landmarks[12].y)/2 hip = (landmarks[23].y + landmarks[24].y)/2 if shoulder - hip < 0.1: # 上身前倾过多 warnings.append("Keep your chest up") return warnings # 在process_frame中调用 warnings = self.check_posture(landmarks) for i, warning in enumerate(warnings): cv2.putText(annotated_image, warning, (50, 150+i*30), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0,0,255), 2)

5.3 数据持久化与报告生成

添加训练数据记录功能：

def save_session_data(self): timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"squat_session_{timestamp}.csv" with open(filename, 'w') as f: f.write("time,angle\n") for i, angle in enumerate(self.angle_history): f.write(f"{i/30:.2f},{angle}\n") # 假设30FPS print(f"Session data saved to {filename}") # 在程序退出时调用 def run(self, video_source=0): try: # ...原有运行逻辑... finally: if len(self.angle_history) > 0: self.save_session_data()

6. 部署与性能优化

6.1 多线程处理

使用生产者-消费者模式提升实时性：

from threading import Thread from queue import Queue class VideoStream: def __init__(self, src=0): self.stream = cv2.VideoCapture(src) self.stopped = False self.Q = Queue(maxsize=128) def start(self): Thread(target=self.update, args=()).start() return self def update(self): while True: if self.stopped: return ret, frame = self.stream.read() if not ret: self.stop() return if not self.Q.full(): self.Q.put(frame) def read(self): return self.Q.get() def stop(self): self.stopped = True # 修改SquatAnalyzer的run方法 def run(self, video_source=0): vs = VideoStream(video_source).start() while True: frame = vs.read() processed_frame = self.process_frame(frame) cv2.imshow('Squat Analysis', processed_frame) if cv2.waitKey(1) & 0xFF == ord('q'): vs.stop() break cv2.destroyAllWindows()

6.2 模型优化技巧

通过调整MediaPipe参数提升性能：

self.pose = self.mp_pose.Pose( static_image_mode=False, model_complexity=0, # 使用轻量级模型 smooth_landmarks=True, enable_segmentation=False, min_detection_confidence=0.5, min_tracking_confidence=0.5)

对于低性能设备，可降低输入分辨率：

def process_frame(self, frame): # 缩小处理分辨率 small_frame = cv2.resize(frame, (0,0), fx=0.5, fy=0.5) results = self.pose.process(small_frame) # 结果坐标需要转换回原始尺寸 if results.pose_landmarks: for landmark in results.pose_landmarks.landmark: landmark.x *= 2 landmark.y *= 2

7. 应用扩展思路

7.1 多动作识别系统

扩展框架支持多种健身动作：

class FitnessAnalyzer: def __init__(self): self.analyzers = { 'squat': SquatAnalyzer(), 'pushup': PushupAnalyzer(), 'lunge': LungeAnalyzer() } self.current_mode = 'squat' def switch_mode(self, mode): if mode in self.analyzers: self.current_mode = mode def process_frame(self, frame): return self.analyzers[self.current_mode].process_frame(frame)

7.2 3D姿态可视化

结合MediaPipe的Z坐标数据实现三维展示：

def plot_3d_skeleton(landmarks): fig = plt.figure() ax = fig.add_subplot(111, projection='3d') xs = [lm.x for lm in landmarks] ys = [lm.y for lm in landmarks] zs = [-lm.z for lm in landmarks] # 转换Z轴方向 # 绘制关键点 ax.scatter(xs, ys, zs) # 绘制骨骼连接 for connection in mp.solutions.pose.POSE_CONNECTIONS: start = connection[0] end = connection[1] ax.plot([xs[start], xs[end]], [ys[start], ys[end]], [zs[start], zs[end]], 'r-') plt.show()

7.3 云端数据同步

实现训练数据的上传与分析：

import requests def upload_session_data(user_id, session_data): api_url = "https://your-api-endpoint.com/sessions" payload = { "user_id": user_id, "exercise": "squat", "data": session_data } try: response = requests.post(api_url, json=payload) if response.status_code == 200: print("Data uploaded successfully") return response.json() except Exception as e: print(f"Upload failed: {str(e)}")

8. 实际应用案例

8.1 家庭健身助手

将系统集成到智能镜中，打造交互式健身体验：

class SmartMirror: def __init__(self): self.analyzer = SquatAnalyzer() self.mirror_display = MirrorDisplay() def run(self): while True: frame = self.mirror_display.get_frame() processed = self.analyzer.process_frame(frame) # 在镜面叠加分析结果 self.mirror_display.show( processed, self.analyzer.angle_history, self.analyzer.squat_count)

8.2 健身房动作分析终端

部署到固定设备供会员使用：

class GymTerminal: def __init__(self): self.analyzer = FitnessAnalyzer() self.user_db = UserDatabase() def start_session(self, user_id): user = self.user_db.get_user(user_id) print(f"Welcome {user.name}! Starting your workout...") cap = cv2.VideoCapture(0) while True: ret, frame = cap.read() processed = self.analyzer.process_frame(frame) cv2.imshow('Workout Analysis', processed) if cv2.waitKey(1) & 0xFF == ord('q'): break cap.release() cv2.destroyAllWindows() # 保存训练数据 self.user_db.save_session( user_id, self.analyzer.get_session_data())

8.3 康复训练监控

为物理治疗患者提供动作质量反馈：

class RehabMonitor: def __init__(self, patient_id): self.patient_id = patient_id self.analyzer = SquatAnalyzer() self.therapist = TherapistNotifier() def check_range_of_motion(self): if len(self.analyzer.angle_history) < 10: return current_range = max(self.analyzer.angle_history[-10:]) - min(self.analyzer.angle_history[-10:]) if current_range < 60: # 活动范围不足 self.therapist.notify( self.patient_id, "Limited range of motion detected")