基于MediaPipe与LSTM的人体动作识别:从姿态检测到时序理解实战
2026/9/5 23:22:16 网站建设 项目流程

简介:本资源是一套基于MediaPipe的人体姿态识别与动作分类Python实现方案,面向计算机、电子信息、人工智能等专业的本科生及研究生,适用于课程设计、期末大作业与毕业设计等实践场景,解决从关键点提取到时序动作判别的一体化技术问题。压缩包共139个文件,含120个npy格式的标准化姿态特征序列(用于DTW匹配与LSTM训练)、8个核心Python脚本(涵盖姿态检测、特征预处理、DTW距离计算、LSTM建模与实时识别)、8个MP4动作演示视频(含左右手、单双臂等典型动作样本),以及模型权重h5、项目说明md和v2配置文件,整体大小为11.03MB。目前已有616人学习下载。用户可直接运行源码完成端到端流程:MediaPipe实时捕获2D/3D关节点→DTW对齐不同节奏动作模板→LSTM网络进行时序分类,同时获得完整数据预处理逻辑、双算法对比结构及可扩展的模型接口,具备良好的教学参考性与工程复用价值。

1. 项目缘起:从姿态“看见”到动作“理解”的跨越

最近在做一个智能健身指导相关的项目,核心需求是让系统能“看懂”用户在做什么动作,比如深蹲、开合跳、或者打一套太极拳。最开始,我理所当然地想到了MediaPipe这个神器。它确实强大,打开摄像头,几行代码就能实时获取人体33个关键点的三维坐标,屏幕上的人体骨架图流畅地跟着你动,感觉已经成功了一大半。

但很快我就发现,事情没那么简单。MediaPipe输出的是一帧一帧的“姿态”,是静态的“点”。而“动作”是连续的、动态的“序列”。系统能稳定地“看见”我在深蹲,但它怎么“知道”我深蹲的幅度够不够标准?速度是不是太快?动作轨迹有没有变形?这才是真正有价值的部分。这就好比给你一连串孤立的汉字(姿态),你需要理解它们组成的句子(动作)甚至段落(行为)的含义。

于是,问题就变成了:如何将MediaPipe提取的、随时间变化的关键点序列,转化成一个机器能够识别和评价的“动作标签”?这就是我这次项目的核心:基于MediaPipe的姿态数据流,结合动态时间规整(DTW)和长短期记忆网络(LSTM),构建一个端到端的人体动作识别系统。DTW用来解决同类动作因执行速度不同导致的时间轴对齐问题,LSTM则擅长捕捉时间序列中的长期依赖关系,两者结合,堪称处理这类时序分类问题的“黄金搭档”。下面,我就把自己从环境搭建、数据处理、模型构建到训练优化的完整过程,以及踩过的无数个坑,毫无保留地分享出来。

2. 环境搭建与MediaPipe实时姿态提取

工欲善其事,必先利其器。这个项目对环境的依赖不算复杂,但有几个版本匹配的细节至关重要,一步错可能导致后面步步报错。

2.1 核心依赖库选型与安装

我的基础环境是Python 3.8,这是一个在兼容性和稳定性上比较折中的版本。太老的版本可能不支持一些新库,太新的版本(如3.11+)有时会遇到预编译轮子(wheel)缺失的问题。使用Anaconda或Miniconda创建一个独立的虚拟环境是强烈推荐的,可以避免包冲突。

# 创建并激活虚拟环境(以conda为例) conda create -n action_recognition python=3.8 conda activate action_recognition

接下来安装核心库。这里面的版本是我经过多次测试后确定的稳定组合:

pip install mediapipe==0.10.0 pip install opencv-python==4.8.1.78 pip install tensorflow==2.10.0 # 注意:2.10是最后一个官方支持Windows Native的版本,后续版本需通过WSL pip install scikit-learn==1.2.2 pip install matplotlib==3.7.1 pip install numpy==1.23.5

注意:TensorFlow的版本是个大坑。如果你在Windows上,强烈建议使用2.10.0或更早版本,或者直接使用WSL2。TF 2.11+ 在Windows上需要手动编译,极其麻烦。MediaPipe 0.10.0 是一个功能稳定且API清晰的版本,够用且省心。

2.2 编写MediaPipe姿态提取模块

MediaPipe的使用非常直观。我们需要初始化姿态检测模型,并编写一个函数来处理每一帧图像,提取出我们需要的33个关键点坐标(x, y, z, visibility)。

import cv2 import mediapipe as mp import numpy as np class PoseDetector: def __init__(self, static_image_mode=False, model_complexity=1, smooth_landmarks=True, enable_segmentation=False, smooth_segmentation=True, min_detection_confidence=0.5, min_tracking_confidence=0.5): """ 初始化MediaPipe Pose模型。 :param static_image_mode: 是否静态图片模式。False为视频流模式,会利用前后帧信息优化。 :param model_complexity: 模型复杂度(0,1,2)。越高越准,但越慢。 :param min_detection_confidence: 检测置信度阈值。 :param min_tracking_confidence: 跟踪置信度阈值。 """ self.mp_pose = mp.solutions.pose self.pose = self.mp_pose.Pose( static_image_mode=static_image_mode, model_complexity=model_complexity, smooth_landmarks=smooth_landmarks, min_detection_confidence=min_detection_confidence, min_tracking_confidence=min_tracking_confidence ) self.mp_drawing = mp.solutions.drawing_utils def find_pose(self, img, draw=True): """ 在图像中检测人体姿态。 :param img: BGR格式的图像。 :param draw: 是否在图像上绘制骨架。 :return: 处理后的图像,以及关键点坐标列表(归一化后的x, y, z, visibility)。 """ img_rgb = cv2.cvtColor(img, cv2.COLOR_BGR2RGB) img_rgb.flags.writeable = False # 提升性能 results = self.pose.process(img_rgb) img_rgb.flags.writeable = True img = cv2.cvtColor(img_rgb, cv2.COLOR_RGB2BGR) landmarks_list = [] if results.pose_landmarks: if draw: self.mp_drawing.draw_landmarks( img, results.pose_landmarks, self.mp_pose.POSE_CONNECTIONS) # 提取所有33个关键点信息 for id, lm in enumerate(results.pose_landmarks.landmark): h, w, c = img.shape # 保存归一化坐标和可见度 landmarks_list.append([lm.x, lm.y, lm.z, lm.visibility]) return img, np.array(landmarks_list) if landmarks_list else None # 使用示例 if __name__ == "__main__": detector = PoseDetector() cap = cv2.VideoCapture(0) # 打开摄像头 while cap.isOpened(): success, img = cap.read() if not success: break img, landmarks = detector.find_pose(img) if landmarks is not None: # landmarks 是一个 (33, 4) 的numpy数组 print(f"检测到{len(landmarks)}个关键点,鼻尖坐标:{landmarks[0]}") cv2.imshow('Pose Detection', img) if cv2.waitKey(1) & 0xFF == ord('q'): break cap.release() cv2.destroyAllWindows()

这段代码运行后,你应该能看到一个实时的人体骨架图。landmarks变量就是我们后续所有工作的数据基石。它是一个形状为(33, 4)的数组,每一行代表一个关键点(0是鼻子,11是左肩,12是右肩等),每一列分别是x, y, z(归一化到[0,1]的坐标)和visibility(可见度,0~1)。

3. 数据工程:从原始坐标到模型可用的序列

拿到原始数据只是第一步,直接把它们扔给模型效果会很差。我们需要进行一系列的数据预处理,构建出干净、有效、模型友好的数据集。

3.1 动作定义与数据采集脚本

首先,明确你要识别的动作。比如,我定义了四个动作:“深蹲”“开合跳”“高抬腿”“休息”(即无特定动作)。每个动作需要采集足够多的样本。

我写了一个简单的数据采集脚本,它会为每个动作创建一个文件夹,并以动作类别_序列编号.npy的格式保存数据。

import os import time # 配置 ACTIONS = ['squat', 'jumping_jack', 'high_knee', 'rest'] DATA_PATH = './MP_Data' # 数据保存根目录 SEQUENCE_LENGTH = 30 # 每个动作序列的帧数 # 创建文件夹 for action in ACTIONS: action_path = os.path.join(DATA_PATH, action) os.makedirs(action_path, exist_ok=True) def collect_data(action_name, num_sequences=30): """ 采集指定动作的数据。 :param action_name: 动作名称,对应ACTIONS中的一项。 :param num_sequences: 需要采集的序列数量。 """ detector = PoseDetector() cap = cv2.VideoCapture(0) print(f'开始采集动作 [{action_name}] 的数据。按‘q’提前终止。') for sequence in range(num_sequences): print(f'准备采集第 {sequence+1} 个序列,3秒后开始...') for countdown in range(3, 0, -1): print(f'{countdown}...') time.sleep(1) sequence_data = [] # 用于存储一个序列的所有帧 frame_count = 0 while frame_count < SEQUENCE_LENGTH: success, img = cap.read() if not success: break img, landmarks = detector.find_pose(img, draw=True) if landmarks is not None: sequence_data.append(landmarks) frame_count += 1 else: print("未检测到人体,请调整位置。") cv2.putText(img, f'Collecting: {action_name.upper()} Seq#{sequence+1} Frame#{frame_count}', (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 0), 2) cv2.imshow('Data Collection', img) if cv2.waitKey(10) & 0xFF == ord('q'): cap.release() cv2.destroyAllWindows() return # 保存一个序列 save_path = os.path.join(DATA_PATH, action_name, f'{sequence}.npy') np.save(save_path, np.array(sequence_data)) # 保存为 (30, 33, 4) 的数组 print(f'序列 {sequence+1} 已保存至 {save_path}') cap.release() cv2.destroyAllWindows() # 示例:采集“深蹲”动作的30个序列 # collect_data('squat', 30)

实操心得:采集数据时,尽量让不同身高、体型、穿衣风格的人参与,并在不同的光照、背景环境下进行,以增强模型的泛化能力。每个序列的帧数(SEQUENCE_LENGTH)需要根据动作的周期来定。对于“深蹲”这种慢速动作,30帧(约1秒,30FPS)可能只记录了半个蹲起,可以增加到60或90帧。而对于“高抬腿”,30帧可能包含了多个周期。一个核心原则是:一个序列最好能完整包含1-3个标准动作周期。

3.2 关键预处理步骤:归一化、去中心化与序列对齐

原始数据不能直接用,主要有三个问题:1) 坐标是相对于图像帧的,人距离摄像头远近会影响坐标绝对值;2) 每个人的身高、臂展不同;3) 同一动作,不同人做起来快慢不同。

步骤一:基于躯干的相对坐标转换这是最关键的一步,目的是消除人体在画面中位置和绝对尺寸的影响。我们以臀部中心点(假设为左髋和右髋的中点)为新的坐标原点,并以左肩到右肩的距离作为参考长度,对所有关键点坐标进行归一化。

def extract_keypoints_from_landmarks(landmarks): """ 将MediaPipe的33个关键点转换为模型可用的特征向量,并进行相对坐标转换。 :param landmarks: (33, 4) 的数组,包含x,y,z,visibility。 :return: 处理后的特征向量,形状为 (99,) 或 (132,)。这里选择 (99,),即x,y,z,忽略visibility。 """ # 选取用于参考的躯干点索引(MediaPipe索引) LEFT_HIP = 23 RIGHT_HIP = 24 LEFT_SHOULDER = 11 RIGHT_SHOULDER = 12 # 计算臀部中心点 hip_center_x = (landmarks[LEFT_HIP, 0] + landmarks[RIGHT_HIP, 0]) / 2 hip_center_y = (landmarks[LEFT_HIP, 1] + landmarks[RIGHT_HIP, 1]) / 2 hip_center_z = (landmarks[LEFT_HIP, 2] + landmarks[RIGHT_HIP, 2]) / 2 # 计算肩宽作为参考长度(欧氏距离) shoulder_width = np.sqrt( (landmarks[RIGHT_SHOULDER, 0] - landmarks[LEFT_SHOULDER, 0]) ** 2 + (landmarks[RIGHT_SHOULDER, 1] - landmarks[LEFT_SHOULDER, 1]) ** 2 + (landmarks[RIGHT_SHOULDER, 2] - landmarks[LEFT_SHOULDER, 2]) ** 2 ) # 防止除零,给一个极小值 scale = shoulder_width if shoulder_width > 1e-6 else 1e-6 processed_keypoints = [] for i in range(len(landmarks)): # 去中心化:减去臀部中心 rel_x = landmarks[i, 0] - hip_center_x rel_y = landmarks[i, 1] - hip_center_y rel_z = landmarks[i, 2] - hip_center_z # 尺度归一化:除以肩宽 rel_x /= scale rel_y /= scale rel_z /= scale processed_keypoints.extend([rel_x, rel_y, rel_z]) # 注意:这里舍弃了visibility,因为它不是空间特征。你也可以选择保留。 return np.array(processed_keypoints)

经过这个处理,无论人站在画面何处,无论身高如何,同一个标准动作的processed_keypoints在数值分布上都会非常接近。

步骤二:构建数据集遍历所有保存的.npy文件,加载序列,对每一帧应用上述转换,并打上标签。

import os from sklearn.model_selection import train_test_split def load_data(data_path, sequence_length=30): """ 加载所有数据并构建特征X和标签y。 """ label_map = {label: num for num, label in enumerate(ACTIONS)} sequences, labels = [], [] for action in ACTIONS: action_dir = os.path.join(data_path, action) if not os.path.exists(action_dir): continue for sequence_file in os.listdir(action_dir): if sequence_file.endswith('.npy'): sequence_path = os.path.join(action_dir, sequence_file) # 加载原始序列 (30, 33, 4) raw_sequence = np.load(sequence_path) processed_sequence = [] for frame_landmarks in raw_sequence: # 对每一帧进行关键点提取和预处理 kp = extract_keypoints_from_landmarks(frame_landmarks) processed_sequence.append(kp) # 确保序列长度一致 if len(processed_sequence) == sequence_length: sequences.append(processed_sequence) labels.append(label_map[action]) X = np.array(sequences) # 形状: (样本数, 序列长度, 特征数) 如 (120, 30, 99) y = np.array(labels) # 形状: (样本数,) return X, y # 加载数据 X, y = load_data(DATA_PATH, SEQUENCE_LENGTH) print(f'数据集形状: X={X.shape}, y={y.shape}')

步骤三:数据集划分与标准化将数据划分为训练集、验证集和测试集。并对特征进行标准化(Standardization),使每个特征维度均值为0,方差为1,有助于模型收敛。

from sklearn.preprocessing import StandardScaler # 首先划分训练+验证集 和 测试集 X_train_val, X_test, y_train_val, y_test = train_test_split(X, y, test_size=0.15, random_state=42, stratify=y) # 再从训练+验证集中划分出验证集 X_train, X_val, y_train, y_val = train_test_split(X_train_val, y_train_val, test_size=0.176, random_state=42, stratify=y_train_val) # 0.176 ≈ 0.15/0.85 print(f'训练集: {X_train.shape}, 验证集: {X_val.shape}, 测试集: {X_test.shape}') # 标准化处理 # 注意:标准化器必须只在训练集上拟合,然后应用到所有集合,避免数据泄露 scaler = StandardScaler() # 将三维数据展平成二维进行拟合 original_shape_train = X_train.shape X_train_flat = X_train.reshape(-1, X_train.shape[-1]) scaler.fit(X_train_flat) # 对三个数据集进行变换 def scale_dataset(data, scaler): original_shape = data.shape data_flat = data.reshape(-1, data.shape[-1]) data_scaled_flat = scaler.transform(data_flat) return data_scaled_flat.reshape(original_shape) X_train_scaled = scale_dataset(X_train, scaler) X_val_scaled = scale_dataset(X_val, scaler) X_test_scaled = scale_dataset(X_test, scaler)

至此,我们得到了干净、标准化、可用于模型训练的数据X_train_scaled,y_train等。

4. 双引擎核心:DTW算法精解与LSTM模型构建

我们的系统采用了一种混合策略。DTW用于计算两个动作序列之间的相似度距离,可以作为特征工程的一部分,或者用于数据增强(寻找典型模板)。LSTM则是主力的序列分类模型。

4.1 动态时间规整(DTW):解决动作速度不一致的利器

为什么需要DTW?想象两个人做同一个“深蹲”,A做得快,10帧完成,B做得慢,30帧完成。如果直接用欧氏距离逐帧比较,会因为时间轴没有对齐而得到很大的距离,尽管动作本身是一样的。DTW的核心思想就是找到两个时间序列之间的最优非线性对齐路径,使得对齐后的累积距离最小。

from scipy.spatial.distance import euclidean from fastdtw import fastdtw # 需要安装:pip install fastdtw def calculate_dtw_distance(seq1, seq2): """ 计算两个序列之间的DTW距离。 :param seq1: 序列1,形状 (长度1, 特征维度) :param seq2: 序列2,形状 (长度2, 特征维度) :return: DTW距离 """ # 使用快速DTW算法(一种近似算法,比标准DTW快很多) distance, path = fastdtw(seq1, seq2, dist=euclidean) return distance # 应用:构建一个“模板库” def build_template_library(X_train, y_train, actions): """ 为每个动作类别计算一个或多个典型序列(模板)。 简单起见,这里取每个类别所有训练序列的质心(均值序列)作为模板。 """ template_dict = {} for action_idx, action_name in enumerate(actions): # 获取该类别所有序列 action_sequences = X_train[y_train == action_idx] # 计算均值序列作为模板。注意:这要求所有序列长度相同。 template = np.mean(action_sequences, axis=0) # 形状 (序列长度, 特征维度) template_dict[action_name] = template return template_dict # 使用DTW进行简单分类(KNN思想) def dtw_knn_predict(sequence, template_dict, actions): """ 使用DTW距离,通过最近邻模板进行预测。 :param sequence: 待预测的序列。 :param template_dict: 模板字典。 :param actions: 动作列表。 :return: 预测的动作标签。 """ min_distance = float('inf') predicted_action = None for action_name, template in template_dict.items(): dist = calculate_dtw_distance(sequence, template) if dist < min_distance: min_distance = dist predicted_action = action_name return predicted_action

踩坑实录:直接使用所有帧的完整高维特征(如99维)计算DTW,计算量会非常大,且高维空间中的欧氏距离可能失效(维度灾难)。一个有效的技巧是降维。我们可以只选取一些核心关节点(如四肢大关节、躯干)的坐标,或者使用PCA将99维特征降至10-20维,再用DTW计算距离,效果几乎不变,但速度提升一个数量级。

4.2 LSTM模型:捕捉时序依赖的深度网络

LSTM是处理这类问题的更强大、更通用的工具。我们将构建一个基于TensorFlow/Keras的LSTM分类模型。

from tensorflow.keras.models import Sequential from tensorflow.keras.layers import LSTM, Dense, Dropout, BatchNormalization from tensorflow.keras.callbacks import EarlyStopping, ReduceLROnPlateau from tensorflow.keras.utils import to_categorical # 准备标签的one-hot编码 y_train_cat = to_categorical(y_train, num_classes=len(ACTIONS)) y_val_cat = to_categorical(y_val, num_classes=len(ACTIONS)) y_test_cat = to_categorical(y_test, num_classes=len(ACTIONS)) def build_lstm_model(input_shape, num_classes): """ 构建LSTM模型。 :param input_shape: (序列长度, 特征维度) :param num_classes: 动作类别数 """ model = Sequential() # 第一层LSTM,返回整个序列的输出,以便接入下一层LSTM model.add(LSTM(128, return_sequences=True, input_shape=input_shape, kernel_regularizer='l2')) model.add(BatchNormalization()) # 加速收敛,稳定训练 model.add(Dropout(0.3)) # 防止过拟合 # 第二层LSTM model.add(LSTM(64, return_sequences=False, kernel_regularizer='l2')) model.add(BatchNormalization()) model.add(Dropout(0.3)) # 全连接层 model.add(Dense(64, activation='relu', kernel_regularizer='l2')) model.add(Dropout(0.3)) # 输出层 model.add(Dense(num_classes, activation='softmax')) model.compile(optimizer='adam', loss='categorical_crossentropy', metrics=['accuracy']) model.summary() return model # 获取输入形状 input_shape = (X_train_scaled.shape[1], X_train_scaled.shape[2]) # (序列长度, 特征维度) model = build_lstm_model(input_shape, len(ACTIONS)) # 定义回调函数 early_stopping = EarlyStopping(monitor='val_loss', patience=15, restore_best_weights=True, verbose=1) reduce_lr = ReduceLROnPlateau(monitor='val_loss', factor=0.5, patience=5, min_lr=1e-6, verbose=1) # 训练模型 history = model.fit( X_train_scaled, y_train_cat, validation_data=(X_val_scaled, y_val_cat), epochs=100, # 设置一个较大的值,靠早停来结束 batch_size=32, callbacks=[early_stopping, reduce_lr], verbose=1 )

模型设计要点解析:

  1. 双层LSTM:第一层return_sequences=True将每个时间步的输出都传递给下一层,让第二层LSTM能接触到更丰富的时序信息。对于动作识别,两层LSTM通常足够。
  2. BatchNormalization:这是稳定RNN/LSTM训练的利器。它放在激活函数之前,对每一批数据的输入进行归一化,可以缓解梯度消失/爆炸,允许使用更大的学习率。
  3. Dropout:LSTM层和全连接层后都加了Dropout,这是防止过拟合的标准操作。0.3-0.5的丢弃率是常见范围。
  4. 正则化:在LSTM和Dense层使用了kernel_regularizer='l2',给权重加上L2惩罚项,进一步约束模型复杂度。
  5. 回调函数EarlyStopping监控验证集损失,如果连续多个epoch不下降就停止训练,并恢复最佳权重。ReduceLROnPlateau在验证损失停滞时降低学习率,有助于模型跳出局部最优。

4.3 模型评估与可视化

训练完成后,我们需要全面评估模型性能。

import matplotlib.pyplot as plt # 1. 绘制训练历史 def plot_training_history(history): fig, axes = plt.subplots(1, 2, figsize=(12, 4)) # 准确率 axes[0].plot(history.history['accuracy'], label='Train Acc') axes[0].plot(history.history['val_accuracy'], label='Val Acc') axes[0].set_title('Model Accuracy') axes[0].set_xlabel('Epoch') axes[0].set_ylabel('Accuracy') axes[0].legend() axes[0].grid(True) # 损失 axes[1].plot(history.history['loss'], label='Train Loss') axes[1].plot(history.history['val_loss'], label='Val Loss') axes[1].set_title('Model Loss') axes[1].set_xlabel('Epoch') axes[1].set_ylabel('Loss') axes[1].legend() axes[1].grid(True) plt.tight_layout() plt.show() plot_training_history(history) # 2. 在测试集上最终评估 test_loss, test_acc = model.evaluate(X_test_scaled, y_test_cat, verbose=0) print(f'\n测试集结果:损失 = {test_loss:.4f}, 准确率 = {test_acc:.4f}') # 3. 生成分类报告和混淆矩阵 from sklearn.metrics import classification_report, confusion_matrix import seaborn as sns y_pred = model.predict(X_test_scaled) y_pred_classes = np.argmax(y_pred, axis=1) y_true_classes = np.argmax(y_test_cat, axis=1) print('\n分类报告:') print(classification_report(y_true_classes, y_pred_classes, target_names=ACTIONS)) # 混淆矩阵 cm = confusion_matrix(y_true_classes, y_pred_classes) plt.figure(figsize=(8,6)) sns.heatmap(cm, annot=True, fmt='d', cmap='Blues', xticklabels=ACTIONS, yticklabels=ACTIONS) plt.title('Confusion Matrix on Test Set') plt.ylabel('True Label') plt.xlabel('Predicted Label') plt.tight_layout() plt.show()

通过混淆矩阵,你可以清晰看到模型在哪些动作上容易混淆。比如,“深蹲”和“高抬腿”的腿部运动模式有相似之处,可能会分错。这为你后续优化指明了方向:可能需要收集更多区分性的数据,或者引入新的特征(如关节角度、运动速度)。

5. 系统集成与实时动作识别

模型训练好了,最终我们要把它用起来,做成一个实时识别系统。

5.1 模型保存与加载

首先,保存训练好的模型。

model.save('action_recognition_lstm_model.h5') # 保存为H5格式 print("模型已保存为 'action_recognition_lstm_model.h5'")

5.2 编写实时识别循环

这个循环将之前的所有模块串联起来:摄像头捕获 -> MediaPipe提取姿态 -> 预处理 -> 构建序列 -> 模型预测。

import time from collections import deque class ActionRecognizer: def __init__(self, model_path, actions, sequence_length=30, scaler=None): self.model = tf.keras.models.load_model(model_path) self.actions = actions self.sequence_length = sequence_length self.scaler = scaler # 必须传入训练时用的同一个scaler对象 # 使用一个队列来缓存最近的帧,构成一个滑动窗口序列 self.sequence_queue = deque(maxlen=sequence_length) self.pose_detector = PoseDetector() # 预测平滑:使用一个简单队列来平滑预测结果,避免抖动 self.smooth_window = 5 self.prediction_history = deque(maxlen=self.smooth_window) def predict_action(self, sequence): """ 对一个完整的序列进行预测。 """ if len(sequence) < self.sequence_length: return None, 0.0 # 序列预处理 processed_frames = [] for frame_landmarks in sequence: kp = extract_keypoints_from_landmarks(frame_landmarks) processed_frames.append(kp) processed_sequence = np.array(processed_frames) # (seq_len, features) # 标准化 if self.scaler: original_shape = processed_sequence.shape seq_flat = processed_sequence.reshape(-1, original_shape[-1]) seq_scaled_flat = self.scaler.transform(seq_flat) processed_sequence = seq_scaled_flat.reshape(original_shape) # 增加批次维度并预测 input_data = np.expand_dims(processed_sequence, axis=0) # (1, seq_len, features) predictions = self.model.predict(input_data, verbose=0)[0] predicted_idx = np.argmax(predictions) confidence = predictions[predicted_idx] return self.actions[predicted_idx], confidence def run_realtime(self): """ 启动实时摄像头进行动作识别。 """ cap = cv2.VideoCapture(0) # 预热队列,用空数据填充 for _ in range(self.sequence_length): self.sequence_queue.append(None) while cap.isOpened(): success, img = cap.read() if not success: break img, landmarks = self.pose_detector.find_pose(img, draw=True) current_time = time.time() # 更新序列队列 if landmarks is not None: self.sequence_queue.append(landmarks) else: self.sequence_queue.append(None) # 未检测到人,加入空值 # 只有当队列中有效帧足够多时才进行预测(例如超过80%) valid_frames = [f for f in self.sequence_queue if f is not None] if len(valid_frames) >= int(self.sequence_length * 0.8): # 用最近的有效帧补齐序列(简单策略) sequence_for_pred = [] for frame in self.sequence_queue: if frame is not None: sequence_for_pred.append(frame) else: # 如果遇到空帧,用上一个有效帧填充(或均值填充) if sequence_for_pred: sequence_for_pred.append(sequence_for_pred[-1]) else: continue # 确保长度正确 sequence_for_pred = sequence_for_pred[-self.sequence_length:] action, confidence = self.predict_action(sequence_for_pred) if action: # 平滑预测结果 self.prediction_history.append(action) # 取历史窗口中最常见的动作作为最终输出 from collections import Counter if len(self.prediction_history) == self.smooth_window: smoothed_action = Counter(self.prediction_history).most_common(1)[0][0] else: smoothed_action = action # 在图像上显示结果 cv2.putText(img, f'Action: {smoothed_action}', (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2) cv2.putText(img, f'Conf: {confidence:.2f}', (10, 60), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 0), 2) else: cv2.putText(img, 'Collecting Frames...', (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 0, 255), 2) cv2.imshow('Real-time Action Recognition', img) if cv2.waitKey(10) & 0xFF == ord('q'): break cap.release() cv2.destroyAllWindows() # 运行实时识别(假设scaler已从训练中保存并加载) # recognizer = ActionRecognizer('action_recognition_lstm_model.h5', ACTIONS, SEQUENCE_LENGTH, scaler) # recognizer.run_realtime()

5.3 性能优化与部署考量

实时系统对性能有要求。MediaPipe和LSTM模型都是计算密集型。

  1. MediaPipe优化:在初始化PoseDetector时,可以降低model_complexity(从1降到0),并适当提高min_detection_confidencemin_tracking_confidence阈值,以减少计算量。对于固定角度的场景(如健身镜),可以降低输入图像的分辨率。
  2. LSTM模型优化:训练完成后,可以考虑使用TensorFlow Lite将模型量化并转换为.tflite格式,在边缘设备(如树莓派、手机)上部署时能大幅提升推理速度。
  3. 预测频率:不需要每帧都预测。可以每5帧或10帧预测一次,然后用平滑滤波(如代码中的prediction_history)来稳定输出结果,这能极大降低CPU/GPU负载。
  4. 多线程:可以将图像采集、姿态检测、序列构建、模型预测放在不同的线程中,利用流水线提高整体帧率。

6. 项目总结与扩展思考

这个项目从零搭建了一个完整的人体动作识别流水线。核心在于将MediaPipe提供的“空间姿态”通过序列建模(LSTM)和序列对齐(DTW)技术,提升到了“时序动作”的理解层面。

我个人在实操中的几点深刻体会:

  1. 数据质量决定天花板:模型再精巧,如果数据不行,一切白搭。采集数据时,动作的规范性、多样性(不同人、不同速度、不同角度)至关重要。一个技巧是,可以先用DTW计算所有训练序列与类中心模板的距离,剔除那些距离过远的“异常动作”样本,能有效提升数据纯净度。
  2. 预处理是魔法:基于躯干的相对坐标归一化,其效果提升可能比换一个更复杂的模型还要显著。它从根本上解决了识别对象尺度、位置变化的问题。
  3. LSTM不是银弹:对于周期性强、节奏固定的动作(如跑步、跳绳),LSTM效果很好。但对于复杂、非周期、依赖长程上下文的行为(如“拿起水杯喝水然后放下”),可能需要更复杂的模型,如Transformer或CNN+LSTM的混合结构。
  4. 实时性的权衡SEQUENCE_LENGTH越长,能包含的时序信息越多,但系统延迟也越高。你需要根据具体动作的持续时间来权衡。对于“举手”这种短动作,15帧可能就够了;对于“完成一次引体向上”,可能需要60帧以上。

后续可以探索的扩展方向:

  • 融合DTW与LSTM:不是二选一。可以用DTW为每个训练样本计算其与类内其他样本的平均距离,作为一个“典型性”分数,在训练LSTM时给典型样本更高权重。或者,用DTW对齐后的序列作为LSTM的输入,可能对速度变化有更好的鲁棒性。
  • 引入空间注意力:不是所有关节点对识别某个动作都同等重要。识别“挥手”时手部关节更重要,识别“踢腿”时腿部关节更重要。可以在LSTM之前或之后加入注意力机制,让模型自动聚焦于相关关节。
  • 从识别到评估:这是更有价值的延伸。不仅识别出是“深蹲”,还要判断深度是否足够、膝盖是否内扣、背部是否挺直。这需要定义更精细的基于关节角度和轨迹的规则,或者收集带有质量标签(标准、不标准)的数据进行回归或分级分类训练。

这个项目代码框架清晰,模块化程度高,你可以很方便地替换动作类别、调整模型结构、增加新的功能。希望这份超详细的拆解,能帮你避开我踩过的那些坑,顺利搭建起属于自己的动作识别系统。

本文还有配套的精品资源,点击获取

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询