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
| import numpy as np
from typing import Dict
class TakeoverReadinessAssessor:
"""
L3接管准备度评估器
基于综述的关键发现：
1. 接管平均需6秒
2. 多模态融合显著优于单模态
3. NDRT参与度是L3分心核心指标
4. 认知负荷在TOR时激增
评估维度：
- 视觉准备度：视线是否在路上
- 认知准备度：是否在做NDRT
- 生理准备度：疲劳/困倦水平
- 运动准备度：手是否在方向盘附近
- 历史准备度：过去N秒的状态趋势
"""
def __init__(self):
self.weights = {
'visual': 0.30,
'cognitive': 0.25,
'physiological': 0.20,
'motor': 0.15,
'temporal': 0.10
}
def assess(self, state: Dict) -> Dict:
"""
评估接管准备度
Args:
state: 包含各维度状态的字典
Returns:
readiness: 0-1, 越高越准备好
level: 'ready'/'marginal'/'unready'
time_to_ready: 预估达到ready所需秒数
"""
visual = self._score_visual(state)
cognitive = self._score_cognitive(state)
physiological = self._score_physio(state)
motor = self._score_motor(state)
temporal = self._score_temporal(state)
readiness = (
self.weights['visual'] * visual +
self.weights['cognitive'] * cognitive +
self.weights['physiological'] * physiological +
self.weights['motor'] * motor +
self.weights['temporal'] * temporal
)
if readiness >= 0.7:
level = 'ready'
ttr = 0
elif readiness >= 0.4:
level = 'marginal'
ttr = self._estimate_time(visual, cognitive, motor)
else:
level = 'unready'
ttr = self._estimate_time(visual, cognitive, motor) + 2
return {
'readiness': readiness,
'level': level,
'time_to_ready': ttr,
'scores': {
'visual': visual,
'cognitive': cognitive,
'physiological': physiological,
'motor': motor,
'temporal': temporal
}
}
def _score_visual(self, state):
"""视觉准备度：视线是否在路上"""
gaze_on_road = state.get('gaze_on_road_ratio', 0)
return gaze_on_road
def _score_cognitive(self, state):
"""认知准备度：是否在做NDRT"""
ndrt = state.get('ndrt_engaged', False)
if ndrt:
ndrt_duration = state.get('ndrt_duration', 0)
return max(0, 1 - ndrt_duration / 30)
return 1.0
def _score_physio(self, state):
"""生理准备度"""
fatigue = state.get('fatigue_level', 0)
return 1 - fatigue
def _score_motor(self, state):
"""运动准备度：手是否在方向盘附近"""
hands_on_wheel = state.get('hands_on_wheel', False)
return 1.0 if hands_on_wheel else 0.3
def _score_temporal(self, state):
"""时序趋势：过去状态是否稳定"""
recent_readiness = state.get('recent_readiness_trend', [])
if recent_readiness:
return np.mean(recent_readiness[-10:])
return 0.5
def _estimate_time(self, visual, cognitive, motor):
"""估算达到ready所需时间"""
t = 0
if visual < 0.7:
t += 1
if cognitive < 0.7:
t += 3
if motor < 0.7:
t += 2
return t
if __name__ == "__main__":
assessor = TakeoverReadinessAssessor()
state1 = {
'gaze_on_road_ratio': 0.2,
'ndrt_engaged': True,
'ndrt_duration': 120,
'fatigue_level': 0.1,
'hands_on_wheel': False,
'recent_readiness_trend': [0.3, 0.3, 0.2, 0.2]
}
r1 = assessor.assess(state1)
print(f"场景1 (看手机): readiness={r1['readiness']:.2f}, "
f"level={r1['level']}, ttr={r1['time_to_ready']}s")
state2 = {
'gaze_on_road_ratio': 0.85,
'ndrt_engaged': False,
'fatigue_level': 0.05,
'hands_on_wheel': True,
'recent_readiness_trend': [0.8, 0.85, 0.82, 0.85]
}
r2 = assessor.assess(state2)
print(f"场景2 (正常监控): readiness={r2['readiness']:.2f}, "
f"level={r2['level']}, ttr={r2['time_to_ready']}s")
state3 = {
'gaze_on_road_ratio': 0.3,
'ndrt_engaged': True,
'ndrt_duration': 300,
'fatigue_level': 0.6,
'hands_on_wheel': False,
'recent_readiness_trend': [0.2, 0.15, 0.1, 0.1]
}
r3 = assessor.assess(state3)
print(f"场景3 (疲劳+分心): readiness={r3['readiness']:.2f}, "
f"level={r3['level']}, ttr={r3['time_to_ready']}s")
|
