飞行员EEG认知负荷检测：机器学习方法区分飞行阶段工作量

飞行员EEG认知负荷检测：机器学习方法区分飞行阶段工作量

论文信息

  • 标题：Using machine learning methods and EEG to discriminate aircraft pilot cognitive workload during flight
  • 来源：Nature Scientific Reports
  • 年份：2023年
  • 关键成果：低/中/高负荷分类准确率80%

核心创新

本研究首次在真实飞行环境中使用EEG区分认知负荷等级：

  1. 多飞行阶段数据采集（起飞/巡航/着陆）
  2. 机器学习分类低/中/高负荷
  3. 个体化校准提升准确率
  4. 汽车座舱借鉴路径清晰

方法详解

1. 实验设计

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
Experiment_Design:
  participants: 12名飞行员
  aircraft: Cessna 172S
  
  flight_phases:
    - phase: "起飞"
      cognitive_load: "高"
      tasks: ["无线电通讯", "姿态控制", "环境感知"]
      
    - phase: "巡航"
      cognitive_load: "低"
      tasks: ["自动驾驶监控"]
      
    - phase: "着陆"
      cognitive_load: "高"
      tasks: ["仪表扫描", "跑道对准", "速度控制"]
  
  eeg_device:
    type: "wireless EEG headset"
    channels: 14
    sampling_rate: 256 Hz

2. EEG特征提取

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
# EEG认知负荷特征提取
import numpy as np
from scipy.signal import welch
from scipy.stats import entropy

class EEGCognitiveWorkloadExtractor:
    """
    EEG认知负荷特征提取
    
    基于Nature 2023论文实现
    """
    
    def __init__(self, fs: int = 256):
        self.fs = fs
        self.frequency_bands = {
            'delta': (0.5, 4),
            'theta': (4, 8),
            'alpha': (8, 13),
            'beta': (13, 30),
            'gamma': (30, 45)
        }
    
    def extract_features(self, eeg_signal: np.ndarray) -> dict:
        """
        提取EEG特征
        
        Args:
            eeg_signal: EEG信号 (channels, samples)
        
        Returns:
            features: {
                'psd_bands': 各频段功率谱密度,
                'entropy': 各通道熵值,
                'asymmetry': 左右半球不对称性
            }
        """
        features = {}
        
        # 1. 功率谱密度（PSD）
        psd_features = self._extract_psd(eeg_signal)
        features['psd_bands'] = psd_features
        
        # 2. 熵值特征
        entropy_features = self._extract_entropy(eeg_signal)
        features['entropy'] = entropy_features
        
        # 3. 左右半球不对称性
        asymmetry = self._calculate_asymmetry(eeg_signal)
        features['asymmetry'] = asymmetry
        
        return features
    
    def _extract_psd(self, signal: np.ndarray) -> dict:
        """提取各频段PSD"""
        psd_bands = {}
        
        for band_name, (low, high) in self.frequency_bands.items():
            band_power = []
            for channel in signal:
                freqs, psd = welch(channel, fs=self.fs, nperseg=256)
                # 提取频段功率
                band_mask = (freqs >= low) & (freqs <= high)
                band_power.append(np.sum(psd[band_mask]))
            
            psd_bands[band_name] = np.array(band_power)
        
        return psd_bands
    
    def _extract_entropy(self, signal: np.ndarray) -> np.ndarray:
        """提取信号熵值"""
        entropy_values = []
        
        for channel in signal:
            # 功率谱熵
            freqs, psd = welch(channel, fs=self.fs)
            psd_normalized = psd / np.sum(psd)
            spectral_entropy = entropy(psd_normalized, base=2)
            entropy_values.append(spectral_entropy)
        
        return np.array(entropy_values)
    
    def _calculate_asymmetry(self, signal: np.ndarray) -> dict:
        """计算左右半球不对称性"""
        # 假设通道顺序：左半球 0-6，右半球 7-13
        left_channels = signal[:7]
        right_channels = signal[7:]
        
        # 计算各频段不对称性
        asymmetry = {}
        
        for band_name, (low, high) in self.frequency_bands.items():
            left_power = []
            right_power = []
            
            for channel in left_channels:
                freqs, psd = welch(channel, fs=self.fs)
                band_mask = (freqs >= low) & (freqs <= high)
                left_power.append(np.sum(psd[band_mask]))
            
            for channel in right_channels:
                freqs, psd = welch(channel, fs=self.fs)
                band_mask = (freqs >= low) & (freqs <= high)
                right_power.append(np.sum(psd[band_mask]))
            
            # 不对称指数
            asymmetry[band_name] = np.log(np.mean(left_power) / (np.mean(right_power) + 1e-10))
        
        return asymmetry


class CognitiveWorkloadClassifier:
    """
    认知负荷分类器
    
    低/中/高三级分类
    """
    
    def __init__(self):
        self.model = None
        self.feature_extractor = EEGCognitiveWorkloadExtractor()
        
    def train(self, eeg_data: np.ndarray, labels: np.ndarray):
        """
        训练分类器
        
        Args:
            eeg_data: (N, channels, samples)
            labels: 'low', 'medium', 'high'
        """
        from sklearn.svm import SVC
        from sklearn.preprocessing import StandardScaler
        
        # 提取特征
        features_list = []
        for eeg in eeg_data:
            features = self.feature_extractor.extract_features(eeg)
            # 展平特征
            flat_features = []
            for band_power in features['psd_bands'].values():
                flat_features.extend(band_power)
            flat_features.extend(features['entropy'])
            for asym_val in features['asymmetry'].values():
                flat_features.append(asym_val)
            
            features_list.append(flat_features)
        
        X = np.array(features_list)
        
        # 标准化
        self.scaler = StandardScaler()
        X_scaled = self.scaler.fit_transform(X)
        
        # 训练SVM
        self.model = SVC(kernel='rbf', C=1.0, gamma='scale')
        self.model.fit(X_scaled, labels)
    
    def predict(self, eeg_signal: np.ndarray) -> str:
        """
        预测认知负荷
        
        Returns:
            workload_level: 'low' | 'medium' | 'high'
        """
        features = self.feature_extractor.extract_features(eeg_signal)
        
        # 展平特征
        flat_features = []
        for band_power in features['psd_bands'].values():
            flat_features.extend(band_power)
        flat_features.extend(features['entropy'])
        for asym_val in features['asymmetry'].values():
            flat_features.append(asym_val)
        
        X = np.array([flat_features])
        X_scaled = self.scaler.transform(X)
        
        return self.model.predict(X_scaled)[0]


# 测试代码
if __name__ == "__main__":
    # 模拟EEG数据（14通道，256Hz，10秒）
    np.random.seed(42)
    eeg_simulated = np.random.randn(14, 2560)
    
    # 高负荷：增加theta波功率
    eeg_simulated[0:7] += 0.5 * np.sin(2 * np.pi * 6 * np.arange(2560) / 256)
    
    # 提取特征
    extractor = EEGCognitiveWorkloadExtractor()
    features = extractor.extract_features(eeg_simulated)
    
    print("=== EEG特征提取结果 ===")
    print(f"Theta功率（左半球）: {features['psd_bands']['theta'][:7].mean():.4f}")
    print(f"Theta功率（右半球）: {features['psd_bands']['theta'][7:].mean():.4f}")
    print(f"Alpha不对称性: {features['asymmetry']['alpha']:.4f}")
    print(f"平均熵值: {features['entropy'].mean():.4f}")

性能指标

论文实验结果

分类任务 准确率 方法
低 vs 高负荷 92% SVM + PSD特征
低 vs 中负荷 85% 随机森林
三分类（低/中/高） 80% SVM + 熵特征

特征重要性排名

排名 特征 贡献度
1 Theta波功率 28%
2 Alpha波不对称性 22%
3 功率谱熵 18%
4 Beta波功率 15%
5 Delta波功率 10%

汽车座舱借鉴

1. DMS认知分心检测映射

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
# 汽车DMS认知分心检测（借鉴航空EEG方案）
class DMSCognitiveDistractionDetector:
    """
    DMS认知分心检测
    
    借鉴航空EEG认知负荷检测方法
    """
    
    def __init__(self):
        # 汽车座舱无EEG，用眼动替代
        self.gaze_entropy_threshold = 0.85
        self.pupil_variance_threshold = 0.20
        
    def detect(self, gaze_data: dict) -> dict:
        """
        认知分心检测
        
        Args:
            gaze_data: {
                'positions': (N, 2),
                'pupil_diameter': (N,),
                'blink_rate': float
            }
        
        Returns:
            result: {
                'workload': 'low' | 'medium' | 'high',
                'is_distracted': bool,
                'confidence': float
            }
        """
        # 1. 眼动熵值（替代EEG theta波）
        gaze_entropy = self._calculate_entropy(gaze_data['positions'])
        
        # 2. 瞳孔波动（替代EEG alpha不对称）
        pupil_var = np.var(gaze_data['pupil_diameter']) / np.mean(gaze_data['pupil_diameter'])
        
        # 3. 眨眼频率（补充指标）
        blink_rate = gaze_data['blink_rate']
        
        # 判断认知负荷等级
        if gaze_entropy > 0.85 or pupil_var > 0.25:
            workload = 'high'
            is_distracted = True
        elif gaze_entropy > 0.65 or pupil_var > 0.15:
            workload = 'medium'
            is_distracted = False
        else:
            workload = 'low'
            is_distracted = False
        
        # 计算置信度
        confidence = 1 - (gaze_entropy + pupil_var) / 2
        
        return {
            'workload': workload,
            'is_distracted': is_distracted,
            'gaze_entropy': gaze_entropy,
            'pupil_variance': pupil_var,
            'confidence': confidence
        }
    
    def _calculate_entropy(self, positions: np.ndarray) -> float:
        """计算眼动熵值"""
        grid_size = 10
        grid_x = np.floor(positions[:, 0] * grid_size).astype(int)
        grid_y = np.floor(positions[:, 1] * grid_size).astype(int)
        
        cells = grid_x * grid_size + grid_y
        counts = np.bincount(cells, minlength=grid_size**2)
        probs = counts / len(positions)
        
        from scipy.stats import entropy
        entropy_value = entropy(probs, base=2)
        max_entropy = np.log2(grid_size**2)
        
        return entropy_value / max_entropy

2. 航空→汽车迁移策略

航空方案 汽车替代方案 可行性
EEG theta波功率 眼动熵值 ✅ 已验证
Alpha不对称性 瞳孔波动 ✅ 已验证
功率谱熵 眨眼频率 🟡 中等相关
14通道EEG 单眼动摄像头 ✅ 成本优化

Euro NCAP认知分心检测对齐

场景映射

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
# Euro NCAP 2027认知分心场景
COGNITIVE_DISTRACTION_SCENARIOS = {
    "CD-01": {
        "name": "轻度认知分心",
        "detection_method": "眼动熵值 > 0.65",
        "threshold": {
            "entropy": 0.65,
            "duration": "> 10s"
        },
        "response": "一级警告"
    },
    "CD-02": {
        "name": "重度认知分心",
        "detection_method": "眼动熵值 > 0.85 + 瞳孔波动",
        "threshold": {
            "entropy": 0.85,
            "pupil_var": 0.20,
            "duration": "> 5s"
        },
        "response": "二级警告"
    }
}

IMS开发启示

1. 算法移植优先级

优先级 功能 来源 工作量
🔴 高 眼动熵值计算 航空EEG熵值迁移 2周
🔴 高 瞳孔波动检测 航空Alpha不对称迁移 1周
🟡 中 三级负荷分类 SVM模型复用 2周
🟢 低 眨眼频率分析 航空功率谱熵迁移 1周

2. 硬件需求

1
2
3
4
5
6
7
8
9
10
11
12
13
DMS_Hardware_for_Cognitive:
  camera:
    type: "IR camera"
    resolution: "1280x720"
    fps: 30
    requirements: "瞳孔直径检测"
  
  processing:
    platform: "QCS8255"
    compute: "2 TOPS for entropy + pupil"
    
  cost_estimate:
    incremental: "$5"  # 现有DMS基础上的增量

参考资源

总结

航空EEG认知负荷检测关键成果：

维度 性能
三分类准确率 80%
低/高分类准确率 92%
特征可迁移性 ✅ 眼动/瞳孔可替代
硬件成本 汽车DMS兼容

IMS开发优先级：

  • 🔴 高：眼动熵值模块实现
  • 🔴 高：瞳孔波动检测集成
  • 🟡 中：三级负荷分类器训练
  • 🟢 低：个体化校准功能

2026-07-11 研究笔记 | Nature Scientific Reports 2023

IMS研究
#DMS #Euro NCAP #驾驶员监控
飞行员EEG认知负荷检测：机器学习方法区分飞行阶段工作量
https://dapalm.com/2026/07/11/2026-07-11-pilot-eeg-cognitive-workload-machine-learning-discrimination-aviation-automotive/
作者
Mars
发布于
2026年7月11日
许可协议