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
| """ DMS 分心检测:手机使用场景识别 基于 Euro NCAP 2026 DSM 协议 + IIHS 数据 """
import numpy as np from dataclasses import dataclass from typing import Optional, Tuple from enum import IntEnum
class DistractionLevel(IntEnum): NONE = 0 LOW = 1 MODERATE = 2 HIGH = 3
class PhoneUseType(IntEnum): NONE = 0 HANDHELD_CALL = 1 PHONE_MANIPULATION = 2 NAVIGATION = 3 HANDSFREE = 4
@dataclass class GazeMetrics: """视线指标""" gaze_x: float gaze_y: float gaze_confidence: float head_yaw: float head_pitch: float timestamp: float
@dataclass class DistractionResult: level: DistractionLevel phone_use_type: PhoneUseType gaze_off_road_duration: float confidence: float alert_required: bool alert_level: int
class DistractionDetector: """ 分心检测器 Euro NCAP 2026 DSM 分心场景: - D-01: 视线偏离道路 ≤2s → 不警告 - D-02: 视线偏离道路 >2s 且 <3.5s → 一级警告 - D-03: 视线偏离道路 ≥3.5s → 二级警告 - D-04: 手持手机至耳边 → 一级警告 ≤3s - D-05: 手机操作(打字/滑动) → 一级警告 ≤3s """ def __init__(self): self.gaze_off_threshold = 0.3 self.gaze_off_time_low = 2.0 self.gaze_off_time_moderate = 3.5 self.gaze_off_time_high = 5.0 self.phone_detection_threshold = 0.7 self.gaze_history = [] self.gaze_off_duration = 0.0 self.last_gaze_on_road = True def update(self, metrics: GazeMetrics, phone_detected: bool = False, phone_type: PhoneUseType = PhoneUseType.NONE) -> DistractionResult: """更新检测状态""" self.gaze_history.append(metrics) if len(self.gaze_history) > 300: self.gaze_history.pop(0) gaze_on_road = self._is_gaze_on_road(metrics) if not gaze_on_road: if self.last_gaze_on_road: self.gaze_off_duration = 0.0 self.gaze_off_duration += 1/30 else: self.gaze_off_duration = 0.0 self.last_gaze_on_road = gaze_on_road if phone_detected and phone_type == PhoneUseType.PHONE_MANIPULATION: return DistractionResult( level=DistractionLevel.HIGH, phone_use_type=phone_type, gaze_off_road_duration=self.gaze_off_duration, confidence=0.85, alert_required=True, alert_level=2, ) if phone_detected and phone_type == PhoneUseType.HANDHELD_CALL: return DistractionResult( level=DistractionLevel.MODERATE, phone_use_type=phone_type, gaze_off_road_duration=self.gaze_off_duration, confidence=0.80, alert_required=True, alert_level=1, ) if self.gaze_off_duration >= self.gaze_off_time_high: return DistractionResult( level=DistractionLevel.HIGH, phone_use_type=PhoneUseType.NONE, gaze_off_road_duration=self.gaze_off_duration, confidence=0.90, alert_required=True, alert_level=2, ) elif self.gaze_off_duration >= self.gaze_off_time_moderate: return DistractionResult( level=DistractionLevel.MODERATE, phone_use_type=PhoneUseType.NONE, gaze_off_road_duration=self.gaze_off_duration, confidence=0.85, alert_required=True, alert_level=1, ) elif self.gaze_off_duration >= self.gaze_off_time_low: return DistractionResult( level=DistractionLevel.LOW, phone_use_type=PhoneUseType.NONE, gaze_off_road_duration=self.gaze_off_duration, confidence=0.75, alert_required=False, alert_level=0, ) return DistractionResult( level=DistractionLevel.NONE, phone_use_type=PhoneUseType.NONE, gaze_off_road_duration=0.0, confidence=0.95, alert_required=False, alert_level=0, ) def _is_gaze_on_road(self, metrics: GazeMetrics) -> bool: """判断视线是否在道路上""" if abs(metrics.gaze_x) < self.gaze_off_threshold and \ -0.2 < metrics.gaze_y < 0.3 and \ metrics.gaze_confidence > 0.5: return True return False
if __name__ == "__main__": detector = DistractionDetector() for i in range(60): m = GazeMetrics(gaze_x=0.0, gaze_y=0.0, gaze_confidence=0.95, head_yaw=0, head_pitch=0, timestamp=i/30) r = detector.update(m) print(f"Normal: level={r.level.name}, alert={r.alert_required}") for i in range(120): m = GazeMetrics(gaze_x=0.5, gaze_y=0.4, gaze_confidence=0.85, head_yaw=20, head_pitch=-15, timestamp=i/30) r = detector.update(m) print(f"Phone gaze 4s: level={r.level.name}, duration={r.gaze_off_road_duration:.1f}s, alert=L{r.alert_level}") m = GazeMetrics(gaze_x=0.6, gaze_y=0.5, gaze_confidence=0.80, head_yaw=25, head_pitch=-20, timestamp=0) r = detector.update(m, phone_detected=True, phone_type=PhoneUseType.PHONE_MANIPULATION) print(f"Phone manipulation: level={r.level.name}, alert=L{r.alert_level}")
|