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
| import numpy as np from scipy.stats import entropy
class EyeTrackingCognitiveFeatures: """ 眼动追踪认知分心特征提取 综述总结的关键眼动指标: 1. 注视分布熵 - 反映注意力分散程度 2. 扫视频率 - 反映信息搜索速率 3. 注视持续时间 - 反映信息处理深度 4. 瞳孔直径变化 - 反映认知负荷 5. 扫视速度 - 反映注意力切换速率 """ def __init__(self, fps: float = 30.0): self.fps = fps def gaze_distribution_entropy(self, gaze_points: np.ndarray, grid_size: int = 8) -> float: """ 注视分布熵 高熵 = 注意力分散(认知分心) 低熵 = 注意力集中 Args: gaze_points: (N, 2) 注视点序列 (x, y) grid_size: 空间网格大小 Returns: entropy_val: 注视分布的香农熵 """ H, W = grid_size, grid_size grid = np.zeros((H, W)) for x, y in gaze_points: gx = min(int(x * W), W - 1) gy = min(int(y * H), H - 1) grid[gy, gx] += 1 total = grid.sum() if total > 0: grid = grid / total grid_flat = grid.flatten() grid_flat = grid_flat[grid_flat > 0] entropy_val = entropy(grid_flat) else: entropy_val = 0.0 return float(entropy_val) def saccade_frequency(self, gaze_points: np.ndarray, velocity_threshold: float = 0.3) -> float: """ 扫视频率 Args: gaze_points: (N, 2) 归一化注视点 velocity_threshold: 扫视速度阈值 Returns: freq: 扫视/秒 """ if len(gaze_points) < 2: return 0.0 velocities = np.linalg.norm( np.diff(gaze_points, axis=0), axis=1 ) * self.fps saccades = np.sum(velocities > velocity_threshold) duration_sec = len(gaze_points) / self.fps return float(saccades / duration_sec) def fixation_duration_stats(self, gaze_points: np.ndarray, velocity_threshold: float = 0.05) -> dict: """ 注视持续时间统计 认知分心时: - 平均注视时间缩短(快速切换) - 注视时间方差增大(不规则) Args: gaze_points: (N, 2) Returns: stats: dict with mean, std, max fixation duration """ if len(gaze_points) < 2: return {'mean': 0, 'std': 0, 'max': 0} velocities = np.linalg.norm( np.diff(gaze_points, axis=0), axis=1 ) is_fixation = velocities < velocity_threshold durations = [] current_dur = 1 for i in range(1, len(is_fixation)): if is_fixation[i]: current_dur += 1 else: if current_dur > 3: durations.append(current_dur / self.fps) current_dur = 1 if current_dur > 3: durations.append(current_dur / self.fps) if not durations: return {'mean': 0, 'std': 0, 'max': 0} return { 'mean': float(np.mean(durations)), 'std': float(np.std(durations)), 'max': float(np.max(durations)), 'count': len(durations) } def pupil_diameter_variability(self, pupil_diameters: np.ndarray) -> dict: """ 瞳孔直径变异性 认知负荷↑ → 瞳孔直径↑ + 变异性↑ 注意:需去除光反射影响 Args: pupil_diameters: (N,) 瞳孔直径序列(mm) Returns: stats: dict """ from scipy.signal import detrend detrended = detrend(pupil_diameters) return { 'mean': float(np.mean(pupil_diameters)), 'std': float(np.std(detrended)), 'range': float(np.max(pupil_diameters) - np.min(pupil_diameters)), 'trend': float(np.polyfit(np.arange(len(pupil_diameters)), pupil_diameters, 1)[0]) } def composite_distraction_score(self, gaze_points: np.ndarray, pupil_diameters: np.ndarray = None) -> float: """ 综合分心评分 融合多个眼动指标的加权评分 Args: gaze_points: (N, 2) pupil_diameters: (N,) optional Returns: score: 0-1, 1=高度分心 """ entropy_val = self.gaze_distribution_entropy(gaze_points) sacc_freq = self.saccade_frequency(gaze_points) fix_stats = self.fixation_duration_stats(gaze_points) entropy_norm = min(entropy_val / 3.0, 1.0) sacc_norm = min(sacc_freq / 3.0, 1.0) fix_norm = 1.0 - min(fix_stats['mean'] / 1.0, 1.0) score = 0.4 * entropy_norm + 0.3 * sacc_norm + 0.3 * fix_norm if pupil_diameters is not None: pupil_stats = self.pupil_diameter_variability(pupil_diameters) pupil_norm = min(pupil_stats['std'] / 0.5, 1.0) score = 0.7 * score + 0.3 * pupil_norm return float(np.clip(score, 0, 1))
class EEGEyeFusionModel: """ EEG + 眼动追踪融合模型 基于系统性综述的融合策略: 1. 特征级融合(最常用,47篇中32篇) 2. 决策级融合(较少) 3. 深度学习端到端融合(新兴趋势) """ def __init__(self): self.eeg_extractor = EEGCognitiveLoadFeatures(fs=250.0) self.eye_extractor = EyeTrackingCognitiveFeatures(fps=30.0) def extract_features(self, eeg_signal: np.ndarray, gaze_points: np.ndarray, pupil_diameters: np.ndarray = None) -> dict: """ 提取EEG+眼动融合特征 Args: eeg_signal: (num_channels, num_samples) gaze_points: (N, 2) pupil_diameters: (N,) Returns: features: dict containing all features """ cli = self.eeg_extractor.cognitive_load_index(eeg_signal) ei = self.eeg_extractor.engagement_index(eeg_signal) gaze_entropy = self.eye_extractor.gaze_distribution_entropy(gaze_points) sacc_freq = self.eye_extractor.saccade_frequency(gaze_points) fix_stats = self.eye_extractor.fixation_duration_stats(gaze_points) distraction_score = self.eye_extractor.composite_distraction_score( gaze_points, pupil_diameters ) feature_vector = np.array([ cli, ei, gaze_entropy, sacc_freq, fix_stats['mean'], fix_stats['std'], distraction_score, ]) return { 'feature_vector': feature_vector, 'eeg_cli': cli, 'eeg_ei': ei, 'eye_entropy': gaze_entropy, 'eye_saccade_freq': sacc_freq, 'eye_fixation_mean': fix_stats['mean'], 'eye_distraction_score': distraction_score, } def classify_state(self, features: dict) -> dict: """ 简化的认知状态分类 综述发现:融合方案的决策边界通常为: - CLI > 0.8 + distraction > 0.6 → 认知分心 - EI < 0.3 + fixation_mean < 0.2 → 疲劳 - 正常范围 → 正常驾驶 """ cli = features['eeg_cli'] ei = features['eeg_ei'] distraction = features['eye_distraction_score'] fix_mean = features['eye_fixation_mean'] if cli > 0.8 and distraction > 0.6: state = 'cognitive_distraction' confidence = min((cli - 0.8) / 0.5 + (distraction - 0.6) / 0.4, 1.0) elif ei < 0.3 and fix_mean < 0.2: state = 'fatigue' confidence = min((0.3 - ei) / 0.3 + (0.2 - fix_mean) / 0.2, 1.0) else: state = 'normal' confidence = 1.0 - max(cli - 0.5, 0) - max(0.3 - ei, 0) return { 'state': state, 'confidence': float(np.clip(confidence, 0, 1)), 'features': features }
if __name__ == "__main__": np.random.seed(42) eeg = np.random.randn(14, 1250) * 10 t = np.arange(1250) / 250.0 for ch in range(14): eeg[ch] += 5 * np.sin(2 * np.pi * 10 * t) eeg[ch] += 3 * np.sin(2 * np.pi * 6 * t) gaze = np.random.rand(150, 2) * 0.3 + np.array([0.4, 0.3]) pupil = 4 + np.random.randn(150) * 0.3 model = EEGEyeFusionModel() features = model.extract_features(eeg, gaze, pupil) result = model.classify_state(features) print("=" * 60) print("EEG + 眼动追踪融合检测结果") print("=" * 60) print(f"\n状态: {result['state']}") print(f"置信度: {result['confidence']:.2%}") print(f"\n特征值:") print(f" EEG认知负荷指数(CLI): {result['features']['eeg_cli']:.4f}") print(f" EEG参与度指数(EI): {result['features']['eeg_ei']:.4f}") print(f" 注视分布熵: {result['features']['eye_entropy']:.4f}") print(f" 扫视频率(次/秒): {result['features']['eye_saccade_freq']:.2f}") print(f" 平均注视时间(秒): {result['features']['eye_fixation_mean']:.3f}") print(f" 综合分心评分: {result['features']['eye_distraction_score']:.4f}")
|