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
| """ 耳内 EEG 认知分心检测
研究来源: - Predicting driver distraction using a single channel ear EEG - bioRxiv 2026.01.24
核心发现: - 认知分心在眼动指标变化前就出现在 EEG 中 - Theta/Alpha 波段是最佳指标 """
class EEGDistractionDetector: """EEG 分心检测器""" def __init__(self): self.frequency_bands = { "delta": (0.5, 4), "theta": (4, 8), "alpha": (8, 13), "beta": (13, 30), "gamma": (30, 100) } def analyze_eeg(self, eeg_signal: np.ndarray, fs: int = 250) -> dict: """ 分析 EEG 信号 Args: eeg_signal: EEG 时间序列 fs: 采样率 Returns: 各频段功率及分心判断 """ eeg_filtered = self._preprocess(eeg_signal, fs) freqs, psd = self._compute_psd(eeg_filtered, fs) band_power = {} for band_name, (low, high) in self.frequency_bands.items(): idx = np.where((freqs >= low) & (freqs <= high)) band_power[band_name] = np.mean(psd[idx]) theta_alpha_ratio = band_power["theta"] / (band_power["alpha"] + 1e-6) is_distracted = theta_alpha_ratio > 1.5 return { "band_power": band_power, "theta_alpha_ratio": theta_alpha_ratio, "is_distracted": is_distracted, "confidence": min(1.0, theta_alpha_ratio / 2.0) } def _preprocess(self, signal: np.ndarray, fs: int) -> np.ndarray: """EEG 预处理""" from scipy import signal as sp_signal b, a = sp_signal.butter(4, [0.5, 50], btype='band', fs=fs) filtered = sp_signal.filtfilt(b, a, signal) filtered = filtered - np.mean(filtered) return filtered def _compute_psd(self, signal: np.ndarray, fs: int) -> tuple: """计算功率谱密度""" from scipy import signal as sp_signal freqs, psd = sp_signal.welch(signal, fs, nperseg=min(256, len(signal))) return freqs, psd
import numpy as np
if __name__ == "__main__": detector = EEGDistractionDetector() fs = 250 t = np.linspace(0, 10, 10 * fs) normal_eeg = ( np.sin(2 * np.pi * 10 * t) * 10 + np.sin(2 * np.pi * 6 * t) * 5 + np.random.randn(len(t)) * 2 ) distracted_eeg = ( np.sin(2 * np.pi * 10 * t) * 5 + np.sin(2 * np.pi * 6 * t) * 15 + np.random.randn(len(t)) * 2 ) normal_result = detector.analyze_eeg(normal_eeg, fs) distracted_result = detector.analyze_eeg(distracted_eeg, fs) print(f"正常驾驶 Theta/Alpha: {normal_result['theta_alpha_ratio']:.2f}") print(f"认知分心 Theta/Alpha: {distracted_result['theta_alpha_ratio']:.2f}")
|