1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156
| import numpy as np import torch import torch.nn as nn from typing import Tuple
class vPPGPipeline: """ Video Photoplethysmography (vPPG) 信号提取管道 核心原理: 1. 面部皮肤血液体积随心跳周期性变化 2. 血液吸收绿光较多,摄像头可捕捉肤色微变化 3. 通过信号处理提取脉冲信号 → 心率/HRV 信号处理链: 视频帧 → 面部检测 → ROI提取 → 像素平均 → 去趋势 → 带通滤波 → 脉冲信号 → 峰值检测 → HR/HRV """ def __init__(self, fps: int = 30): self.fps = fps def extract_ppg_signal(self, frames: np.ndarray, roi_mask: np.ndarray = None) -> np.ndarray: """ 从视频帧序列中提取PPG信号 Args: frames: (T, H, W, 3) RGB视频帧 roi_mask: (H, W) 面部ROI掩码 Returns: ppg_signal: (T,) 归一化PPG信号 """ T = len(frames) if roi_mask is None: roi_mask = np.ones(frames.shape[1:]) roi_pixels = frames[:, roi_mask > 0] green_avg = roi_pixels[:, :, 1].mean(axis=1) green_norm = (green_avg - green_avg.mean()) / (green_avg.std() + 1e-8) from scipy.signal import detrend green_detrend = detrend(green_norm) from scipy.signal import butter, filtfilt nyquist = self.fps / 2 low, high = 0.7 / nyquist, min(4.0 / nyquist, 0.99) b, a = butter(3, [low, high], btype='band') ppg_signal = filtfilt(b, a, green_detrend) return ppg_signal def estimate_heart_rate(self, ppg_signal: np.ndarray) -> dict: """ 从PPG信号估计心率 Returns: {'heart_rate_bpm': float, 'confidence': float} """ n = len(ppg_signal) freqs = np.fft.rfftfreq(n, d=1.0/self.fps) spectrum = np.abs(np.fft.rfft(ppg_signal)) valid = (freqs >= 40/60) & (freqs <= 200/60) if not valid.any(): return {'heart_rate_bpm': 0, 'confidence': 0} peak_idx = np.argmax(spectrum[valid]) hr_freq = freqs[valid][peak_idx] hr_bpm = hr_freq * 60 confidence = spectrum[valid][peak_idx] / (spectrum[valid].mean() + 1e-8) return { 'heart_rate_bpm': float(hr_bpm), 'confidence': float(min(confidence / 5.0, 1.0)) } def estimate_hrv(self, ppg_signal: np.ndarray) -> dict: """ 从PPG信号估计心率变异性 时域指标: - SDNN: RR间期标准差 - RMSSD: 相邻RR间期差值均方根 Returns: {'sdnn_ms': float, 'rmssd_ms': float, 'ln_rmssd': float} """ from scipy.signal import find_peaks peaks, _ = find_peaks(ppg_signal, distance=self.fps*0.4) if len(peaks) < 3: return {'sdnn_ms': 0, 'rmssd_ms': 0, 'ln_rmssd': 0} rr_intervals = np.diff(peaks) / self.fps * 1000 sdnn = np.std(rr_intervals, ddof=1) rr_diff = np.diff(rr_intervals) rmssd = np.sqrt(np.mean(rr_diff**2)) return { 'sdnn_ms': float(sdnn), 'rmssd_ms': float(rmssd), 'ln_rmssd': float(np.log(rmssd + 1e-6)), 'rr_intervals': rr_intervals.tolist(), 'n_beats': len(peaks) }
if __name__ == "__main__": pipeline = vPPGPipeline(fps=30) T = 30 * 60 frames = np.random.randint(100, 200, (T, 64, 64, 3)).astype(np.float32) t = np.arange(T) / 30 pulse = 2 * np.sin(2 * np.pi * 1.2 * t) frames[:, 20:44, 20:44, 1] += pulse * 5 ppg = pipeline.extract_ppg_signal(frames) hr_result = pipeline.estimate_heart_rate(ppg) print(f"估计心率: {hr_result['heart_rate_bpm']:.1f} BPM") print(f"置信度: {hr_result['confidence']:.2f}") hrv_result = pipeline.estimate_hrv(ppg) print(f"SDNN: {hrv_result['sdnn_ms']:.1f} ms") print(f"RMSSD: {hrv_result['rmssd_ms']:.1f} ms") print(f"ln(RMSSD): {hrv_result['ln_rmssd']:.3f}") print(f"检测到 {hrv_result['n_beats']} 个心跳")
|