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
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
| """
SAE L3 被动疲劳多模态检测
论文核心: 眼动 + ECG 多模态融合
关键发现:
1. 被动疲劳在 20-40 分钟内即可发生
2. 眼动特征最敏感 (扫视频率↓, 注视时长↑)
3. ECG 特征有延迟 (HRV 变化在 10-15 分钟后)
4. 融合检测 F1 优于单一模态
IMS 可用模态:
- 眼动: DMS 摄像头 (PERCLOS, 扫视, 注视)
- ECG: rPPG (非接触) 或方向盘电极 (接触)
- 行为: 方向盘修正频率, 车道保持
"""
import torch
import torch.nn as nn
import torch.nn.functional as F
from typing import Dict, Tuple, List
import numpy as np
from dataclasses import dataclass
from enum import Enum
class FatigueLevel(Enum):
"""疲劳等级"""
AWAKE = 0
MILD = 1
MODERATE = 2
SEVERE = 3
@dataclass
class EyeMovementFeatures:
"""眼动特征 (DMS 摄像头可提取)"""
perclos: float
blink_rate: float
blink_duration: float
saccade_freq: float
saccade_velocity: float
fixation_duration: float
gaze_dispersion: float
pupil_diameter: float
@classmethod
def from_dms(cls, perclos: float, blink_rate: float,
saccade_freq: float, fixation_duration: float,
pupil_diameter: float) -> 'EyeMovementFeatures':
"""从 DMS 摄像头数据提取"""
return cls(
perclos=perclos,
blink_rate=blink_rate,
blink_duration=200 + perclos * 100,
saccade_freq=saccade_freq,
saccade_velocity=200 - perclos * 100,
fixation_duration=fixation_duration,
gaze_dispersion=5.0 + perclos * 10,
pupil_diameter=pupil_diameter
)
@dataclass
class ECGFeatures:
"""ECG/rPPG 特征"""
heart_rate: float
hrv_rmssd: float
hrv_sdnn: float
lf_hf_ratio: float
poincare_s1: float
poincare_s2: float
@classmethod
def from_rppg(cls, hr: float, hrv: float) -> 'ECGFeatures':
"""从 rPPG 提取"""
return cls(
heart_rate=hr,
hrv_rmssd=hrv,
hrv_sdnn=hrv * 1.5,
lf_hf_ratio=2.0 + (1 - hrv/50) * 3,
poincare_s1=hrv * 0.5,
poincare_s2=hrv * 0.8
)
class PassiveFatigueDetector(nn.Module):
"""
被动疲劳检测器
论文核心方法: 眼动 + ECG 多模态融合
管道:
1. 眼动特征 → MLP
2. ECG 特征 → MLP
3. 融合 → 分类
4. 输出: 4 级疲劳
航空经验:
- 微睡眠检测延迟 ≤ 3s (眼动)
- HRV 变化延迟 10-15 min (ECG)
- 融合可缩短检测延迟
"""
def __init__(self, eye_dim: int = 8, ecg_dim: int = 6,
hidden_dim: int = 64, n_classes: int = 4):
super().__init__()
self.eye_encoder = nn.Sequential(
nn.Linear(eye_dim, hidden_dim),
nn.ReLU(),
nn.Dropout(0.2),
nn.Linear(hidden_dim, hidden_dim),
nn.ReLU()
)
self.ecg_encoder = nn.Sequential(
nn.Linear(ecg_dim, hidden_dim),
nn.ReLU(),
nn.Dropout(0.2),
nn.Linear(hidden_dim, hidden_dim),
nn.ReLU()
)
self.fusion_attention = nn.Sequential(
nn.Linear(hidden_dim * 2, 2),
nn.Softmax(dim=-1)
)
self.classifier = nn.Sequential(
nn.Linear(hidden_dim, hidden_dim // 2),
nn.ReLU(),
nn.Dropout(0.3),
nn.Linear(hidden_dim // 2, n_classes)
)
self.takeover_head = nn.Linear(hidden_dim, 1)
def forward(self, eye_feat: torch.Tensor,
ecg_feat: torch.Tensor) -> Dict[str, torch.Tensor]:
"""
Args:
eye_feat: (B, 8) 眼动特征
ecg_feat: (B, 6) ECG 特征
Returns:
fatigue_logits: (B, 4) 疲劳等级
takeover_readiness: (B, 1) 接管就绪度 0-1
modality_weights: (B, 2) 模态权重
"""
eye_encoded = self.eye_encoder(eye_feat)
ecg_encoded = self.ecg_encoder(ecg_feat)
concat = torch.cat([eye_encoded, ecg_encoded], dim=-1)
weights = self.fusion_attention(concat)
fused = (weights[:, 0:1] * eye_encoded +
weights[:, 1:2] * ecg_encoded)
fatigue_logits = self.classifier(fused)
takeover_score = torch.sigmoid(self.takeover_head(fused))
return {
'fatigue_logits': fatigue_logits,
'takeover_readiness': takeover_score.squeeze(-1),
'modality_weights': weights
}
class SAE_L3_TakeoverManager:
"""
SAE L3 接管管理器
结合被动疲劳检测和接管就绪评估
航空经验:
1. 预警分级 (注意→警告→紧急)
2. 渐进式接管请求
3. 最小风险策略 (无响应时)
"""
def __init__(self):
self.detector = PassiveFatigueDetector()
self.thresholds = {
FatigueLevel.AWAKE: {'takeover_time': 2.0, 'action': '正常'},
FatigueLevel.MILD: {'takeover_time': 3.0, 'action': '一级提醒'},
FatigueLevel.MODERATE: {'takeover_time': 5.0, 'action': '二级警告'},
FatigueLevel.SEVERE: {'takeover_time': 10.0, 'action': '紧急接管'},
}
def assess(self, eye_feat: EyeMovementFeatures,
ecg_feat: ECGFeatures) -> dict:
"""
评估 L3 接管就绪度
Returns:
assessment: {
'fatigue_level': FatigueLevel,
'takeover_time': 预期接管时间 (秒),
'takeover_readiness': 0-1,
'action': str,
'mrm': 最小风险策略 (如无响应)
}
"""
eye_tensor = torch.tensor([[
eye_feat.perclos, eye_feat.blink_rate,
eye_feat.blink_duration, eye_feat.saccade_freq,
eye_feat.saccade_velocity, eye_feat.fixation_duration,
eye_feat.gaze_dispersion, eye_feat.pupil_diameter
]], dtype=torch.float32)
ecg_tensor = torch.tensor([[
ecg_feat.heart_rate, ecg_feat.hrv_rmssd,
ecg_feat.hrv_sdnn, ecg_feat.lf_hf_ratio,
ecg_feat.poincare_s1, ecg_feat.poincare_s2
]], dtype=torch.float32)
with torch.no_grad():
output = self.detector(eye_tensor, ecg_tensor)
level = FatigueLevel(output['fatigue_logits'].argmax().item())
readiness = output['takeover_readiness'].item()
weights = output['modality_weights'][0].tolist()
config = self.thresholds[level]
if level == FatigueLevel.SEVERE and readiness < 0.3:
mrm = 'MRM: 靠边停车 + 双闪 + 紧急呼叫'
elif level == FatigueLevel.MODERATE and readiness < 0.5:
mrm = 'MRM: 减速 + 车道保持 + 加速警告'
else:
mrm = '无需 MRM'
return {
'fatigue_level': level.name,
'takeover_time': config['takeover_time'],
'takeover_readiness': readiness,
'action': config['action'],
'mrm': mrm,
'eye_weight': weights[0],
'ecg_weight': weights[1],
'fatigue_probs': F.softmax(output['fatigue_logits'], dim=-1)[0].tolist()
}
class AviationFatigueLessons:
"""
航空座舱疲劳管理经验 → 汽车座舱
航空经验总结:
1. 飞行时间限制 → SAE L3 监督时间限制
2. 双驾驶员 → 接管就绪评估
3. FRMS → 疲劳风险管理系统
4. 微睡眠检测 → PERCLOS + 扫视
5. 莫达非尼 → 不适用汽车 (但感知原理可参考)
"""
LESSONS = [
{
'aviation': '飞行时间限制 (FAR Part 117)',
'auto': 'SAE L3 监督时间应限制在 30 分钟内',
'ims': 'DMS 应追踪连续监督时间'
},
{
'aviation': '双驾驶员交叉检查',
'auto': 'DMS + AV 传感器交叉验证',
'ims': 'DMS 检测疲劳 + AV 检测环境风险'
},
{
'aviation': 'FRMS (疲劳风险管理系统)',
'auto': 'SOTIF (ISO 21448) 包含疲劳场景',
'ims': '疲劳应作为 SOTIF 已知场景'
},
{
'aviation': '微睡眠检测 (眼动)',
'auto': 'PERCLOS + 扫视频率',
'ims': 'DMS 已有 PERCLOS, 需加扫视频率'
},
{
'aviation': '渐进式警告 (注意→警告→紧急)',
'auto': '一级→二级→紧急接管',
'ims': '已有分级, 需优化触发逻辑'
},
{
'aviation': '最小风险策略 (MRM)',
'auto': '靠边停车 + 紧急呼叫',
'ims': 'L3 必须有 MRM, DMS 触发'
}
]
if __name__ == "__main__":
print("=== SAE L3 被动疲劳检测器测试 ===")
detector = PassiveFatigueDetector()
B = 4
eye_feat = torch.randn(B, 8)
ecg_feat = torch.randn(B, 6)
output = detector(eye_feat, ecg_feat)
print(f"输入: 眼动({eye_feat.shape}) + ECG({ecg_feat.shape})")
print(f"疲劳等级: {output['fatigue_logits'].shape}")
print(f"接管就绪: {output['takeover_readiness'].shape}")
print(f"模态权重: {output['modality_weights'].shape}")
manager = SAE_L3_TakeoverManager()
scenarios = {
'清醒': (EyeMovementFeatures.from_dms(0.05, 15, 3.0, 200, 3.5),
ECGFeatures.from_rppg(72, 45)),
'轻度疲劳': (EyeMovementFeatures.from_dms(0.15, 18, 2.5, 250, 3.2),
ECGFeatures.from_rppg(68, 35)),
'中度疲劳': (EyeMovementFeatures.from_dms(0.30, 12, 2.0, 350, 3.0),
ECGFeatures.from_rppg(65, 25)),
'重度疲劳': (EyeMovementFeatures.from_dms(0.50, 8, 1.5, 500, 2.8),
ECGFeatures.from_rppg(60, 15)),
}
print(f"\n=== L3 接管就绪评估 ===")
for name, (eye, ecg) in scenarios.items():
result = manager.assess(eye, ecg)
print(f"\n{name}:")
print(f" 疲劳等级: {result['fatigue_level']}")
print(f" 接管就绪: {result['takeover_readiness']:.2f}")
print(f" 接管时间: {result['takeover_time']:.1f}s")
print(f" 动作: {result['action']}")
print(f" MRM: {result['mrm']}")
print(f" 模态权重: 眼={result['eye_weight']:.2f} ECG={result['ecg_weight']:.2f}")
print(f"\n=== 航空→汽车经验迁移 ===")
av = AviationFatigueLessons()
for i, lesson in enumerate(av.LESSONS, 1):
print(f"\n{i}. {lesson['aviation']}")
print(f" → {lesson['auto']}")
print(f" IMS: {lesson['ims']}")
print(f"\n=== 主动 vs 被动疲劳特征 ===")
print(f"{'特征':<20} {'主动疲劳':<15} {'被动疲劳'}")
print(f"{'PERCLOS':<20} {'0.3-0.5':<15} {'0.15-0.3'}}")
print(f"{'眨眼频率':<20} {'↓ (少)':<15} {'↑ (多, 轻度)'}}")
print(f"{'扫视频率':<20} {'↓':<15} {'↓↓ (显著)'}}")
print(f"{'注视时长':<20} {'↑':<15} {'↑↑ (显著)'}}")
print(f"{'HR (bpm)':<20} {'60-65':<15} {'65-70 (略高)'}}")
print(f"{'HRV RMSSD':<20} {'25-35':<15} {'15-30 (略低)'}}")
print(f"{'LF/HF':<20} {'2-3':<15} {'3-5 (交感↑)'}}")
print(f"{'发生时间':<20} {'2-4h':<15} {'20-40min'}}")
print(f"{'检测延迟':<20} {'眼动 2s':<15} {'眼动 3s + ECG 10min'}}")
|