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
| import torch import torch.nn as nn
class P2EVQ(nn.Module): """ P2E-VQ: PPG to ECG via discrete patch retrieval 核心思路: 1. 将 ECG 信号编码为离散 patch(VQ-VAE) 2. 从 PPG 信号检索对应的 ECG patch 3. 拼接 patch 重建 ECG 波形 """ def __init__(self, signal_length=2560, patch_size=64, codebook_size=512, hidden=128): super().__init__() self.patch_size = patch_size self.num_patches = signal_length // patch_size self.ecg_encoder = nn.Sequential( nn.Linear(patch_size, hidden), nn.GELU(), nn.Linear(hidden, hidden) ) self.codebook = nn.Embedding(codebook_size, hidden) self.ecg_decoder = nn.Sequential( nn.Linear(hidden, hidden), nn.GELU(), nn.Linear(hidden, patch_size) ) self.ppg_encoder = nn.Sequential( nn.Linear(patch_size, hidden), nn.GELU(), nn.Linear(hidden, hidden) ) self.retrieval_proj = nn.Linear(hidden, hidden) def encode_ecg_patches(self, ecg_signal): """将 ECG 编码为离散 patch""" B = ecg_signal.shape[0] patches = ecg_signal.view(B, self.num_patches, self.patch_size) encoded = self.ecg_encoder(patches) dist = torch.cdist(encoded, self.codebook.weight.unsqueeze(0)) indices = dist.argmin(dim=-1) quantized = self.codebook(indices) quantized = encoded + (quantized - encoded).detach() return quantized, indices def retrieve_from_ppg(self, ppg_signal): """从 PPG 信号检索 ECG patch""" B = ppg_signal.shape[0] patches = ppg_signal.view(B, self.num_patches, self.patch_size) ppg_encoded = self.ppg_encoder(patches) projected = self.retrieval_proj(ppg_encoded) dist = torch.cdist(projected, self.codebook.weight.unsqueeze(0)) indices = dist.argmin(dim=-1) retrieved = self.codebook(indices) ecg_patches = self.ecg_decoder(retrieved) ecg_signal = ecg_patches.view(B, -1) return ecg_signal, indices
def test_p2evq(): print("=" * 60) print("P2E-VQ PPG→ECG 测试") print("=" * 60) model = P2EVQ(signal_length=2560, patch_size=64) ppg = torch.randn(2, 2560) ecg, indices = model.retrieve_from_ppg(ppg) print(f"PPG 输入: {ppg.shape}") print(f"ECG 输出: {ecg.shape}") print(f"Patch 索引: {indices.shape}") print(f"使用码本: {len(indices.unique())}/{512}") total = sum(p.numel() for p in model.parameters()) print(f"参数量: {total / 1e6:.2f}M") return model
if __name__ == "__main__": test_p2evq()
|