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
| """ SAE L3 被动疲劳多模态检测
眼动特征 + ECG 特征 → 融合分类 """
import numpy as np from typing import Dict, Tuple, List from dataclasses import dataclass
@dataclass class L3FatigueConfig: """L3 被动疲劳检测配置""" eye_tracker_fps: int = 60 ecg_fs: int = 250 window_sec: float = 60.0 n_levels: int = 3
class EyeMovementFeatures: """眼动特征提取""" @staticmethod def extract(gaze_x: np.ndarray, gaze_y: np.ndarray, blink_flags: np.ndarray, fps: int = 60) -> Dict[str, float]: """ Args: gaze_x, gaze_y: 视线坐标, shape=(N,) blink_flags: 眨眼标记, shape=(N,) fps: 帧率 Returns: features: 眼动特征字典 """ N = len(gaze_x) duration = N / fps blink_rate = np.sum(blink_flags) / duration eye_openness = 1 - blink_flags perclos = np.sum(eye_openness < 0.2) / N * 100 gaze_std_x = np.std(gaze_x) gaze_std_y = np.std(gaze_y) gaze_dispersion = np.sqrt(gaze_std_x**2 + gaze_std_y**2) saccades = np.sum( np.sqrt(np.diff(gaze_x)**2 + np.diff(gaze_y)**2) > 0.1 ) saccade_rate = saccades / duration microsaccade = np.mean( np.sqrt(np.diff(gaze_x)**2 + np.diff(gaze_y)**2) ) return { 'blink_rate': blink_rate, 'perclos': perclos, 'gaze_dispersion': gaze_dispersion, 'saccade_rate': saccade_rate, 'microsaccade_amplitude': microsaccade, 'gaze_fixation_ratio': np.sum( np.sqrt(np.diff(gaze_x)**2 + np.diff(gaze_y)**2) < 0.01 ) / (N - 1) }
class ECGFeatures: """ECG 特征提取""" @staticmethod def extract(ecg: np.ndarray, fs: int = 250) -> Dict[str, float]: """ Args: ecg: ECG 信号, shape=(N,) fs: 采样率 Returns: features: ECG 特征字典 """ from scipy.signal import find_peaks r_peaks, _ = find_peaks(ecg, distance=fs*0.6, height=np.mean(ecg)*1.5) if len(r_peaks) < 2: return {'hr_mean': 0, 'hrv_rmssd': 0, 'hrv_sdnn': 0, 'lf_hf': 0} rr_intervals = np.diff(r_peaks) / fs hr_mean = 60 / np.mean(rr_intervals) if len(rr_intervals) > 0 else 0 rmssd = np.sqrt(np.mean(np.diff(rr_intervals)**2)) sdnn = np.std(rr_intervals) from numpy.fft import rfft, rfftfreq rr_interp = np.interp( np.arange(0, len(rr_intervals), 0.25), np.arange(len(rr_intervals)), rr_intervals ) spectrum = np.abs(rfft(rr_interp)) freqs = rfftfreq(len(rr_interp), 0.25) lf_power = np.sum(spectrum[(freqs >= 0.04) & (freqs <= 0.15)]) hf_power = np.sum(spectrum[(freqs >= 0.15) & (freqs <= 0.40)]) lf_hf = lf_power / max(hf_power, 1e-10) return { 'hr_mean': hr_mean, 'hrv_rmssd': rmssd * 1000, 'hrv_sdnn': sdnn * 1000, 'lf_hf_ratio': lf_hf }
class L3FatigueDetector: """L3 被动疲劳多模态检测器""" THRESHOLDS = { 'alert': {'perclos': 15, 'hrv_rmssd': 40, 'lf_hf': 1.5}, 'mild': {'perclos': 25, 'hrv_rmssd': 25, 'lf_hf': 2.0}, 'severe': {'perclos': 40, 'hrv_rmssd': 15, 'lf_hf': 3.0} } def classify(self, eye_feat: dict, ecg_feat: dict) -> str: """分类疲劳等级""" score = 0 if eye_feat['perclos'] > self.THRESHOLDS['severe']['perclos']: score += 3 elif eye_feat['perclos'] > self.THRESHOLDS['mild']['perclos']: score += 2 elif eye_feat['perclos'] > self.THRESHOLDS['alert']['perclos']: score += 1 if ecg_feat['hrv_rmssd'] < self.THRESHOLDS['severe']['hrv_rmssd']: score += 3 elif ecg_feat['hrv_rmssd'] < self.THRESHOLDS['mild']['hrv_rmssd']: score += 2 elif ecg_feat['hrv_rmssd'] < self.THRESHOLDS['alert']['hrv_rmssd']: score += 1 if score >= 5: return 'Severe Fatigue' elif score >= 3: return 'Mild Fatigue' else: return 'Alert'
if __name__ == "__main__": np.random.seed(42) N_eye = 60 * 60 N_ecg = 60 * 250 alert_gaze_x = np.random.randn(N_eye) * 0.1 alert_gaze_y = np.random.randn(N_eye) * 0.1 alert_blink = (np.random.rand(N_eye) < 0.005).astype(int) alert_ecg = np.sin(2 * np.pi * 1.2 * np.arange(N_ecg) / 250) fatigue_gaze_x = np.random.randn(N_eye) * 0.02 fatigue_gaze_y = np.random.randn(N_eye) * 0.02 fatigue_blink = (np.random.rand(N_eye) < 0.002).astype(int) fatigue_ecg = np.sin(2 * np.pi * 1.0 * np.arange(N_ecg) / 250) eye_feat_alert = EyeMovementFeatures.extract(alert_gaze_x, alert_gaze_y, alert_blink) eye_feat_fatigue = EyeMovementFeatures.extract(fatigue_gaze_x, fatigue_gaze_y, fatigue_blink) ecg_feat_alert = ECGFeatures.extract(alert_ecg) ecg_feat_fatigue = ECGFeatures.extract(fatigue_ecg) detector = L3FatigueDetector() print("=== L3 被动疲劳多模态检测 ===") print(f"\n{'指标':<25} {'清醒':<15} {'疲劳':<15}") for key in eye_feat_alert: print(f" {key:<23} {eye_feat_alert[key]:<15.2f} {eye_feat_fatigue[key]:<15.2f}") for key in ecg_feat_alert: print(f" {key:<23} {ecg_feat_alert[key]:<15.2f} {ecg_feat_fatigue[key]:<15.2f}") print(f"\n清醒 → {detector.classify(eye_feat_alert, ecg_feat_alert)}") print(f"疲劳 → {detector.classify(eye_feat_fatigue, ecg_feat_fatigue)}")
|