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
| import torch import torch.nn as nn import numpy as np
class MultiSiteRadarPTT(nn.Module): """ PolyPulse 多部位雷达 PTT 测量模型 论文核心方法: 1. mmWave 波束成形指向 4 个身体部位 2. 每个部位提取微动信号(脉搏引起的皮肤位移) 3. 检测脉搏到达时间 4. 计算多路径 PTT """ def __init__(self, num_sites=4, num_antennas=16): super().__init__() self.num_sites = num_sites self.beamformer = nn.ModuleList([ nn.Conv1d(num_antennas, 1, kernel_size=1, bias=False) for _ in range(num_sites) ]) self.pulse_detector = nn.ModuleList([ nn.Sequential( nn.Conv1d(1, 32, kernel_size=15, padding=7), nn.BatchNorm1d(32), nn.ReLU(), nn.Conv1d(32, 64, kernel_size=11, padding=5), nn.BatchNorm1d(64), nn.ReLU(), nn.Conv1d(64, 1, kernel_size=5, padding=2), nn.Sigmoid() ) for _ in range(num_sites) ]) self.ptt_estimator = nn.Sequential( nn.Linear(num_sites * 64, 256), nn.ReLU(), nn.Linear(256, 128), nn.ReLU(), nn.Linear(128, num_sites * (num_sites - 1) // 2) ) self.site_names = ['heart_apex', 'carotid', 'radial', 'mastoid'] def forward(self, radar_tensor: torch.Tensor) -> dict: """ Args: radar_tensor: (B, num_antennas, T) - 多天线雷达回波 Returns: ptt_matrix: dict of site pairs -> PTT estimates """ B, A, T = radar_tensor.shape site_signals = [] for i in range(self.num_sites): beam = self.beamformer[i](radar_tensor) site_signals.append(beam) pulse_features = [] for i in range(self.num_sites): feat = self.pulse_detector[i](site_signals[i]) feat_flat = feat.mean(dim=2) pulse_features.append(feat_flat) all_features = torch.cat(pulse_features, dim=1) ptt_raw = self.ptt_estimator(all_features) return { 'site_signals': site_signals, 'ptt_raw': ptt_raw, 'sites': self.site_names }
class PolyPulseSystem(nn.Module): """ PolyPulse: 完整的非接触雷达血压感知系统 论文:arXiv:2510.18141, Nature Communications 应用:座舱心血管监测、居家健康监测 """ def __init__(self): super().__init__() self.radar_ptt = MultiSiteRadarPTT(num_sites=4, num_antennas=16) def forward(self, radar_data: torch.Tensor) -> dict: """ 完整推理管道 Args: radar_data: (B, 16, T) - 16天线 mmWave 雷达数据 Returns: health_assessment: dict with PTT, BP estimate, risk """ ptt_results = self.radar_ptt(radar_data) ptt_heart_radial = ptt_results['ptt_raw'][0] pwv = 0.6 / (ptt_heart_radial / 1000) bp_diastolic = 80 + 0.5 * max(0, pwv - 5) ** 2 if pwv > 10: stiffness_risk = 'high' elif pwv > 7: stiffness_risk = 'moderate' else: stiffness_risk = 'low' return { 'ptt_ms': ptt_heart_radial.item(), 'pwv': pwv, 'bp_diastolic_estimate': bp_diastolic, 'arterial_stiffness': stiffness_risk, **ptt_results }
if __name__ == "__main__": print("=" * 60) print("PolyPulse 非接触雷达血压感知测试") print("=" * 60) model = PolyPulseSystem() T = 1000 radar_data = torch.randn(1, 16, T) * 0.1 results = model(radar_data) print(f"输入: {radar_data.shape} (B, antennas, T)") print(f"PTT (ms): {results['ptt_ms']:.2f}") print(f"PWV (m/s): {results['pwv']:.2f}") print(f"舒张压估算: {results['bp_diastolic_estimate']:.1f} mmHg") print(f"动脉硬度风险: {results['arterial_stiffness']}") total_params = sum(p.numel() for p in model.parameters()) print(f"\n模型参数量: {total_params / 1e3:.1f}K") print(f"\n硬件参数:") print(f" 尺寸: 16.5 × 14 cm") print(f" 天线: 16 阵列") print(f" 频段: mmWave (60GHz)") print(f" 测量部位: 4 个(心尖、颈动脉、桡动脉、乳突)")
|