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
| """
ATC→IMS 多模态疲劳检测模型
基于综述发现：
1. HRV W分量 → 认知负荷（方向盘ECG可获取）
2. 眨眼率变化 → vigilance早期标志（DMS摄像头）
3. 扫视熵 → 持续注意力下降（DMS视线追踪）
4. 方向盘操作 → 行为疲劳指标（CAN总线）
融合策略：特征级融合 + 加权决策
"""
import numpy as np
from dataclasses import dataclass
from typing import Tuple
@dataclass
class ModalityConfig:
name: str
feature_dim: int
sample_rate: int
window_sec: int
source: str
class MultiModalFatigueDetector:
"""
多模态疲劳检测器
模态权重基于ATC综述多模态融合最优组合：
- 眼动 45% (vigilance最直接)
- ECG 30% (认知负荷补充)
- 行为 25% (累积效应)
"""
def __init__(self):
self.modalities = {
"eye": ModalityConfig("眼动", 6, 30, 60, "DMS摄像头"),
"ecg": ModalityConfig("心电", 4, 250, 60, "方向盘电极"),
"behavior": ModalityConfig("行为", 3, 10, 300, "CAN总线")
}
self.weights = {"eye": 0.45, "ecg": 0.30, "behavior": 0.25}
self.fatigue_levels = {0: "正常", 1: "轻度疲劳", 2: "中度疲劳", 3: "重度疲劳"}
def extract_eye_features(self, eye_data: np.ndarray) -> np.ndarray:
"""提取眼动特征: PERCLOS, 眨眼率, 扫视熵, 瞳孔方差, 注视偏离, 闭眼时长"""
eye_openness, gaze_x, gaze_y, pupil_d = eye_data.T
perclos = np.mean(eye_openness < 0.3)
blink_events = np.diff((eye_openness < 0.3).astype(int))
blink_rate = np.sum(blink_events == -1) / (len(eye_openness) / 30 / 60)
hist, _, _ = np.histogram2d(gaze_x, gaze_y, bins=10, range=[[-1,1],[-1,1]])
hist = hist / (hist.sum() + 1e-8)
saccade_entropy = -np.sum(hist * np.log(hist + 1e-8))
pupil_var = np.std(pupil_d)
gaze_offset = np.sqrt(gaze_x**2 + gaze_y**2)
gaze_offset_ratio = np.mean(gaze_offset > 0.5)
blink_count = np.sum(blink_events == -1)
avg_closure = np.sum(eye_openness < 0.3) / max(blink_count, 1) / 30
return np.array([perclos, blink_rate, saccade_entropy,
pupil_var, gaze_offset_ratio, avg_closure])
def extract_ecg_features(self, ecg_data: np.ndarray) -> np.ndarray:
"""提取HRV特征: Mean NN, RMSSD, LF/HF, W分量"""
from scipy.signal import find_peaks, welch
peaks, _ = find_peaks(ecg_data, distance=150, prominence=0.5)
if len(peaks) < 10:
return np.zeros(4)
rr = np.diff(peaks) / 250
mean_nn = np.mean(rr)
rmssd = np.sqrt(np.mean(np.diff(rr)**2))
freqs, psd = welch(rr, fs=1/mean_nn, nperseg=min(len(rr), 64))
lf = np.sum(psd[(freqs >= 0.04) & (freqs < 0.15)])
hf = np.sum(psd[(freqs >= 0.15) & (freqs < 0.40)])
w = np.sum(psd[(freqs >= 0.01) & (freqs < 0.04)])
return np.array([mean_nn, rmssd, lf/(hf+1e-8), w])
def extract_behavior_features(self, behavior_data: np.ndarray) -> np.ndarray:
"""提取行为特征: 方向盘频率, 车道偏移, 速度波动"""
steering, lane_offset, speed = behavior_data.T
steering_freq = np.sum(np.abs(np.diff(steering)) > 1.0) / (len(steering)/10/60)
return np.array([steering_freq, np.std(lane_offset), np.std(speed)])
def detect(self, eye_data, ecg_data, behavior_data) -> dict:
"""多模态疲劳检测主函数"""
eye_f = self.extract_eye_features(eye_data)
ecg_f = self.extract_ecg_features(ecg_data)
beh_f = self.extract_behavior_features(behavior_data)
eye_score = (eye_f[0]*0.3 + eye_f[1]*0.01 +
(1-eye_f[2]/2.3)*0.2 + eye_f[4]*0.3 + eye_f[5]*0.2)
ecg_score = ((1/(ecg_f[0]+0.1))*0.3 + (1/(ecg_f[1]+0.1))*0.3 +
ecg_f[2]*0.2 + (1/(ecg_f[3]+0.001))*0.2)
beh_score = ((1/(beh_f[0]+0.1))*0.4 + beh_f[1]*0.4 + beh_f[2]*0.2)
total = (self.weights["eye"]*eye_score +
self.weights["ecg"]*ecg_score +
self.weights["behavior"]*beh_score)
level = min(int(total * 4), 3)
return {
'level': level,
'level_name': self.fatigue_levels[level],
'confidence': min(total, 1.0),
'modalities': {'eye': float(eye_score), 'ecg': float(ecg_score), 'behavior': float(beh_score)},
'features': {
'eye': {'PERCLOS': float(eye_f[0]), 'blink_rate': float(eye_f[1]), 'saccade_entropy': float(eye_f[2])},
'ecg': {'mean_nn': float(ecg_f[0]), 'rmssd': float(ecg_f[1]), 'w_component': float(ecg_f[3])},
'behavior': {'steering_freq': float(beh_f[0]), 'lane_var': float(beh_f[1])}
}
}
if __name__ == "__main__":
np.random.seed(42)
det = MultiModalFatigueDetector()
eye_normal = np.column_stack([
np.random.normal(0.85, 0.05, 1800),
np.random.normal(0, 0.15, 1800),
np.random.normal(0, 0.15, 1800),
np.random.normal(3.5, 0.2, 1800)
])
ecg_normal = np.random.normal(0, 0.1, 15000)
ecg_normal[np.random.choice(15000, 60, replace=False)] = 1.0
beh_normal = np.column_stack([
np.random.normal(0, 2, 3000),
np.random.normal(0, 0.1, 3000),
np.random.normal(60, 1, 3000)
])
result = det.detect(eye_normal, ecg_normal, beh_normal)
print(f"正常驾驶: {result['level_name']} (置信度: {result['confidence']:.3f})")
eye_fatigue = np.column_stack([
np.random.normal(0.6, 0.15, 1800),
np.random.normal(0.3, 0.3, 1800),
np.random.normal(0.3, 0.3, 1800),
np.random.normal(3.0, 0.4, 1800)
])
eye_fatigue[::30, 0] = 0.1
ecg_fatigue = np.random.normal(0, 0.08, 15000)
ecg_fatigue[np.random.choice(15000, 45, replace=False)] = 0.8
beh_fatigue = np.column_stack([
np.random.normal(0, 0.5, 3000),
np.random.normal(0.3, 0.3, 3000),
np.random.normal(60, 3, 3000)
])
result2 = det.detect(eye_fatigue, ecg_fatigue, beh_fatigue)
print(f"疲劳驾驶: {result2['level_name']} (置信度: {result2['confidence']:.3f})")
print(f" 眼动评分: {result2['modalities']['eye']:.3f}")
print(f" 心电评分: {result2['modalities']['ecg']:.3f}")
print(f" 行为评分: {result2['modalities']['behavior']:.3f}")
|
