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
| import numpy as np
from dataclasses import dataclass
from typing import List, Dict
@dataclass
class ImpairmentIndicators:
"""损伤指标"""
gaze_instability: float
blink_rate_abnormal: float
pupil_response_slow: float
head_movement_erratic: float
reaction_time_slow: float
lane_keeping_poor: float
steering_jerky: float
class AlcoholImpairmentDetector:
"""酒精损伤检测器（基于行为指标）"""
def __init__(self):
self.weights = {
'gaze_instability': 0.20,
'blink_rate_abnormal': 0.15,
'pupil_response_slow': 0.15,
'head_movement_erratic': 0.10,
'reaction_time_slow': 0.15,
'lane_keeping_poor': 0.15,
'steering_jerky': 0.10,
}
self.impairment_threshold = 0.6
def detect(self, indicators: ImpairmentIndicators) -> dict:
"""检测酒精损伤
Args:
indicators: 行为指标
Returns:
{
'impairment_score': float (0-1),
'is_impaired': bool,
'confidence': float,
'primary_indicators': list
}
"""
indicator_values = {
'gaze_instability': indicators.gaze_instability,
'blink_rate_abnormal': indicators.blink_rate_abnormal,
'pupil_response_slow': indicators.pupil_response_slow,
'head_movement_erratic': indicators.head_movement_erratic,
'reaction_time_slow': indicators.reaction_time_slow,
'lane_keeping_poor': indicators.lane_keeping_poor,
'steering_jerky': indicators.steering_jerky,
}
impairment_score = sum(
indicator_values[k] * self.weights[k]
for k in self.weights
)
confidence = self._compute_confidence(indicator_values)
is_impaired = impairment_score > self.impairment_threshold
primary_indicators = [
k for k, v in sorted(indicator_values.items(), key=lambda x: -x[1])
if v > 0.5
][:3]
return {
'impairment_score': impairment_score,
'is_impaired': is_impaired,
'confidence': confidence,
'primary_indicators': primary_indicators,
'indicator_values': indicator_values
}
def _compute_confidence(self, indicators: Dict[str, float]) -> float:
"""计算置信度（指标一致性）"""
values = list(indicators.values())
std = np.std(values)
confidence = 1 - min(std, 0.5) * 2
return max(confidence, 0.3)
class RealTimeImpairmentMonitor:
"""实时损伤监控器"""
def __init__(self):
self.detector = AlcoholImpairmentDetector()
self.history = []
self.history_window = 30
def update(self,
gaze_features: dict,
facial_features: dict,
driving_features: dict) -> dict:
"""更新监控状态
Args:
gaze_features: {
'saccade_velocity': float,
'blink_rate': float,
'pupil_diameter': float
}
facial_features: {
'head_pose': tuple,
'expression': str
}
driving_features: {
'steering_angle': float,
'lane_position': float,
'reaction_time': float
}
Returns:
检测结果
"""
indicators = ImpairmentIndicators(
gaze_instability=self._analyze_gaze(gaze_features),
blink_rate_abnormal=self._analyze_blink(gaze_features),
pupil_response_slow=self._analyze_pupil(gaze_features),
head_movement_erratic=self._analyze_head(facial_features),
reaction_time_slow=self._analyze_reaction(driving_features),
lane_keeping_poor=self._analyze_lane(driving_features),
steering_jerky=self._analyze_steering(driving_features)
)
result = self.detector.detect(indicators)
self.history.append(result)
if len(self.history) > self.history_window:
self.history.pop(0)
smoothed_score = np.mean([r['impairment_score'] for r in self.history])
result['smoothed_score'] = smoothed_score
result['is_impaired'] = smoothed_score > self.detector.impairment_threshold
return result
def _analyze_gaze(self, features: dict) -> float:
"""分析视线不稳定度"""
saccade_velocity = features.get('saccade_velocity', 0)
return min(saccade_velocity / 600, 1.0)
def _analyze_blink(self, features: dict) -> float:
"""分析眨眼频率异常"""
blink_rate = features.get('blink_rate', 0)
if blink_rate < 10 or blink_rate > 30:
return 0.8
elif blink_rate < 12 or blink_rate > 25:
return 0.5
return 0.2
def _analyze_pupil(self, features: dict) -> float:
"""分析瞳孔反应"""
pupil_diameter = features.get('pupil_diameter', 4.0)
if pupil_diameter > 5:
return 0.7
return 0.2
def _analyze_head(self, features: dict) -> float:
"""分析头部运动"""
return 0.3
def _analyze_reaction(self, features: dict) -> float:
"""分析反应时间"""
reaction_time = features.get('reaction_time', 0.5)
return min(reaction_time / 1.0, 1.0)
def _analyze_lane(self, features: dict) -> float:
"""分析车道保持"""
lane_position = features.get('lane_position', 0)
return min(abs(lane_position) / 0.5, 1.0)
def _analyze_steering(self, features: dict) -> float:
"""分析方向盘操作"""
return 0.3
|