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
| """ EEG+眼动融合的认知分心检测 """ import numpy as np from scipy import signal from typing import Tuple
class CognitiveDistractionDetector: """认知分心检测器""" def __init__(self, fs_eeg: int = 256, fs_eye: int = 60): """ 初始化检测器 Args: fs_eeg: EEG采样率 fs_eye: 眼动采样率 """ self.fs_eeg = fs_eeg self.fs_eye = fs_eye self.freq_bands = { 'theta': (4, 8), 'alpha': (8, 12), 'beta': (12, 30) } def extract_eeg_features(self, eeg_data: np.ndarray) -> dict: """ 从EEG信号提取频段特征 Args: eeg_data: EEG数据 (channels, samples) Returns: features: 频段能量特征 """ features = {} for band_name, (f_low, f_high) in self.freq_bands.items(): sos = signal.butter(4, [f_low, f_high], btype='band', fs=self.fs_eeg, output='sos') filtered = signal.sosfiltfilt(sos, eeg_data, axis=1) power = np.mean(filtered ** 2, axis=1) features[f'{band_name}_power'] = power features['theta_alpha_ratio'] = ( features['theta_power'] / (features['alpha_power'] + 1e-6) ) return features def extract_eye_features(self, gaze_data: np.ndarray) -> dict: """ 从眼动数据提取熵值特征 Args: gaze_data: 凝视数据 (samples, 2) - (x, y)坐标 Returns: features: 眼动特征 """ velocity = np.diff(gaze_data, axis=0) speed = np.sqrt(velocity[:, 0]**2 + velocity[:, 1]**2) gaze_entropy = self._calculate_gaze_entropy(gaze_data) spatial_entropy = self._calculate_spatial_entropy(gaze_data) saccade_count = self._count_saccades(speed) saccade_rate = saccade_count / (len(gaze_data) / self.fs_eye) return { 'gaze_entropy': gaze_entropy, 'spatial_entropy': spatial_entropy, 'saccade_rate': saccade_rate, 'mean_speed': np.mean(speed) } def _calculate_gaze_entropy(self, gaze_data: np.ndarray) -> float: """计算凝视熵值""" bins = 20 x_bins = np.linspace(0, 1, bins + 1) y_bins = np.linspace(0, 1, bins + 1) hist, _ = np.histogramdd(gaze_data, bins=[x_bins, y_bins]) prob = hist.flatten() / hist.sum() entropy = -np.sum(prob * np.log2(prob + 1e-10)) return entropy def _calculate_spatial_entropy(self, gaze_data: np.ndarray) -> float: """计算空间分布熵""" grid_size = 10 grid = np.zeros((grid_size, grid_size)) x_idx = np.clip(gaze_data[:, 0] * grid_size, 0, grid_size - 1).astype(int) y_idx = np.clip(gaze_data[:, 1] * grid_size, 0, grid_size - 1).astype(int) grid[x_idx, y_idx] = 1 coverage = np.sum(grid) / (grid_size * grid_size) return coverage def _count_saccades(self, speed: np.ndarray, threshold: float = 100.0) -> int: """计算扫视次数""" saccades = speed > threshold saccade_count = np.sum(np.diff(saccades.astype(int)) > 0) return saccade_count def detect_distraction(self, eeg_data: np.ndarray, gaze_data: np.ndarray) -> dict: """ 检测认知分心 Args: eeg_data: EEG数据 (channels, samples) gaze_data: 凝视数据 (samples, 2) Returns: result: 检测结果 """ eeg_features = self.extract_eeg_features(eeg_data) eye_features = self.extract_eye_features(gaze_data) theta_alpha_high = eeg_features['theta_alpha_ratio'] > 1.5 gaze_entropy_low = eye_features['gaze_entropy'] < 2.0 saccade_rate_low = eye_features['saccade_rate'] < 2.0 is_distracted = theta_alpha_high and (gaze_entropy_low or saccade_rate_low) return { 'is_distracted': is_distracted, 'theta_alpha_ratio': eeg_features['theta_alpha_ratio'], 'gaze_entropy': eye_features['gaze_entropy'], 'saccade_rate': eye_features['saccade_rate'], 'confidence': self._calculate_confidence( eeg_features, eye_features ) } def _calculate_confidence(self, eeg_features: dict, eye_features: dict) -> float: """计算检测置信度""" scores = [] scores.append(min(eeg_features['theta_alpha_ratio'] / 2.0, 1.0)) scores.append(min(2.5 / (eye_features['gaze_entropy'] + 0.1), 1.0)) return np.mean(scores)
if __name__ == "__main__": np.random.seed(42) eeg_data = np.random.randn(14, 2560) gaze_data = np.random.rand(600, 2) detector = CognitiveDistractionDetector() result = detector.detect_distraction(eeg_data, gaze_data) print(f"认知分心检测: {result['is_distracted']}") print(f"置信度: {result['confidence']:.2f}") print(f"theta/alpha比值: {result['theta_alpha_ratio']:.2f}") print(f"凝视熵值: {result['gaze_entropy']:.2f}")
|