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
| """
EEG认知分心检测处理流程
论文：Mind the road: attention related neuromarkers during automated and manual simulated driving
"""
import numpy as np
import scipy.signal as signal
from scipy.integrate import simps
from typing import Tuple, Dict
class EEGCognitiveAnalyzer:
"""
EEG认知状态分析仪
功能：
1. 预处理（滤波、去噪）
2. 功率谱密度计算
3. 频段提取（δ/θ/α/β/γ）
4. 认知状态评估
参数：
- 采样率：256 Hz
- 通道数：16通道（Fp1, Fp2, F3, F4, C3, C4, P3, P4, O1, O2, F7, F8, T3, T4, T5, T6）
"""
def __init__(self, fs: int = 256, channels: int = 16):
self.fs = fs
self.channels = channels
self.freq_bands = {
'delta': (0.5, 4),
'theta': (4, 8),
'alpha': (8, 13),
'beta': (13, 30),
'gamma': (30, 100)
}
self.frontal_channels = [0, 1, 2, 3, 10, 11]
self.parietal_channels = [6, 7]
self.occipital_channels = [8, 9]
def preprocess(self, eeg_raw: np.ndarray) -> np.ndarray:
"""
EEG预处理
步骤：
1. 带通滤波（0.5-50 Hz）
2. 陷波滤波（50 Hz工频干扰）
3. 平均参考
Args:
eeg_raw: (C, T) 原始EEG信号
Returns:
eeg_clean: (C, T) 预处理后信号
"""
b_band, a_band = signal.butter(4, [0.5, 50], btype='band', fs=self.fs)
eeg_filtered = signal.filtfilt(b_band, a_band, eeg_raw, axis=1)
b_notch, a_notch = signal.iirnotch(50, 30, self.fs)
eeg_filtered = signal.filtfilt(b_notch, a_notch, eeg_filtered, axis=1)
eeg_clean = eeg_filtered - eeg_filtered.mean(axis=0, keepdims=True)
return eeg_clean
def compute_psd(self, eeg_data: np.ndarray,
nperseg: int = 1024) -> Tuple[np.ndarray, np.ndarray]:
"""
计算功率谱密度（Welch方法）
Args:
eeg_data: (C, T) EEG信号
nperseg: FFT窗口长度
Returns:
freqs: 频率轴
psd: (C, F) 功率谱密度
"""
freqs, psd = signal.welch(eeg_data, fs=self.fs,
nperseg=nperseg,
axis=1)
return freqs, psd
def extract_band_power(self, freqs: np.ndarray,
psd: np.ndarray,
band: str) -> np.ndarray:
"""
提取特定频段的功率
Args:
freqs: 频率轴
psd: (C, F) 功率谱密度
band: 频段名称（'delta', 'theta', 'alpha', 'beta', 'gamma'）
Returns:
band_power: (C,) 各通道频段功率
"""
f_low, f_high = self.freq_bands[band]
idx_band = np.logical_and(freqs >= f_low, freqs <= f_high)
band_power = simps(psd[:, idx_band], freqs[idx_band], axis=1)
return band_power
def analyze_cognitive_state(self, eeg_raw: np.ndarray) -> Dict:
"""
认知状态综合分析
Args:
eeg_raw: (C, T) 原始EEG数据，T = fs * duration
Returns:
results: 认知状态指标字典
"""
eeg_clean = self.preprocess(eeg_raw)
freqs, psd = self.compute_psd(eeg_clean)
theta_power = self.extract_band_power(freqs, psd, 'theta')
alpha_power = self.extract_band_power(freqs, psd, 'alpha')
beta_power = self.extract_band_power(freqs, psd, 'beta')
results = {
'frontal_theta': np.mean(theta_power[self.frontal_channels]),
'parietal_alpha': np.mean(alpha_power[self.parietal_channels]),
'frontal_beta': np.mean(beta_power[self.frontal_channels]),
'theta_alpha_ratio': np.mean(theta_power[self.frontal_channels]) /
(np.mean(alpha_power[self.parietal_channels]) + 1e-6),
'theta_power': theta_power,
'alpha_power': alpha_power,
'beta_power': beta_power
}
return results
if __name__ == "__main__":
"""
模拟EEG数据分析
"""
fs = 256
duration = 10
channels = 16
analyzer = EEGCognitiveAnalyzer(fs=fs, channels=channels)
t = np.linspace(0, duration, fs * duration)
np.random.seed(42)
eeg_manual = np.zeros((channels, len(t)))
eeg_auto = np.zeros((channels, len(t)))
for i in range(channels):
noise = 10 * np.random.randn(len(t))
theta = 5 * np.sin(2 * np.pi * 6 * t)
alpha = 20 * np.sin(2 * np.pi * 10 * t)
beta = 8 * np.sin(2 * np.pi * 20 * t)
if i in analyzer.parietal_channels:
eeg_manual[i] = noise + theta + 10 * np.sin(2 * np.pi * 10 * t) + beta
elif i in analyzer.frontal_channels:
eeg_manual[i] = noise + 8 * np.sin(2 * np.pi * 6 * t) + alpha + 15 * np.sin(2 * np.pi * 20 * t)
else:
eeg_manual[i] = noise + theta + alpha + beta
if i in analyzer.parietal_channels:
eeg_auto[i] = noise + theta + 30 * np.sin(2 * np.pi * 10 * t) + 5 * np.sin(2 * np.pi * 20 * t)
elif i in analyzer.frontal_channels:
eeg_auto[i] = noise + 12 * np.sin(2 * np.pi * 6 * t) + alpha + 6 * np.sin(2 * np.pi * 20 * t)
else:
eeg_auto[i] = noise + theta + alpha + beta
print("=" * 60)
print("手动驾驶 EEG 分析")
print("=" * 60)
results_manual = analyzer.analyze_cognitive_state(eeg_manual)
print(f"额叶θ功率: {results_manual['frontal_theta']:.4f} µV²")
print(f"顶叶α功率: {results_manual['parietal_alpha']:.4f} µV²")
print(f"额叶β功率: {results_manual['frontal_beta']:.4f} µV²")
print(f"θ/α比值（疲劳指数）: {results_manual['theta_alpha_ratio']:.4f}")
print("\n" + "=" * 60)
print("自动驾驶 EEG 分析")
print("=" * 60)
results_auto = analyzer.analyze_cognitive_state(eeg_auto)
print(f"额叶θ功率: {results_auto['frontal_theta']:.4f} µV²")
print(f"顶叶α功率: {results_auto['parietal_alpha']:.4f} µV²")
print(f"额叶β功率: {results_auto['frontal_beta']:.4f} µV²")
print(f"θ/α比值（疲劳指数）: {results_auto['theta_alpha_ratio']:.4f}")
print("\n" + "=" * 60)
print("手动 vs 自动驾驶 对比")
print("=" * 60)
print(f"顶叶α功率变化: {(results_auto['parietal_alpha']/results_manual['parietal_alpha']-1)*100:+.1f}%")
print(f"额叶θ功率变化: {(results_auto['frontal_theta']/results_manual['frontal_theta']-1)*100:+.1f}%")
print(f"额叶β功率变化: {(results_auto['frontal_beta']/results_manual['frontal_beta']-1)*100:+.1f}%")
print("\n✅ 结论：自动驾驶时顶叶α功率显著升高，反映认知欠载和疲劳风险")
|