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
| """ 消费级 EEG 耳机数据 → 座舱认知状态监测
应用场景: 1. 驾驶前状态评估(是否适合驾驶) 2. 驾驶中实时认知负荷监测 3. 疲劳/微睡眠预警 """
import numpy as np from dataclasses import dataclass from typing import Optional
@dataclass class EEGSample: """EEG 采样数据""" channels: np.ndarray fs: int timestamp: float
class ConsumerEEGIntegration: """ 消费级 EEG 耳机数据集成到座舱 DMS 硬件假设: - Neurable 类耳机(4-8通道 EEG) - 蓝牙传输到车机 - 采样率 256Hz """ def __init__(self): self.fs = 256 self.channels = ["F3", "F4", "P7", "P8", "O1", "O2"] self.buffer_size = 256 * 5 self.buffer = np.zeros((len(self.channels), self.buffer_size)) def process_eeg(self, sample: EEGSample) -> dict: """处理EEG数据""" results = {} bands = self._extract_frequency_bands(sample.channels, sample.fs) results["bands"] = bands results["vigilance_index"] = self._compute_vigilance_index(bands) results["cognitive_load"] = self._compute_cognitive_load(bands) results["fatigue_level"] = self._compute_fatigue(bands) results["recommendation"] = self._get_recommendation(results) return results def _extract_frequency_bands(self, eeg: np.ndarray, fs: int) -> dict: """提取各频段功率""" bands = {} freqs = np.fft.rfftfreq(eeg.shape[1], 1/fs) fft_power = np.abs(np.fft.rfft(eeg, axis=1))**2 band_ranges = { "delta": (1, 4), "theta": (4, 8), "alpha": (8, 13), "beta": (13, 30), "gamma": (30, 45), } for name, (low, high) in band_ranges.items(): mask = (freqs >= low) & (freqs <= high) power = np.mean(fft_power[:, mask], axis=1) bands[name] = np.mean(power) return bands def _compute_vigilance_index(self, bands: dict) -> float: """ 警觉力指数 (0-1) 基于 θ/α比 和 α/β比 """ theta = bands.get("theta", 1) alpha = bands.get("alpha", 1) beta = bands.get("beta", 0.1) theta_alpha_ratio = theta / (alpha + 1e-6) alpha_beta_ratio = alpha / (beta + 1e-6) vigilance = 1.0 / (1.0 + 0.3 * theta_alpha_ratio + 0.2 * alpha_beta_ratio) return min(1.0, max(0.0, vigilance)) def _compute_cognitive_load(self, bands: dict) -> float: """认知负荷 (0-1)""" theta = bands.get("theta", 0) beta = bands.get("beta", 0) load = theta / (beta + theta + 1e-6) return min(1.0, load * 2.0) def _compute_fatigue(self, bands: dict) -> float: """疲劳水平 (0-1)""" theta = bands.get("theta", 0) alpha = bands.get("alpha", 0) beta = bands.get("beta", 0.01) fatigue = (theta + alpha) / (beta + theta + alpha + 1e-6) return min(1.0, fatigue * 0.5) def _get_recommendation(self, results: dict) -> str: """获取建议""" v = results["vigilance_index"] load = results["cognitive_load"] fatigue = results["fatigue_level"] if fatigue > 0.7: return "FATIGUE_CRITICAL: 建议立即停车休息" elif v < 0.3: return "VIGILANCE_LOW: 警觉力严重不足,开启强提醒" elif load > 0.8: return "LOAD_HIGH: 认知负荷过高,建议简化HUD信息" elif fatigue > 0.4: return "FATIGUE_MILD: 轻度疲劳,播放提神音乐+冷风" else: return "NORMAL: 状态良好"
eeg_system = ConsumerEEGIntegration()
normal_eeg = np.random.randn(6, 256*5) * 10 normal_eeg[0:3, :] += np.sin(2*np.pi*10*np.arange(256*5)/256) * 5 sample = EEGSample(normal_eeg, 256, 0) result = eeg_system.process_eeg(sample) print("=== 正常驾驶 ===") print(f"警觉力: {result['vigilance_index']:.2f}") print(f"认知负荷: {result['cognitive_load']:.2f}") print(f"疲劳: {result['fatigue_level']:.2f}") print(f"建议: {result['recommendation']}")
fatigued_eeg = np.random.randn(6, 256*5) * 8 fatigued_eeg[0:3, :] += np.sin(2*np.pi*5*np.arange(256*5)/256) * 15 sample2 = EEGSample(fatigued_eeg, 256, 0) result2 = eeg_system.process_eeg(sample2) print("\n=== 疲劳驾驶 ===") print(f"警觉力: {result2['vigilance_index']:.2f}") print(f"认知负荷: {result2['cognitive_load']:.2f}") print(f"疲劳: {result2['fatigue_level']:.2f}") print(f"建议: {result2['recommendation']}")
|