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
| """ Euro NCAP 2026 无响应驾驶员检测 基于多维度状态监测的无响应判定
判定维度: 1. 注视状态(警告后响应检测) 2. 眼睑状态(闭眼持续检测) 3. 肢体运动(方向盘/踏板操作检测) 4. 头部姿态(异常姿态检测) """
import numpy as np from typing import Dict, Tuple, Optional from dataclasses import dataclass from enum import Enum
class UnresponsiveLevel(Enum): """无响应等级""" RESPONSIVE = 0 WARNING_NO_RESPONSE = 1 EYES_CLOSED_LONG = 2 NO_BODY_MOVEMENT = 3 MRM_TRIGGER = 4
@dataclass class DriverStateMonitor: """驾驶员状态监测数据""" gaze_forward: bool eyes_closed: bool eyes_closed_duration_sec: float steering_activity: bool pedal_activity: bool no_activity_duration_sec: float head_pose_normal: bool head_pose_abnormal_type: str
class UnresponsiveDriverDetector: """ 无响应驾驶员检测器 Euro NCAP判定逻辑: 1. 分心警告后3秒内注视未恢复 → 一级无响应 2. 闭眼持续≥6秒 → 二级无响应 3. 无肢体运动≥10秒 → 二级无响应 4. 无响应持续≥15秒 → 三级MRM触发 时间参数: - warning_response_window_sec = 3 # 警告后响应窗口 - eyes_closed_threshold_sec = 6 # 闭眼判定阈值 - no_activity_threshold_sec = 10 # 无操作判定阈值 - mrm_trigger_threshold_sec = 15 # MRM触发阈值 """ def __init__(self, warning_response_window_sec: float = 3.0, eyes_closed_threshold_sec: float = 6.0, no_activity_threshold_sec: float = 10.0, mrm_trigger_threshold_sec: float = 15.0): """ Args: warning_response_window_sec: 警告后响应窗口(Euro NCAP要求3秒) eyes_closed_threshold_sec: 闭眼判定阈值(Euro NCAP要求6秒) no_activity_threshold_sec: 无操作判定阈值(Euro NCAP要求10秒) mrm_trigger_threshold_sec: MRM触发阈值(Euro NCAP要求15秒) """ self.warning_response_window = warning_response_window_sec self.eyes_closed_threshold = eyes_closed_threshold_sec self.no_activity_threshold = no_activity_threshold_sec self.mrm_trigger_threshold = mrm_trigger_threshold_sec self.current_level = UnresponsiveLevel.RESPONSIVE self.warning_sent_time: Optional[float] = None self.unresponsive_start_time: Optional[float] = None self.gaze_history: list = [] self.activity_history: list = [] def evaluate_driver_state(self, driver_state: DriverStateMonitor, current_time: float, distraction_warning_sent: bool) -> Tuple[UnresponsiveLevel, dict]: """ 评估驾驶员无响应等级 Args: driver_state: 驾驶员状态监测数据 current_time: 当前时间戳 distraction_warning_sent: 是否已发送分心警告 Returns: (level, intervention_command): 无响应等级,干预指令 """ self.gaze_history.append({ 'time': current_time, 'gaze_forward': driver_state.gaze_forward, 'eyes_closed': driver_state.eyes_closed }) self.activity_history.append({ 'time': current_time, 'steering': driver_state.steering_activity, 'pedal': driver_state.pedal_activity }) self.gaze_history = self.gaze_history[-100:] self.activity_history = self.activity_history[-300:] if distraction_warning_sent: if self.warning_sent_time is None: self.warning_sent_time = current_time elapsed_since_warning = current_time - self.warning_sent_time if elapsed_since_warning <= self.warning_response_window: if not driver_state.gaze_forward: self.current_level = UnresponsiveLevel.WARNING_NO_RESPONSE else: self.current_level = UnresponsiveLevel.RESPONSIVE self.warning_sent_time = None if driver_state.eyes_closed: if driver_state.eyes_closed_duration_sec >= self.eyes_closed_threshold: self.current_level = UnresponsiveLevel.EYES_CLOSED_LONG if self.unresponsive_start_time is None: self.unresponsive_start_time = current_time if not driver_state.steering_activity and not driver_state.pedal_activity: if driver_state.no_activity_duration_sec >= self.no_activity_threshold: self.current_level = UnresponsiveLevel.NO_BODY_MOVEMENT if self.unresponsive_start_time is None: self.unresponsive_start_time = current_time if self.current_level.value >= 2: elapsed_since_unresponsive = current_time - self.unresponsive_start_time if elapsed_since_unresponsive >= self.mrm_trigger_threshold: self.current_level = UnresponsiveLevel.MRM_TRIGGER intervention_command = self._generate_intervention_command(current_time) return self.current_level, intervention_command def _generate_intervention_command(self, current_time: float) -> dict: """生成干预指令""" command = { 'level': self.current_level.value, 'adas_adjustment': None, 'lka_activation': False, 'mrm_trigger': False, 'ecall_trigger': False, 'visual_warning': None, 'audible_warning': None } if self.current_level == UnresponsiveLevel.WARNING_NO_RESPONSE: command['adas_adjustment'] = { 'fcw_sensitivity_boost': 0.5, 'aeb_threshold_reduction': 0.2, 'ldw_sensitivity_boost': 0.3 } command['visual_warning'] = { 'icon': 'unresponsive_driver_yellow', 'text': 'DRIVER NOT RESPONDING - Please focus on road', 'duration': 'persistent' } elif self.current_level in [UnresponsiveLevel.EYES_CLOSED_LONG, UnresponsiveLevel.NO_BODY_MOVEMENT]: command['lka_activation'] = True command['adas_adjustment'] = { 'acc_deceleration_m_s2': 1.0, 'lka_gain_increase': 0.5 } command['visual_warning'] = { 'icon': 'unresponsive_driver_red', 'text': 'UNRESPONSIVE DRIVER - LANE KEEPING ACTIVE', 'duration': 'persistent' } command['audible_warning'] = { 'type': 'urgent_beep_1000hz', 'duration_sec': 30 } elif self.current_level == UnresponsiveLevel.MRM_TRIGGER: command['mrm_trigger'] = True command['ecall_trigger'] = True command['visual_warning'] = { 'icon': 'mrm_active_red_flashing', 'text': 'EMERGENCY STOP IN PROGRESS - CALLING EMERGENCY SERVICES', 'duration': 'persistent' } command['audible_warning'] = { 'type': 'emergency_siren', 'duration_sec': 'until_stop' } return command def reset_state(self): """重置状态(驾驶员恢复响应)""" self.current_level = UnresponsiveLevel.RESPONSIVE self.warning_sent_time = None self.unresponsive_start_time = None
if __name__ == "__main__": detector = UnresponsiveDriverDetector() print("=== 无响应驾驶员检测测试 ===") state1 = DriverStateMonitor( gaze_forward=False, eyes_closed=False, eyes_closed_duration_sec=0, steering_activity=True, pedal_activity=True, no_activity_duration_sec=0, head_pose_normal=True, head_pose_abnormal_type="" ) level1, cmd1 = detector.evaluate_driver_state(state1, 10.0, True) print(f"\n场景1(警告后无响应):") print(f" 等级:{level1.name}") print(f" ADAS调整:{cmd1['adas_adjustment']}") state2 = DriverStateMonitor( gaze_forward=False, eyes_closed=True, eyes_closed_duration_sec=7.0, steering_activity=False, pedal_activity=False, no_activity_duration_sec=7.0, head_pose_normal=False, head_pose_abnormal_type="back" ) level2, cmd2 = detector.evaluate_driver_state(state2, 20.0, False) print(f"\n场景2(闭眼≥6秒):") print(f" 等级:{level2.name}") print(f" LKA激活:{cmd2['lka_activation']}") level3, cmd3 = detector.evaluate_driver_state(state2, 35.0, False) print(f"\n场景3(MRM触发):") print(f" 等级:{level3.name}") print(f" MRM触发:{cmd3['mrm_trigger']}") print(f" eCall触发:{cmd3['ecall_trigger']}")
|