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
| """ 置信度驱动的自适应时间窗口 PERCLOS 计算
核心创新: 1. 动态调整观测窗口(置信度驱动) 2. 多数据集验证 3. L2/L3 接管就绪评估 """
import numpy as np from typing import Tuple, Dict from dataclasses import dataclass
@dataclass class PERCLOSResult: """PERCLOS 结果""" perclos_value: float fatigue_level: str confidence: float window_size: float
def adaptive_perclos_calculation( eye_openness: np.ndarray, fps: int = 30, base_window: float = 60.0, confidence_threshold: float = 0.8 ) -> PERCLOSResult: """ 自适应 PERCLOS 计算 Args: eye_openness: 眼睑开度序列 (0-1), shape=(N,) fps: 帧率 base_window: 基线时间窗口(秒) confidence_threshold: 置信度阈值 Returns: result: PERCLOS 结果 Frontiers 2026 方法: 自适应时间窗口: - 置信度高 → 缩短窗口(快速检测) - 置信度低 → 延长窗口(提高稳定性) """ closed_threshold = 0.25 is_closed = eye_openness < closed_threshold base_frames = int(base_window * fps) if len(eye_openness) < base_frames: base_perclos = np.mean(is_closed) * 100 else: base_perclos = np.mean(is_closed[-base_frames:]) * 100 variance = np.std(eye_openness[-min(len(eye_openness), base_frames):]) confidence = min(1.0, variance / 0.15) if confidence > confidence_threshold: window_size = max(30.0, base_window * 0.5) else: window_size = min(120.0, base_window * 1.5) adaptive_frames = int(window_size * fps) if len(eye_openness) >= adaptive_frames: adaptive_perclos = np.mean(is_closed[-adaptive_frames:]) * 100 else: adaptive_perclos = base_perclos if adaptive_perclos >= 70: fatigue_level = "severe" elif adaptive_perclos >= 50: fatigue_level = "moderate" elif adaptive_perclos >= 30: fatigue_level = "mild" else: fatigue_level = "normal" return PERCLOSResult( perclos_value=adaptive_perclos, fatigue_level=fatigue_level, confidence=confidence, window_size=window_size )
def l3_takeover_readiness_assessment( perclos_result: PERCLOSResult, hrv_rmssd: float, gaze_deviation: float ) -> Tuple[float, str]: """ L3 接管就绪评估 Args: perclos_result: PERCLOS 结果 hrv_rmssd: HRV RMSSD(ms) gaze_deviation: 视线偏离度 Returns: takeover_time: 预测接管时间(秒) readiness: 就绪状态 L3 应用: PERCLOS >70% → 接管时间显著延长 HRV RMSSD <20ms → 自主神经疲劳 视线偏离 >30° → 情境意识丧失 """ base_takeover = 2.5 if perclos_result.perclos_value >= 70: perclos_penalty = 8.0 elif perclos_result.perclos_value >= 50: perclos_penalty = 4.0 elif perclos_result.perclos_value >= 30: perclos_penalty = 2.0 else: perclos_penalty = 0 if hrv_rmssd < 20: hrv_penalty = 3.0 elif hrv_rmssd < 30: hrv_penalty = 1.5 else: hrv_penalty = 0 if gaze_deviation > 30: gaze_penalty = 5.0 elif gaze_deviation > 15: gaze_penalty = 2.0 else: gaze_penalty = 0 takeover_time = base_takeover + perclos_penalty + hrv_penalty + gaze_penalty if takeover_time <= 10.0: readiness = "ready" elif takeover_time <= 15.0: readiness = "conditional" else: readiness = "unready" return takeover_time, readiness
if __name__ == "__main__": normal_eye = np.random.normal(0.8, 0.05, 1800) normal_result = adaptive_perclos_calculation(normal_eye) print("="*60) print("PERCLOS 自适应计算测试") print("="*60) print(f"\n正常驾驶:") print(f" PERCLOS: {normal_result.perclos_value:.1f}%") print(f" 疲劳等级: {normal_result.fatigue_level}") print(f" 置信度: {normal_result.confidence:.2f}") print(f" 窗口大小: {normal_result.window_size:.0f}s") fatigue_eye = np.concatenate([ np.random.normal(0.8, 0.05, 1200), np.random.normal(0.2, 0.03, 600) ]) fatigue_result = adaptive_perclos_calculation(fatigue_eye) print(f"\n疲劳驾驶:") print(f" PERCLOS: {fatigue_result.perclos_value:.1f}%") print(f" 疲劳等级: {fatigue_result.fatigue_level}") print(f" 置信度: {fatigue_result.confidence:.2f}") print(f" 窗口大小: {fatigue_result.window_size:.0f}s") takeover_time, readiness = l3_takeover_readiness_assessment( fatigue_result, hrv_rmssd=15.0, gaze_deviation=20.0 ) print(f"\nL3 接管就绪评估:") print(f" 预测接管时间: {takeover_time:.1f}s") print(f" 就绪状态: {readiness}")
|