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
| import numpy as np from typing import Dict, Optional from dataclasses import dataclass from enum import Enum import time
class UnresponsiveState(Enum): """无响应状态枚举""" NORMAL = "normal" DISTRACTED = "distracted" WARNED = "warned" UNRESPONSIVE = "unresponsive" CRITICAL = "critical"
@dataclass class DriverState: """驾驶员状态""" timestamp: float eyes_closed: bool eyes_closed_duration: float gaze_on_road: bool head_movement_detected: bool hand_on_wheel: bool
@dataclass class VehicleState: """车辆状态""" timestamp: float speed: float steering_angle: float throttle: float brake: float lane_position: float
class UnresponsiveDriverDetector: """ 无响应驾驶员检测器 Euro NCAP 2026 合规实现 """ def __init__(self): self.gaze_return_timeout = 3.0 self.eyes_closed_threshold = 6.0 self.no_movement_threshold = 10.0 self.current_state = UnresponsiveState.NORMAL self.state_start_time: Optional[float] = None self.warning_issued_time: Optional[float] = None self.driver_state_history: list = [] self.vehicle_state_history: list = [] def update(self, driver_state: DriverState, vehicle_state: VehicleState) -> Dict: """ 更新检测状态 Args: driver_state: 驾驶员状态 vehicle_state: 车辆状态 Returns: result: 检测结果 """ current_time = driver_state.timestamp if vehicle_state.speed < 50: return self._normal_result("Speed below threshold") self.driver_state_history.append(driver_state) self.vehicle_state_history.append(vehicle_state) max_history = 300 if len(self.driver_state_history) > max_history: self.driver_state_history.pop(0) self.vehicle_state_history.pop(0) result = self._detect(current_time, driver_state, vehicle_state) return result def _detect(self, current_time: float, driver_state: DriverState, vehicle_state: VehicleState) -> Dict: """执行检测""" if driver_state.eyes_closed and driver_state.eyes_closed_duration >= self.eyes_closed_threshold: return self._transition_state( UnresponsiveState.CRITICAL, current_time, "Eyes closed for >= 6 seconds" ) if self.current_state == UnresponsiveState.WARNED: if self.warning_issued_time is not None: time_since_warning = current_time - self.warning_issued_time if time_since_warning >= self.gaze_return_timeout: if not driver_state.gaze_on_road: return self._transition_state( UnresponsiveState.UNRESPONSIVE, current_time, "Gaze not returned after warning" ) no_movement = self._check_no_movement(driver_state, vehicle_state) if no_movement['detected'] and no_movement['duration'] >= self.no_movement_threshold: return self._transition_state( UnresponsiveState.UNRESPONSIVE, current_time, "No movement detected" ) if driver_state.gaze_on_road and not driver_state.eyes_closed: return self._transition_state( UnresponsiveState.NORMAL, current_time, "Driver responsive" ) elif not driver_state.gaze_on_road: if self.current_state == UnresponsiveState.NORMAL: return self._transition_state( UnresponsiveState.DISTRACTED, current_time, "Driver distracted" ) return self._current_result() def _check_no_movement(self, driver_state: DriverState, vehicle_state: VehicleState) -> Dict: """检测无动作""" recent_driver = self.driver_state_history[-100:] head_movement = any(s.head_movement_detected for s in recent_driver) hand_on_wheel = any(s.hand_on_wheel for s in recent_driver) recent_vehicle = self.vehicle_state_history[-100:] steering_change = any( abs(v.steering_angle - recent_vehicle[0].steering_angle) > 5 for v in recent_vehicle ) throttle_input = any(v.throttle > 0.1 for v in recent_vehicle) brake_input = any(v.brake > 0.1 for v in recent_vehicle) no_movement = not (head_movement or hand_on_wheel or steering_change or throttle_input or brake_input) return { 'detected': no_movement, 'duration': len(recent_driver) / 30.0 if no_movement else 0.0 } def _transition_state(self, new_state: UnresponsiveState, timestamp: float, reason: str) -> Dict: """状态转换""" old_state = self.current_state self.current_state = new_state if new_state == UnresponsiveState.WARNED: self.warning_issued_time = timestamp return { 'state': new_state.value, 'state_changed': old_state != new_state, 'previous_state': old_state.value, 'timestamp': timestamp, 'reason': reason, 'esf_required': new_state == UnresponsiveState.CRITICAL } def _current_result(self) -> Dict: """返回当前状态""" return { 'state': self.current_state.value, 'state_changed': False, 'esf_required': False } def _normal_result(self, reason: str) -> Dict: """返回正常状态""" return { 'state': UnresponsiveState.NORMAL.value, 'state_changed': False, 'reason': reason, 'esf_required': False }
|