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
| """
单通道耳内EEG认知分心解码完整示例
"""
import numpy as np
from sklearn.discriminant_analysis import LinearDiscriminantAnalysis
from sklearn.model_selection import StratifiedKFold
import matplotlib.pyplot as plt
class EarEEGDecoder:
"""耳内EEG认知分心解码器"""
def __init__(self, sfreq=250, epoch_window=(-0.5, 2.0)):
self.sfreq = sfreq
self.epoch_window = epoch_window
self.preprocessor = EEGPreprocessing(sfreq)
def decode_cognitive_load(self, raw_eeg, events, labels):
"""
解码认知负荷
Args:
raw_eeg: 原始EEG数据 (n_samples,)
events: 事件标记时间点 (n_events,)
labels: 认知负荷标签 (n_events,) 0=低, 1=高
Returns:
results: 解码结果字典
"""
clean_eeg = self.preprocessor.preprocess_pipeline(raw_eeg)
epochs = self._extract_epochs(clean_eeg, events)
features = self._extract_features(epochs)
decoding_curve = self._time_resolved_decode(features, labels)
gen_matrix = self._temporal_generalization(features, labels)
return {
'decoding_accuracy': decoding_curve,
'generalization_matrix': gen_matrix,
'peak_accuracy': np.max(decoding_curve),
'peak_latency': np.argmax(decoding_curve) * 4
}
def _extract_epochs(self, eeg, events):
"""提取事件相关分段"""
pre_samples = int(abs(self.epoch_window[0]) * self.sfreq)
post_samples = int(self.epoch_window[1] * self.sfreq)
epoch_len = pre_samples + post_samples
epochs = []
for event_time in events:
start = event_time - pre_samples
end = event_time + post_samples
if start >= 0 and end < len(eeg):
epochs.append(eeg[start:end])
return np.array(epochs)
def _extract_features(self, epochs):
"""提取频域特征"""
from scipy.fft import fft
n_epochs, n_times = epochs.shape
features = np.zeros((n_epochs, n_times, 5))
freqs = np.fft.fftfreq(n_times, 1/self.sfreq)
for i, epoch in enumerate(epochs):
fft_vals = np.abs(fft(epoch))
bands = {
'delta': (1, 4),
'theta': (4, 8),
'alpha': (8, 12),
'beta': (12, 30),
'gamma': (30, 40)
}
for j, (band, (low, high)) in enumerate(bands.items()):
mask = (freqs >= low) & (freqs < high)
features[i, :, j] = fft_vals[mask].mean() if mask.any() else 0
return features
def _time_resolved_decode(self, features, labels):
"""时序解码"""
n_epochs, n_times, n_features = features.shape
decoding_acc = []
cv = StratifiedKFold(n_splits=5, shuffle=True)
for t in range(n_times):
X_t = features[:, t, :]
scores = []
for train_idx, test_idx in cv.split(X_t, labels):
X_train, X_test = X_t[train_idx], X_t[test_idx]
y_train, y_test = labels[train_idx], labels[test_idx]
lda = LinearDiscriminantAnalysis()
lda.fit(X_train, y_train)
scores.append(lda.score(X_test, y_test))
decoding_acc.append(np.mean(scores))
return np.array(decoding_acc)
def _temporal_generalization(self, features, labels):
"""时间泛化矩阵"""
n_epochs, n_times, n_features = features.shape
models = []
for t in range(n_times):
X_t = features[:, t, :]
lda = LinearDiscriminantAnalysis()
lda.fit(X_t, labels)
models.append(lda)
gen_matrix = np.zeros((n_times, n_times))
for i, model in enumerate(models):
for j in range(n_times):
X_test = features[:, j, :]
gen_matrix[i, j] = model.score(X_test, labels)
return gen_matrix
if __name__ == "__main__":
np.random.seed(42)
n_samples = 60000
n_events = 100
raw_eeg = np.random.randn(n_samples) * 50e-6
events = np.sort(np.random.randint(1000, n_samples-1000, n_events))
labels = np.random.randint(0, 2, n_events)
decoder = EarEEGDecoder()
results = decoder.decode_cognitive_load(raw_eeg, events, labels)
print(f"峰值准确率: {results['peak_accuracy']:.1%}")
print(f"峰值延迟: {results['peak_latency']} ms")
|