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
| from dataclasses import dataclass
from typing import List, Tuple
from enum import Enum
class DistractionType(Enum):
"""分心类型"""
NONE = "none"
PHONE_USE = "phone_use"
LOOKING_AWAY = "looking_away"
DAYDREAMING = "daydreaming"
REACHING = "reaching"
@dataclass
class DistractionEvent:
"""分心事件"""
timestamp: float
distraction_type: DistractionType
duration: float
severity: float
class DistractionDetector:
"""分心检测器"""
def __init__(self,
gaze_off_threshold: float = 0.3,
phone_gaze_regions: List[Tuple[float, float, float, float]] = None):
self.gaze_off_threshold = gaze_off_threshold
self.phone_gaze_regions = phone_gaze_regions or [
(200, 400, 300, 480),
(440, 540, 300, 480),
]
self.gaze_history: List[Tuple[float, float, float]] = []
self.in_distraction = False
self.distraction_start: float = 0
self.distraction_events: List[DistractionEvent] = []
def update(self, timestamp: float, gaze_x: float, gaze_y: float) -> Optional[DistractionEvent]:
"""更新注视数据并检测分心"""
self.gaze_history.append((timestamp, gaze_x, gaze_y))
cutoff = timestamp - 10
self.gaze_history = [(t, x, y) for t, x, y in self.gaze_history if t >= cutoff]
is_off_road = self._is_gaze_off_road(gaze_x, gaze_y)
event = None
if is_off_road and not self.in_distraction:
self.in_distraction = True
self.distraction_start = timestamp
elif not is_off_road and self.in_distraction:
duration = timestamp - self.distraction_start
if duration >= self.gaze_off_threshold:
distraction_type = self._classify_distraction_type()
severity = self._calculate_severity(duration, distraction_type)
event = DistractionEvent(
timestamp=self.distraction_start,
distraction_type=distraction_type,
duration=duration,
severity=severity
)
self.distraction_events.append(event)
self.in_distraction = False
return event
def _is_gaze_off_road(self, gaze_x: float, gaze_y: float) -> bool:
"""判断注视是否偏离前方道路"""
road_region = (200, 440, 150, 330)
x_min, x_max, y_min, y_max = road_region
return not (x_min <= gaze_x <= x_max and y_min <= gaze_y <= y_max)
def _classify_distraction_type(self) -> DistractionType:
"""分类分心类型"""
if len(self.gaze_history) < 3:
return DistractionType.LOOKING_AWAY
recent_gaze = self.gaze_history[-10:]
avg_x = np.mean([x for _, x, _ in recent_gaze])
avg_y = np.mean([y for _, _, y in recent_gaze])
for x_min, x_max, y_min, y_max in self.phone_gaze_regions:
if x_min <= avg_x <= x_max and y_min <= avg_y <= y_max:
return DistractionType.PHONE_USE
if len(recent_gaze) >= 5:
variance = np.var([(x, y) for _, x, y in recent_gaze], axis=0).sum()
if variance < 100:
return DistractionType.DAYDREAMING
return DistractionType.LOOKING_AWAY
def _calculate_severity(self, duration: float, distraction_type: DistractionType) -> float:
"""计算分心严重度"""
base_severity = min(1.0, duration / 3.0)
type_weights = {
DistractionType.PHONE_USE: 1.5,
DistractionType.LOOKING_AWAY: 1.0,
DistractionType.DAYDREAMING: 1.2,
DistractionType.REACHING: 1.3,
DistractionType.NONE: 0.0
}
weighted_severity = base_severity * type_weights.get(distraction_type, 1.0)
return min(1.0, weighted_severity)
def test_distraction_detector():
"""测试分心检测器"""
detector = DistractionDetector()
print("模拟正常驾驶...")
np.random.seed(42)
for i, t in enumerate(np.linspace(0, 30, 300)):
event = detector.update(t,
np.random.normal(320, 30),
np.random.normal(240, 25))
if event:
print(f"检测到分心: {event.distraction_type.value}, 持续{event.duration:.2f}秒")
print("\n模拟手机使用...")
for i, t in enumerate(np.linspace(30, 35, 50)):
event = detector.update(t,
np.random.normal(490, 20),
np.random.normal(390, 20))
if event:
print(f"⚠️ 检测到分心: {event.distraction_type.value}, "
f"持续{event.duration:.2f}秒, 严重度{event.severity:.2f}")
print(f"\n累计分心事件: {len(detector.distraction_events)} 次")
if __name__ == "__main__":
test_distraction_detector()
|