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 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176
| import torch import torch.nn as nn
class rePPGGenerator(nn.Module): """ rePPG生成器:将PPG信号嵌入到面部视频帧中 论文核心思想: 1. 提取面部皮肤区域 2. 根据PPG信号调制皮肤像素的绿色通道 3. 生成含有目标心率的视频 """ def __init__(self): super().__init__() self.encoder = nn.Sequential( nn.Conv2d(3, 64, 3, padding=1), nn.BatchNorm2d(64), nn.ReLU(), nn.Conv2d(64, 128, 3, padding=1), nn.BatchNorm2d(128), nn.ReLU(), nn.Conv2d(128, 256, 3, padding=1), nn.BatchNorm2d(256), nn.ReLU(), ) self.ppg_proj = nn.Linear(1, 256) self.decoder = nn.Sequential( nn.Conv2d(256, 128, 3, padding=1), nn.BatchNorm2d(128), nn.ReLU(), nn.Conv2d(128, 64, 3, padding=1), nn.BatchNorm2d(64), nn.ReLU(), nn.Conv2d(64, 3, 3, padding=1), nn.Tanh(), ) def forward(self, frame: torch.Tensor, ppg_value: torch.Tensor, skin_mask: torch.Tensor) -> torch.Tensor: """ Args: frame: [B, 3, H, W] 原始面部帧 ppg_value: [B, 1] PPG信号值(归一化) skin_mask: [B, 1, H, W] 皮肤区域掩码 Returns: relit_frame: [B, 3, H, W] 嵌入PPG的视频帧 """ feat = self.encoder(frame) ppg_feat = self.ppg_proj(ppg_value) ppg_feat = ppg_feat.unsqueeze(-1).unsqueeze(-1) fused = feat + ppg_feat modulated = self.decoder(fused) relit = frame * (1 - skin_mask) + modulated * skin_mask return relit
def embed_ppg_to_video( frames: np.ndarray, ppg_signal: np.ndarray, skin_mask: np.ndarray, generator: rePPGGenerator ) -> np.ndarray: """ 将PPG信号嵌入到视频序列中 Args: frames: 原始视频帧 ppg_signal: 目标PPG信号 skin_mask: 皮肤区域掩码 generator: rePPG生成器 Returns: relit_frames: 含有目标PPG的视频 """ T = len(frames) relit_frames = np.zeros_like(frames) for t in range(T): frame = torch.from_numpy(frames[t]).permute(2, 0, 1).float() / 255 frame = frame.unsqueeze(0) ppg = torch.tensor([[ppg_signal[t]]], dtype=torch.float32) mask = torch.from_numpy(skin_mask).float().unsqueeze(0).unsqueeze(0) with torch.no_grad(): relit = generator(frame, ppg, mask) relit_frames[t] = (relit.squeeze(0).permute(1, 2, 0).numpy() * 255).astype(np.uint8) return relit_frames
class rPPGDataAugmentor: """ rPPG数据增强管道 使用rePPG从有限真实数据生成大量训练样本 """ def __init__(self, generator: rePPGGenerator): self.generator = generator def augment(self, original_video, original_ppg, skin_mask): """ 从原始视频生成增强样本 策略: 1. 心率缩放:将原始PPG缩放到不同心率 2. 振幅调整:改变PPG幅度 3. 噪声叠加:添加不同噪声水平 4. 运动合成:叠加虚拟运动 """ augmented = [] for scale in [0.7, 0.85, 1.0, 1.2, 1.5]: scaled_ppg = self._rescale_hr(original_ppg, scale) video = embed_ppg_to_video( original_video, scaled_ppg, skin_mask, self.generator ) augmented.append((video, scaled_ppg)) for amp in [0.5, 1.0, 2.0]: amp_ppg = original_ppg * amp video = embed_ppg_to_video( original_video, amp_ppg, skin_mask, self.generator ) augmented.append((video, amp_ppg)) return augmented def _rescale_hr(self, ppg, scale): """缩放心率""" from scipy.signal import resample n = len(ppg) new_n = int(n / scale) resampled = resample(ppg, new_n)[:n] return resampled
if __name__ == "__main__": import numpy as np gen = rePPGGenerator() frame = torch.randn(1, 3, 64, 64) ppg = torch.tensor([[0.5]]) mask = torch.ones(1, 1, 64, 64) output = gen(frame, ppg, mask) print(f"输入帧: {frame.shape}") print(f"PPG值: {ppg.shape}") print(f"输出帧: {output.shape}") print(f"参数量: {sum(p.numel() for p in gen.parameters()):,}")
|