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
| import numpy as np from scipy.signal import butter, filtfilt
class PupilBasedCognitiveLoad: """ 基于瞳孔直径的认知负荷检测 """ def __init__(self, fps=30): self.fps = fps self.lowcut = 0.05 self.highcut = 0.5 def calculate_cognitive_load(self, pupil_diameter_series): """ 计算认知负荷指数 Args: pupil_diameter_series: 瞳孔直径序列, shape=(N,) Returns: cognitive_load_index: 认知负荷指数 (0-1) """ clean_series = self._remove_blinks(pupil_diameter_series) filtered = self._bandpass_filter(clean_series) baseline = np.percentile(filtered, 10) dilation = (filtered - baseline) / baseline lf_hf_ratio = self._calculate_lf_hf_ratio(filtered) cognitive_load_index = 0.6 * np.mean(dilation[-30:]) + 0.4 * (lf_hf_ratio - 1) / 2 return np.clip(cognitive_load_index, 0, 1) def _remove_blinks(self, series): """ 去除眨眼伪影 眨眼特征:瞳孔直径突然降至 0 或接近 0 """ blink_threshold = np.mean(series) * 0.3 blink_mask = series < blink_threshold clean_series = series.copy() for i in range(len(clean_series)): if blink_mask[i]: prev_valid = self._find_prev_valid(clean_series, i, blink_mask) next_valid = self._find_next_valid(clean_series, i, blink_mask) if prev_valid is not None and next_valid is not None: clean_series[i] = prev_valid + (next_valid - prev_valid) * (i - prev_idx) / (next_idx - prev_idx) return clean_series def _bandpass_filter(self, signal): """ 带通滤波(0.05-0.5 Hz) 瞳孔震荡频率: - 低频(LF):0.05-0.2 Hz - 高频(HF):0.2-0.5 Hz """ nyquist = self.fps / 2 low = self.lowcut / nyquist high = self.highcut / nyquist b, a = butter(2, [low, high], btype='band') return filtfilt(b, a, signal) def _calculate_lf_hf_ratio(self, signal): """ 计算 LF/HF 比值 LF/HF 比值反映自主神经平衡: - 高比值:交感神经主导(高负荷) - 低比值:副交感神经主导(低负荷) """ n = len(signal) freq = np.fft.fftfreq(n, d=1/self.fps) spectrum = np.fft.fft(signal) lf_mask = (freq > 0.05) & (freq < 0.2) lf_power = np.sum(np.abs(spectrum[lf_mask])**2) hf_mask = (freq > 0.2) & (freq < 0.5) hf_power = np.sum(np.abs(spectrum[hf_mask])**2) if hf_power > 0: return lf_power / hf_power else: return 1.0
|