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
| class ContextualDMS:
"""
NHTSA情境式DMS框架原型
核心：检测结果 × 情境权重 = 风险评分
"""
def __init__(self):
self.driver_monitor = DriverStateDetector()
self.context = ContextPerceiver()
self.decision = ContextualDecision()
def assess(self, camera_frame, vehicle_data, env_data):
"""
情境化评估
Args:
camera_frame: 驾驶员摄像头帧
vehicle_data: {speed, lane, brake, steering}
env_data: {road_type, traffic_density, weather}
Returns:
risk_score: 0-1
action: 'normal' | 'warn_l1' | 'warn_l2' | 'intervene'
"""
d_state = self.driver_monitor.detect(camera_frame)
context = self.context.perceive(vehicle_data, env_data)
base_risk = self._compute_base_risk(d_state)
contextual_risk = self.decision.adjust(
base_risk=base_risk,
context=context,
driver_state=d_state
)
action = self.decision.decide(contextual_risk, context)
return {
'driver_state': d_state,
'context': context,
'base_risk': base_risk,
'contextual_risk': contextual_risk,
'action': action
}
def _compute_base_risk(self, d_state):
"""基础风险计算"""
risk = 0.0
risk += d_state['perclos'] * 0.3
if d_state['gaze_offroad']:
risk += min(d_state['gaze_offroad_duration'] / 3.0, 1.0) * 0.4
if d_state['eyes_closed']:
risk += 0.3
return min(risk, 1.0)
class ContextPerceiver:
"""情境感知模块"""
CONTEXT_WEIGHTS = {
('highway', 'sparse', 'high'): 1.5,
('highway', 'dense', 'high'): 2.0,
('urban', 'medium', 'medium'): 1.0,
('urban', 'dense', 'low'): 0.7,
('stopped', 'any', 'zero'): 0.3,
('curve', 'any', 'medium'): 1.3,
('tunnel', 'any', 'high'): 1.4,
}
def perceive(self, vehicle_data, env_data):
speed = vehicle_data['speed']
road = env_data['road_type']
traffic = env_data['traffic_density']
if speed > 100: speed_level = 'high'
elif speed > 40: speed_level = 'medium'
elif speed > 5: speed_level = 'low'
else: speed_level = 'zero'
weight = self._lookup_weight(road, traffic, speed_level)
return {
'road': road,
'traffic': traffic,
'speed': speed,
'speed_level': speed_level,
'weather': env_data['weather'],
'risk_multiplier': weight
}
def _lookup_weight(self, road, traffic, speed_level):
for (r, t, s), w in self.CONTEXT_WEIGHTS.items():
if (r == road and
(t == traffic or t == 'any') and
s == speed_level):
return w
return 1.0
class ContextualDecision:
"""情境化决策引擎"""
def adjust(self, base_risk, context, driver_state):
"""情境化风险调整"""
adjusted = base_risk * context['risk_multiplier']
if context['speed_level'] == 'zero':
if driver_state.get('eyes_closed'):
adjusted *= 0.2
if context['road'] == 'tunnel':
if driver_state.get('gaze_offroad'):
adjusted *= 1.5
return min(adjusted, 1.0)
def decide(self, risk, context):
"""决策输出"""
if context['speed_level'] == 'high':
thresholds = {'l1': 0.2, 'l2': 0.4, 'l3': 0.6}
elif context['speed_level'] == 'zero':
thresholds = {'l1': 0.6, 'l2': 0.8, 'l3': 1.0}
else:
thresholds = {'l1': 0.3, 'l2': 0.5, 'l3': 0.7}
if risk >= thresholds['l3']:
return 'intervene'
elif risk >= thresholds['l2']:
return 'warn_l2'
elif risk >= thresholds['l1']:
return 'warn_l1'
return 'normal'
|