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 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333
| import numpy as np
class fNIRSProcessor: """ 功能性近红外光谱(fNIRS)处理 原理: - 近红外光(700-900nm)穿透头皮 - 测量氧合血红蛋白(HbO)和脱氧血红蛋白(HbR) - 认知负荷增加 → 前额叶HbO浓度上升 传感器位置: - Fp1, Fp2:前额叶左右 - F3, F4:额叶运动区 - F7, F8:额叶前部 典型配置:8通道×2波长(760nm, 850nm) """ def __init__(self): self.wavelengths = [760, 850] self.num_channels = 8 self.extinction_coeff = { 'HbO': {760: 0.5, 850: 1.0}, 'HbR': {760: 1.0, 850: 0.5} } def compute_hbo(self, intensity_760, intensity_850): """ 计算氧合血红蛋白浓度 Modified Beer-Lambert Law: ΔOD(λ) = -log(I/I0) = ε_HbO(λ)*[HbO]*d + ε_HbR(λ)*[HbR]*d 双波长求解: [HbO] = (ε_HbR(850)*ΔOD(760) - ε_HbR(760)*ΔOD(850)) / (ε_HbO(760)*ε_HbR(850) - ε_HbO(850)*ε_HbR(760)) """ od_760 = -np.log(intensity_760 / np.mean(intensity_760[:100])) od_850 = -np.log(intensity_850 / np.mean(intensity_850[:100])) epsilon = self.extinction_coeff hbo = ( epsilon['HbR'][850] * od_760 - epsilon['HbR'][760] * od_850 ) / ( epsilon['HbO'][760] * epsilon['HbR'][850] - epsilon['HbO'][850] * epsilon['HbR'][760] ) return hbo def extract_features(self, hbo_signal): """ 提取fNIRS特征 特征类型: - 均值:基线水平 - 标准差:波动程度 - 斜率:变化趋势 - 峰值:认知负荷事件 - 熵:信号复杂度 """ features = {} window_size = 30 fs = 10 window_samples = window_size * fs features['mean'] = np.mean(hbo_signal[-window_samples:]) features['std'] = np.std(hbo_signal[-window_samples:]) x = np.arange(window_samples) y = hbo_signal[-window_samples:] slope, _ = np.polyfit(x, y, 1) features['slope'] = slope peaks, _ = self.find_peaks(hbo_signal[-window_samples:]) features['peak_count'] = len(peaks) features['peak_amplitude'] = np.mean(hbo_signal[peaks]) if len(peaks) > 0 else 0 features['entropy'] = self.sample_entropy(hbo_signal[-window_samples:]) return features def find_peaks(self, signal): """ 峰值检测 认知负荷事件会引起HbO峰值 """ peaks = [] threshold = np.mean(signal) + np.std(signal) for i in range(1, len(signal) - 1): if signal[i] > signal[i-1] and signal[i] > signal[i+1]: if signal[i] > threshold: peaks.append(i) return np.array(peaks), None def sample_entropy(self, signal, m=2, r=0.2): """ 样本熵 信号复杂度指标 认知分心 → 熵值变化 """ N = len(signal) std = np.std(signal) threshold = r * std patterns = [] for i in range(N - m + 1): patterns.append(signal[i:i+m]) def count_similar(patterns, threshold): counts = [] for p in patterns: count = 0 for q in patterns: if np.max(np.abs(p - q)) < threshold: count += 1 counts.append(count) return counts counts_m = count_similar(patterns, threshold) phi_m = np.mean(np.log(counts_m)) m += 1 patterns = [] for i in range(N - m + 1): patterns.append(signal[i:i+m]) counts_m1 = count_similar(patterns, threshold) phi_m1 = np.mean(np.log(counts_m1)) sampen = phi_m - phi_m1 return sampen
class CognitiveDistractionFNIRS: """ 基于fNIRS的认知分心检测 检测流程: 1. fNIRS信号采集(前额叶) 2. HbO浓度计算 3. 特征提取 4. 机器学习分类 Euro NCAP D-04应用: - 认知分心检测 - 区分分心类型(认知、视觉-手动、听觉) """ def __init__(self): self.fnirs_processor = fNIRSProcessor() self.classifier = CognitiveDistractionClassifier() def detect(self, fnirs_raw): """ 认知分心检测 Args: fnirs_raw: dict - 'channel_1': (N, 2) 通道1的760nm和850nm信号 - ... Returns: result: dict - 'is_cognitive_distraction': bool - 'distraction_type': str - 'confidence': float - 'cognitive_load': float """ hbo_signals = {} for ch_name, raw_data in fnirs_raw.items(): intensity_760 = raw_data[:, 0] intensity_850 = raw_data[:, 1] hbo_signals[ch_name] = self.fnirs_processor.compute_hbo( intensity_760, intensity_850 ) features = {} for ch_name in ['channel_1', 'channel_2']: features[ch_name] = self.fnirs_processor.extract_features( hbo_signals[ch_name] ) is_distraction, distraction_type, confidence = self.classifier.classify( features ) cognitive_load = self.compute_cognitive_load(features) return { 'is_cognitive_distraction': is_distraction, 'distraction_type': distraction_type, 'confidence': confidence, 'cognitive_load': cognitive_load } def compute_cognitive_load(self, features): """ 计算认知负荷评分(0-100) 基于: - HbO均值(基线偏移) - HbO峰值频率(认知事件频率) - 熵值(信号复杂度) """ mean_hbo = ( features['channel_1']['mean'] + features['channel_2']['mean'] ) / 2 peak_rate = ( features['channel_1']['peak_count'] + features['channel_2']['peak_count'] ) / 2 entropy = ( features['channel_1']['entropy'] + features['channel_2']['entropy'] ) / 2 mean_norm = min(mean_hbo / 0.5, 1.0) peak_norm = min(peak_rate / 5, 1.0) entropy_norm = min(entropy / 2.0, 1.0) cognitive_load = (0.4 * mean_norm + 0.3 * peak_norm + 0.3 * entropy_norm) * 100 return cognitive_load
class CognitiveDistractionClassifier: """ 认知分心分类器 类型: - 正常驾驶 - 认知分心(心不在焉) - 视觉-手动分心 - 听觉分心 """ def __init__(self): self.threshold_hbo_mean = 0.3 self.threshold_entropy = 1.5 def classify(self, features): """ 分类 """ mean_hbo = features['channel_1']['mean'] entropy = features['channel_1']['entropy'] if mean_hbo > self.threshold_hbo_mean and entropy > self.threshold_entropy: is_distraction = True distraction_type = 'cognitive' confidence = 0.75 else: is_distraction = False distraction_type = 'normal' confidence = 0.9 return is_distraction, distraction_type, confidence
def test_euro_ncap_d04(): """ Euro NCAP D-04认知分心测试 """ detector = CognitiveDistractionFNIRS() fnirs_raw_sim = simulate_cognitive_distraction_fnirs( duration=60, cognitive_load='high' ) result = detector.detect(fnirs_raw_sim) print("\n" + "="*60) print("Euro NCAP D-04认知分心检测") print("="*60) print(f"\n是否认知分心: {result['is_cognitive_distraction']}") print(f"分心类型: {result['distraction_type']}") print(f"置信度: {result['confidence']:.2f}") print(f"认知负荷评分: {result['cognitive_load']:.1f}/100") if result['is_cognitive_distraction'] and result['confidence'] > 0.7: print("\n⚠️ Euro NCAP D-04触发:认知分心检测") print("="*60)
if __name__ == "__main__": test_euro_ncap_d04()
|