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
| import torch import torch.nn as nn from typing import Tuple
class MultiBranchStressNet(nn.Module): """ 多分支压力检测网络 融合三路生理信号: - EDA (Electrodermal Activity) → 皮肤电导 - BVP (Blood Volume Pulse) → 心率/HRV - Skin Temperature → 皮温 每路独立编码→融合→分类 """ def __init__(self, hidden_dim: int = 64): super().__init__() self.eda_branch = nn.Sequential( nn.Conv1d(1, 16, 5, 2, 2), nn.ReLU(), nn.Conv1d(16, 32, 5, 2, 2), nn.ReLU(), nn.Conv1d(32, 32, 3, 1, 1), nn.ReLU(), nn.AdaptiveAvgPool1d(1), ) self.bvp_branch = nn.Sequential( nn.Conv1d(1, 16, 5, 2, 2), nn.ReLU(), nn.Conv1d(16, 32, 5, 2, 2), nn.ReLU(), nn.Conv1d(32, 32, 3, 1, 1), nn.ReLU(), nn.AdaptiveAvgPool1d(1), ) self.temp_branch = nn.Sequential( nn.Linear(1, 32), nn.ReLU(), nn.Linear(32, 32), nn.ReLU(), ) fused_dim = 32 + 32 + 32 self.fusion = nn.Sequential( nn.Linear(fused_dim, hidden_dim), nn.ReLU(), nn.Dropout(0.3), nn.Linear(hidden_dim, hidden_dim // 2), nn.ReLU(), nn.Linear(hidden_dim // 2, 3), ) def forward( self, eda: torch.Tensor, bvp: torch.Tensor, temp: torch.Tensor, ) -> torch.Tensor: eda_feat = self.eda_branch(eda).squeeze(-1) bvp_feat = self.bvp_branch(bvp).squeeze(-1) temp_feat = self.temp_branch(temp) fused = torch.cat([eda_feat, bvp_feat, temp_feat], dim=1) return self.fusion(fused)
class LabelEfficientRecalibration: """ 标签高效重校准 论文发现:校准(而非架构)是跨被试的主要瓶颈 解决方案:少量标注数据快速重校准 """ def __init__(self, base_model: nn.Module): self.base_model = base_model for param in self.base_model.parameters(): param.requires_grad = False self.adaptive_head = nn.Sequential( nn.Linear(96, 64), nn.ReLU(), nn.Linear(64, 3), ) def recalibrate( self, eda: torch.Tensor, bvp: torch.Tensor, temp: torch.Tensor, labels: torch.Tensor, num_samples: int = 30, ) -> dict: """ 少量标注数据快速重校准 Args: labels: (N,) 仅N个标注样本 Returns: metrics: 重校准前后性能对比 """ from torch.optim import Adam optimizer = Adam(self.adaptive_head.parameters(), lr=1e-3) criterion = nn.CrossEntropyLoss() with torch.no_grad(): eda_feat = self.base_model.eda_branch(eda).squeeze(-1) bvp_feat = self.base_model.bvp_branch(bvp).squeeze(-1) temp_feat = self.base_model.temp_branch(temp) features = torch.cat([eda_feat, bvp_feat, temp_feat], dim=1) losses = [] for epoch in range(50): optimizer.zero_grad() logits = self.adaptive_head(features) loss = criterion(logits, labels) loss.backward() optimizer.step() losses.append(loss.item()) return { "final_loss": losses[-1], "initial_loss": losses[0], "improvement": losses[0] - losses[-1], "samples_used": num_samples, }
if __name__ == "__main__": model = MultiBranchStressNet() B = 8 eda = torch.randn(B, 1, 240) bvp = torch.randn(B, 1, 256) temp = torch.randn(B, 1) output = model(eda, bvp, temp) print(f"输出: {output.shape}") print(f"参数: {sum(p.numel() for p in model.parameters()) / 1e3:.1f}K") calibrator = LabelEfficientRecalibration(model) labels = torch.randint(0, 3, (B,)) result = calibrator.recalibrate(eda, bvp, temp, labels) print(f"重校准: {result}")
|