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
| import torch import torch.nn as nn import numpy as np
class BayesianFatigueModel(nn.Module): """ 不确定性感知深度迁移学习疲劳检测模型 架构: 1. 五模态特征提取器 2. 域适应模块(实验室→真实驾驶) 3. 贝叶斯分类器(MC Dropout不确定性) """ def __init__(self, n_classes: int = 3): super().__init__() self.eeg_encoder = self._make_encoder(17, 128) self.ecg_encoder = self._make_encoder(1, 64) self.emg_encoder = self._make_encoder(2, 32) self.resp_encoder = self._make_encoder(1, 32) self.eda_encoder = self._make_encoder(1, 32) self.fusion = nn.Sequential( nn.Linear(288, 256), nn.ReLU(), nn.Dropout(0.3), nn.Linear(256, 128), nn.ReLU(), nn.Dropout(0.3), ) self.domain_classifier = nn.Sequential( nn.Linear(128, 64), nn.ReLU(), nn.Linear(64, 2), ) self.classifier = nn.Sequential( nn.Linear(128, 64), nn.ReLU(), nn.Dropout(0.3), nn.Linear(64, n_classes) ) def _make_encoder(self, in_ch, out_dim): return nn.Sequential( nn.Conv1d(in_ch, 32, 7, stride=2), nn.BatchNorm1d(32), nn.ReLU(), nn.Conv1d(32, 64, 5, stride=2), nn.BatchNorm1d(64), nn.ReLU(), nn.AdaptiveAvgPool1d(1), nn.Flatten(), nn.Linear(64, out_dim), nn.ReLU() ) def forward(self, eeg, ecg, emg, resp, eda, n_mc_samples=1, alpha=1.0): """ Args: 各模态信号 n_mc_samples: MC采样次数 alpha: 域适应梯度反转系数 """ e = self.eeg_encoder(eeg) c = self.ecg_encoder(ecg) m = self.emg_encoder(emg) r = self.resp_encoder(resp) d = self.eda_encoder(eda) fused = torch.cat([e, c, m, r, d], dim=-1) logits_list = [] for _ in range(n_mc_samples): h = self.fusion(fused) logits = self.classifier(h) logits_list.append(logits) if n_mc_samples > 1: logits_stack = torch.stack(logits_list) mean_logits = logits_stack.mean(dim=0) epistemic_unc = logits_stack.var(dim=0).mean() probs = torch.softmax(mean_logits, dim=-1) aleatoric_unc = -(probs * torch.log(probs + 1e-8)).sum(dim=-1).mean() else: mean_logits = logits_list[0] epistemic_unc = torch.tensor(0.0) aleatoric_unc = torch.tensor(0.0) reversed_feat = GradReverse.apply(fused, alpha) domain_logits = self.domain_classifier(reversed_feat) return { 'logits': mean_logits, 'domain_logits': domain_logits, 'epistemic_unc': epistemic_unc, 'aleatoric_unc': aleatoric_unc, }
class GradReverse(torch.autograd.Function): """梯度反转层(对抗域适应)""" @staticmethod def forward(ctx, x, alpha): ctx.alpha = alpha return x.view_as(x) @staticmethod def backward(ctx, grad_output): return -ctx.alpha * grad_output, None
if __name__ == "__main__": model = BayesianFatigueModel(n_classes=3) batch = 4 eeg = torch.randn(batch, 17, 200) ecg = torch.randn(batch, 1, 500) emg = torch.randn(batch, 2, 200) resp = torch.randn(batch, 1, 200) eda = torch.randn(batch, 1, 100) result = model(eeg, ecg, emg, resp, eda, n_mc_samples=1) print(f"单次推理: {result['logits'].shape}") result = model(eeg, ecg, emg, resp, eda, n_mc_samples=20) print(f"\n20次MC采样:") print(f" 认知不确定性: {result['epistemic_unc']:.4f}") print(f" 偶然不确定性: {result['aleatoric_unc']:.4f}") total_params = sum(p.numel() for p in model.parameters()) print(f"\n总参数: {total_params:,}")
|