本文基于毕业设计项目《MediaPose》，详细介绍如何将卡尔曼滤波算法应用于人体姿态识别，实现实时平滑与动态预测。完整项目代码已开源，欢迎交流学习。

📌 项目背景

人体姿态识别是计算机视觉领域的热点研究方向，广泛应用于运动分析、人机交互、医疗康复等领域。MediaPipe作为Google推出的开源姿态检测框架，能够实时检测人体33个关键点，但在实际应用中存在以下问题：

  1. 检测抖动：由于光照变化、遮挡、运动模糊等因素，检测结果存在明显抖动
  2. 缺乏预测：无法预测下一时刻的人体姿态，在快速运动场景下表现不佳
  3. 数据噪声：原始检测结果包含大量噪声，影响后续分析

针对这些问题，本项目提出基于卡尔曼滤波的姿态优化方案，通过引入状态估计理论，实现姿态数据的平滑处理和动态预测。

🏗️ 系统架构

整体架构设计

系统采用前后端分离架构，后端负责数据处理和算法实现，前端负责可视化展示和用户交互。

┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│   前端       │     │   后端      │     │   算法层     │
│  (React)    │◄────│  (FastAPI)  │◄────│ (MediaPipe) │
│             │     │             │     │  + Kalman   │
└─────────────┘     └─────────────┘     └─────────────┘
       │                   │                   │
       │                   │                   │
  WebSocket            REST API          算法处理
  实时传输             配置管理           数据融合

核心模块

后端数据处理部分包含以下核心模块：

  1. 图像预处理模块

    • 灰度化处理
    • 高斯滤波降噪
    • 直方图均衡化

  2. 姿态检测模块

    • MediaPipe姿态检测封装
    • 33个关键点提取
    • 可视度评估

  3. 卡尔曼滤波模块

    • 二维卡尔曼滤波器实现
    • 多维滤波器融合系统
    • 概率融合算法

  4. 姿态追踪管理器

    • 各模块协调工作
    • 历史数据管理
    • 统计信息计算

  5. API服务模块

    • RESTful API接口
    • WebSocket实时数据流
    • 视频文件管理

🧮 核心算法详解

1. 卡尔曼滤波器原理

卡尔曼滤波是一种递归的最优估计算法，通过"预测-修正"两个步骤，对含有噪声的测量数据进行最优估计。

状态空间模型

对于每个关键点，我们建立4维状态向量：

# 状态向量：位置和速度
x = [x, y, vx, vy]^T

其中：

  • x, y：关键点的二维坐标
  • vx, vy：关键点在x、y方向的速度
状态转移矩阵

采用匀速运动模型：

F = [[1, 0, Δt,  0],
     [0, 1,  0, Δt],
     [0, 0,  1,  0],
     [0, 0,  0,  1]]

其中Δt为时间步长，系统设置为1/30秒（30fps）。

预测步骤
def predict(self) -> np.ndarray:
    # 状态预测
    self.x = self.F @ self.x
    
    # 协方差预测
    self.P = self.F @ self.P @ self.F.T + self.Q
    
    return self.x

预测步骤利用上一时刻的状态估计，通过状态转移矩阵预测当前时刻的状态。

修正步骤
def update(self, measurement: np.ndarray) -> np.ndarray:
    # 计算卡尔曼增益
    S = self.H @ self.P @ self.H.T + self.R
    K = self.P @ self.H.T @ np.linalg.inv(S)
    
    # 测量残差
    y = measurement.reshape(-1, 1) - self.H @ self.x
    
    # 状态更新
    self.x = self.x + K @ y
    
    # 协方差更新
    I = np.eye(len(self.x))
    self.P = (I - K @ self.H) @ self.P
    
    return self.x

修正步骤利用当前时刻的测量值，通过卡尔曼增益对预测值进行修正。

2. 多维滤波器融合系统

由于MediaPipe输出的是三维坐标（x, y, z），我们设计了多卡尔曼滤波器融合系统，使用三个2D卡尔曼滤波器分别处理XY、YZ、XZ平面。

系统架构
class MultiDimensionKalmanFilter:
    def __init__(self, fusion_method: str = None):
        # 创建三个2D滤波器
        self.filter_xy = KalmanFilter2D(var_names=('x', 'y'))
        self.filter_yz = KalmanFilter2D(var_names=('y', 'z'))
        self.filter_xz = KalmanFilter2D(var_names=('x', 'z'))
融合策略

系统提供两种融合方法：

加权平均法：

def _weighted_fusion(self, xy, yz, xz):
    # 获取各方向的方差
    var_xy = self.filter_xy.get_position_variance()
    var_yz = self.filter_yz.get_position_variance()
    var_xz = self.filter_xz.get_position_variance()
    
    # 计算权重：方差越小，权重越大
    w_xy = 1 / (var_xy[0] + var_xy[1] + 1e-6)
    w_yz = 1 / (var_yz[0] + var_yz[1] + 1e-6)
    w_xz = 1 / (var_xz[0] + var_xz[1] + 1e-6)
    
    # 融合x坐标（来自xy和xz）
    x_val = (w_xy * xy[0] + w_xz * xz[0]) / (w_xy + w_xz)
    
    # 融合y坐标（来自xy和yz）
    y_val = (w_xy * xy[1] + w_yz * yz[0]) / (w_xy + w_yz)
    
    # 融合z坐标（来自yz和xz）
    z_val = (w_yz * yz[1] + w_xz * xz[1]) / (w_yz + w_xz)
    
    return [x_val, y_val, z_val]

概率融合法（推荐）：

def _probability_fusion(self, xy, yz, xz):
    # 基于贝叶斯理论，将各滤波器输出视为高斯分布
    # 使用精度（方差的倒数）作为权重
    precision_xy = 1 / (var_xy[0] + var_xy[1] + 1e-6)
    precision_yz = 1 / (var_yz[0] + var_yz[1] + 1e-6)
    precision_xz = 1 / (var_xz[0] + var_xz[1] + 1e-6)
    
    # 归一化权重
    total = precision_xy + precision_yz + precision_xz
    w_xy = precision_xy / total
    w_yz = precision_yz / total
    w_xz = precision_xz / total
    
    # 加权融合
    x_val = (xy[0] * w_xy + xz[0] * w_xz) / (w_xy + w_xz)
    y_val = (xy[1] * w_xy + yz[0] * w_yz) / (w_xy + w_yz)
    z_val = (yz[1] * w_yz + xz[1] * w_xz) / (w_yz + w_xz)
    
    return [x_val, y_val, z_val]

概率融合法基于贝叶斯理论，将各滤波器输出视为高斯分布进行融合，理论上更加严谨。

3. 姿态滤波器管理器

为MediaPipe的33个关键点各自管理一套多维滤波器：

class PoseKalmanFilterManager:
    def __init__(self, num_keypoints: int = 33):
        # 为每个关键点创建独立的多维滤波器
        self.filters = [
            MultiDimensionKalmanFilter(fusion_method)
            for _ in range(num_keypoints)
        ]
    
    def predict(self) -> np.ndarray:
        # 预测所有关键点的下一时刻状态
        predicted_positions = np.zeros((self.num_keypoints, 3))
        predicted_velocities = np.zeros((self.num_keypoints, 3))
        
        for i, kf in enumerate(self.filters):
            result = kf.predict()
            predicted_positions[i] = result['position']
            predicted_velocities[i] = result['velocity']
        
        return predicted_positions, predicted_velocities
    
    def update(self, measurements: np.ndarray):
        # 更新所有关键点的滤波器状态
        filtered_positions = np.zeros((self.num_keypoints, 3))
        velocities = np.zeros((self.num_keypoints, 3))
        
        for i, (kf, measurement) in enumerate(zip(self.filters, measurements)):
            result = kf.update(measurement)
            filtered_positions[i] = result['position']
            velocities[i] = result['velocity']
        
        return filtered_positions, velocities

💻 代码实现分析

1. 图像预处理流程

根据论文2.3节实现的图像预处理流程，包含三个关键步骤：

class ImagePreprocessor:
    def preprocess(self, frame: np.ndarray) -> np.ndarray:
        result = frame.copy()
        
        # 1. 灰度化（可选）
        if self.enable_gray:
            result = cv2.cvtColor(result, cv2.COLOR_BGR2GRAY)
        
        # 2. 高斯滤波降噪
        if self.enable_gaussian:
            kernel_size = (self.gaussian_kernel, self.gaussian_kernel)
            result = cv2.GaussianBlur(result, kernel_size, self.gaussian_sigma)
        
        # 3. 直方图均衡化
        if self.enable_equalize:
            if len(result.shape) == 2:  # 灰度图
                result = cv2.equalizeHist(result)
            else:  # 彩色图，在YUV空间做均衡化
                yuv = cv2.cvtColor(result, cv2.COLOR_BGR2YUV)
                yuv[:, :, 0] = cv2.equalizeHist(yuv[:, :, 0])
                result = cv2.cvtColor(yuv, cv2.COLOR_YUV2BGR)
        
        return result

预处理效果：

  • 灰度化：减少计算量，突出轮廓信息
  • 高斯滤波：平滑图像，减少高频噪声
  • 直方图均衡化：增强图像对比度，改善光照不均

2. 姿态检测与追踪流程

class PoseTracker:
    def process_frame(self, frame: np.ndarray) -> Dict:
        # 1. 图像预处理
        if self.enable_preprocessing:
            processed_frame = self.preprocessor.preprocess(frame)
        else:
            processed_frame = frame
        
        # 2. 姿态检测
        detection_result = self.detector.detect(processed_frame)
        
        if not detection_result['success']:
            # 检测失败，返回预测结果
            pred_pos, pred_vel = self.kalman_manager.predict()
            return {...}
        
        raw_landmarks = detection_result['landmarks']
        
        # 3. 卡尔曼滤波更新
        filtered_landmarks, velocities = self.kalman_manager.update(raw_landmarks)
        
        # 4. 预测下一帧
        predicted_landmarks, _ = self.kalman_manager.predict()
        
        # 5. 存储历史数据
        self.raw_history.append(raw_landmarks.copy())
        self.filtered_history.append(filtered_landmarks.copy())
        self.predicted_history.append(predicted_landmarks.copy())
        
        return {
            'raw_landmarks': raw_landmarks,
            'filtered_landmarks': filtered_landmarks,
            'predicted_landmarks': predicted_landmarks,
            'velocities': velocities,
            'success': True
        }

3. 实时数据流处理

使用WebSocket实现实时数据传输：

@app.websocket("/ws/pose")
async def websocket_pose_stream(websocket: WebSocket):
    await websocket.accept()
    
    try:
        while True:
            # 读取并处理帧
            result = tracker.read_and_process()
            
            if result is None:
                break
            
            # 创建可视化图像
            vis_frame = visualizer.draw_comparison(
                frame,
                result['raw_landmarks'],
                result['filtered_landmarks'],
                result['predicted_landmarks']
            )
            
            # 编码为JPEG
            _, buffer = cv2.imencode('.jpg', vis_frame)
            frame_b64 = base64.b64encode(buffer).decode('utf-8')
            
            # 构造发送消息
            message = {
                "type": "pose_data",
                "frame_id": result['frame_id'],
                "frame_image": f"data:image/jpeg;base64,{frame_b64}",
                "raw_landmarks": result['raw_landmarks'].tolist(),
                "filtered_landmarks": result['filtered_landmarks'].tolist(),
                "predicted_landmarks": result['predicted_landmarks'].tolist(),
                "velocities": result['velocities'].tolist()
            }
            
            await websocket.send_json(message)
            await asyncio.sleep(1 / 30)  # 控制帧率

📊 效果展示与性能分析

滤波效果量化

系统实时计算滤波效果，通过标准差对比量化抖动减少率：

def get_statistics(self) -> Dict:
    raw_arr = np.array(self.raw_history)
    filtered_arr = np.array(self.filtered_history)
    
    # 计算抖动（标准差）
    raw_std = np.std(raw_arr, axis=0)
    filtered_std = np.std(filtered_arr, axis=0)
    
    # 计算平均抖动减少率
    improvement = (1 - filtered_std / (raw_std + 1e-6)) * 100
    
    return {
        'raw_std': raw_std.mean(axis=0).tolist(),
        'filtered_std': filtered_std.mean(axis=0).tolist(),
        'improvement_percent': improvement.mean(axis=0).tolist()
    }

实验结果

在多种场景下测试，系统表现出色：

场景 原始抖动（像素） 滤波后抖动（像素） 抖动减少率
静态站立 2.3 0.8 65.2%
慢速行走 4.1 1.2 70.7%
快速运动 6.8 2.1 69.1%
遮挡场景 5.2 1.5 71.2%

性能指标

  • 处理帧率：30fps实时处理
  • 延迟：WebSocket传输延迟 < 50ms
  • 内存占用：单个滤波器约 2KB，33个滤波器约 66KB
  • CPU占用：单核占用率约 20-30%

🎯 项目亮点

1. 理论与实践结合

  • 基于论文理论，完整实现卡尔曼滤波算法
  • 多维滤波器融合系统设计新颖
  • 概率融合法基于贝叶斯理论，理论严谨

2. 工程实现优秀

  • 模块化设计，代码结构清晰
  • 配置灵活，支持参数调优
  • 完善的错误处理和日志记录

3. 可视化直观

  • 三种骨架（原始、滤波、预测）清晰区分
  • 实时统计数据显示滤波效果
  • 数据曲线图量化展示优化效果

4. 实时性能优异

  • WebSocket实时传输，延迟低
  • 支持摄像头和视频两种输入
  • 前端Canvas绘制流畅

🔧 技术栈

后端

  • FastAPI 0.109.0：高性能Web框架
  • MediaPipe 0.10.20：人体姿态检测
  • OpenCV 4.9.0：图像处理
  • NumPy 1.26.3：数值计算
  • WebSocket：实时数据传输

前端

  • React 18.2.0：UI框架
  • TypeScript 5.3.3：类型安全
  • Ant Design 5.12.8：UI组件库
  • ECharts 5.4.3：数据可视化
  • Vite 5.0.11：构建工具

📝 使用指南

快速启动

# 1. 安装依赖
pip3 install -r requirements.txt
cd frontend && npm install

# 2. 启动后端
cd backend && python3 main.py

# 3. 启动前端
cd frontend && npm run dev

# 4. 访问系统
# 前端：http://localhost:5173
# 后端：http://localhost:8000

API接口

启动视频源：

POST /api/video/start
{
  "source_type": "camera",  # 或 "video"
  "camera_id": 0,
  "source_path": "/path/to/video.mp4"  # 视频模式时必填
}

获取统计信息：

GET /api/statistics

WebSocket连接：

const ws = new WebSocket('ws://localhost:8000/ws/pose');
ws.onmessage = (event) => {
  const data = JSON.parse(event.data);
  // 处理姿态数据
};

运行结果

在这里插入图片描述
在这里插入图片描述

🚀 未来展望

短期优化

  1. 算法优化：尝试扩展卡尔曼滤波（EKF）处理非线性运动
  2. 性能提升：使用GPU加速MediaPipe检测
  3. 功能扩展：支持多人姿态检测

长期规划

  1. 深度学习融合：结合LSTM、Transformer等深度学习模型
  2. 3D重建：基于单目摄像头实现3D人体重建
  3. 动作识别：基于姿态数据实现动作分类和识别

📚 参考资料

  1. Kalman, R. E. (1960). “A New Approach to Linear Filtering and Prediction Problems”
  2. MediaPipe Pose: https://google.github.io/mediapipe/solutions/pose.html
  3. FastAPI官方文档: https://fastapi.tiangolo.com/

💬 交流与反馈

本项目是毕业设计作品，欢迎各位大佬批评指正！

  • 论文：《基于卡尔曼滤波与视觉检测的人体姿态识别优化与动态预测》
  • 项目：MediaPose - 基于卡尔曼滤波的人体姿态识别与预测系统

如果觉得有帮助，请点赞收藏支持一下！ 🙏

关键词：卡尔曼滤波、人体姿态识别、MediaPipe、计算机视觉、Python、FastAPI

# python # pip # pycharm # vscode
Logo
AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

cover

从“看天吃饭”到“按图赚钱”：2026风光电站为什么必须重做气象服务体系

avatar AtomGit开源社区
cover

【Linux】Shell 脚本解析日志文件提取关键信息

avatar AtomGit开源社区
cover

零基础极速上手：用AI建站工具10分钟创建一个专业网站

avatar AtomGit开源社区