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 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211
| """ 主体无关的时序疲劳检测框架 """
import torch import torch.nn as nn
class SubjectIndependentDrowsinessModel(nn.Module): """ 设计目标:在新驾驶员上泛化 核心策略: 1. 使用主体无关特征(眨眼频率、PERCLOS、头部运动模式) 2. 时序建模捕捉疲劳趋势(而非瞬时状态) 3. 对抗训练消除个体特征 """ def __init__(self, feature_dim: int = 64, hidden_dim: int = 128, num_classes: int = 3): super().__init__() self.temporal_encoder = nn.LSTM( input_size=feature_dim, hidden_size=hidden_dim, num_layers=2, batch_first=True, bidirectional=False, dropout=0.3 ) self.subject_invariant_features = SubjectInvariantExtractor( input_dim=feature_dim, output_dim=32 ) self.classifier = nn.Sequential( nn.Linear(hidden_dim + 32, 64), nn.GELU(), nn.Dropout(0.3), nn.Linear(64, num_classes) ) self.subject_discriminator = nn.Sequential( nn.Linear(hidden_dim, 32), nn.ReLU(), nn.Linear(32, 4) ) def forward(self, x, return_features=False): """ Args: x: (B, T, feature_dim) 时序特征序列 feature_dim 包含: [eye_aspect_ratio, blink_freq, perclos, head_pose, gaze_entropy, ...] """ temporal_out, (h_n, c_n) = self.temporal_encoder(x) temporal_feat = temporal_out[:, -1, :] invariant_feat = self.subject_invariant_features(x[:, -1, :]) combined = torch.cat([temporal_feat, invariant_feat], dim=1) logits = self.classifier(combined) if return_features: return logits, temporal_feat, combined return logits def adversarial_loss(self, temporal_feat, driver_labels): """ 对抗损失:让时序特征无法区分驾驶员身份 梯度反转层(GRL)实现 """ reversed_feat = GradientReversalLayer.apply(temporal_feat, 1.0) driver_pred = self.subject_discriminator(reversed_feat) adv_loss = nn.CrossEntropyLoss()(driver_pred, driver_labels) return adv_loss
class SubjectInvariantExtractor(nn.Module): """ 提取主体无关特征 关键洞察:某些特征与疲劳程度相关但与个人身份无关 - PERCLOS:闭眼时间比例(所有人均适用) - 眨眼频率变化趋势(频率上升=疲劳) - 头部下垂角度变化率(而非绝对角度) 而以下特征是主体相关的,应消除: - 眼睛绝对大小(因人而异) - 头部绝对姿态(坐姿习惯不同) - 眨眼绝对频率(基线不同) """ def __init__(self, input_dim: int, output_dim: int): super().__init__() self.extractor = nn.Sequential( nn.Linear(input_dim, 64), nn.BatchNorm1d(64), nn.GELU(), nn.Linear(64, output_dim) ) def forward(self, x): return self.extractor(x)
class GradientReversalLayer: """梯度反转层""" @staticmethod def apply(x, lambda_): x = x.clone() x.requires_grad = True class GRLFunction(torch.autograd.Function): @staticmethod def forward(ctx, x): return x.view_as(x) @staticmethod def backward(ctx, grad_output): return -lambda_ * grad_output return GRLFunction.apply(x)
def train_subject_independent_model(): """ 主体无关疲劳检测模型训练流程 """ model = SubjectIndependentDrowsinessModel() optimizer = torch.optim.Adam(model.parameters(), lr=1e-3) for fold, (train_idx, test_idx) in enumerate(lodo_splits): print(f"\n=== Fold {fold+1}: 测试驾驶员 = {test_drivers[fold]} ===") for epoch in range(50): model.train() for batch_x, batch_y, batch_driver in train_loader: logits, temporal_feat, _ = model(batch_x, return_features=True) cls_loss = nn.CrossEntropyLoss()(logits, batch_y) adv_loss = model.adversarial_loss(temporal_feat, batch_driver) total_loss = cls_loss + 0.1 * adv_loss optimizer.zero_grad() total_loss.backward() optimizer.step() model.eval() with torch.no_grad(): test_acc = evaluate(model, test_loader) if epoch % 10 == 0: print(f" Epoch {epoch}: cls_loss={cls_loss:.4f}, " f"adv_loss={adv_loss:.4f}, test_acc={test_acc:.2%}")
if __name__ == "__main__": torch.manual_seed(42) n_drivers = 4 n_per_driver = 16482 seq_len = 30 feature_dim = 8 X = torch.randn(n_drivers * 500, seq_len, feature_dim) y = torch.randint(0, 3, (n_drivers * 500,)) driver_ids = torch.repeat_interleave(torch.arange(n_drivers), 500) model = SubjectIndependentDrowsinessModel(feature_dim=feature_dim) logits, temporal_feat, combined = model(X[:4], return_features=True) print(f"输入形状: {X[:4].shape}") print(f"输出形状: {logits.shape}") print(f"时序特征形状: {temporal_feat.shape}") print(f"融合特征形状: {combined.shape}") adv_loss = model.adversarial_loss(temporal_feat, torch.tensor([0,1,2,3])) print(f"对抗损失: {adv_loss.item():.4f}") total_params = sum(p.numel() for p in model.parameters()) print(f"\n模型参数量: {total_params/1e6:.2f}M")
|