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
| import time
from dataclasses import dataclass
from enum import Enum
class ReadinessLevel(Enum):
"""驾驶员接管就绪度等级"""
READY = 1
PARTIAL = 2
NOT_READY = 3
UNRESPONSIVE = 4
@dataclass
class ReadinessCheck:
"""每30秒一次的接管就绪度检查"""
timestamp: float
eye_open_ratio: float
gaze_on_road: bool
response_latency: float
head_pose_normal: bool
hands_on_wheel: bool
readiness: ReadinessLevel
class TakeoverReadinessMonitor:
"""
GB 44721-2026 接管就绪度监测器
要求：每30秒确认驾驶员具备接管能力
判定逻辑：
- READY: 眼开>0.5 + 视线在路 + 响应<2s + 头部正常
- PARTIAL: 1-2项不达标
- NOT_READY: 3+项不达标
- UNRESPONSIVE: 全部不达标（持续>10秒）
"""
CHECK_INTERVAL = 30
def __init__(self):
self.last_check_time = 0
self.consecutive_failures = 0
self.alert_level = 0
def check_readiness(self, dms_output: dict,
adas_status: dict) -> ReadinessCheck:
"""
执行一次接管就绪度检查
Args:
dms_output: DMS输出（眼动/头部/手部）
adas_status: ADAS状态（自动驾驶等级/车速/场景）
Returns:
ReadinessCheck: 就绪度评估结果
"""
now = time.time()
eye_ratio = dms_output.get('eye_open_ratio', 0.8)
gaze_road = dms_output.get('gaze_on_road', True)
latency = dms_output.get('response_latency', 0.5)
head_normal = dms_output.get('head_pose_normal', True)
hands_wheel = dms_output.get('hands_on_wheel', False)
checks = [
eye_ratio > 0.5,
gaze_road,
latency < 2.0,
head_normal,
hands_wheel,
]
pass_count = sum(checks)
if pass_count >= 4:
readiness = ReadinessLevel.READY
elif pass_count >= 3:
readiness = ReadinessLevel.PARTIAL
elif pass_count >= 1:
readiness = ReadinessLevel.NOT_READY
else:
readiness = ReadinessLevel.UNRESPONSIVE
check = ReadinessCheck(
timestamp=now,
eye_open_ratio=eye_ratio,
gaze_on_road=gaze_road,
response_latency=latency,
head_pose_normal=head_normal,
hands_on_wheel=hands_wheel,
readiness=readiness
)
self._update_alert_level(readiness)
return check
def _update_alert_level(self, readiness: ReadinessLevel):
"""更新告警等级"""
if readiness == ReadinessLevel.READY:
self.consecutive_failures = 0
self.alert_level = 0
elif readiness == ReadinessLevel.PARTIAL:
self.consecutive_failures += 1
if self.consecutive_failures >= 2:
self.alert_level = 1
elif readiness == ReadinessLevel.NOT_READY:
self.consecutive_failures += 1
self.alert_level = 2
else:
self.consecutive_failures += 1
if self.consecutive_failures >= 3:
self.alert_level = 3
def get_action(self) -> dict:
"""获取当前应执行的动作"""
actions = {
0: {'action': 'continue', 'description': '正常自动驾驶'},
1: {'action': 'alert', 'description': '声光提醒驾驶员保持就绪'},
2: {'action': 'warn', 'description': '强烈警告+ADAS降级+准备接管'},
3: {'action': 'emergency_stop', 'description': '紧急停车+呼叫救援'},
}
return actions[self.alert_level]
if __name__ == "__main__":
monitor = TakeoverReadinessMonitor()
print("GB 44721-2026 接管就绪度监测模拟")
print("=" * 60)
scenarios = [
{'name': '正常驾驶', 'eye': 0.85, 'gaze': True, 'latency': 0.3,
'head': True, 'hands': True},
{'name': '轻度分心', 'eye': 0.7, 'gaze': False, 'latency': 0.8,
'head': True, 'hands': True},
{'name': '疲劳', 'eye': 0.3, 'gaze': True, 'latency': 1.5,
'head': True, 'hands': True},
{'name': '严重疲劳', 'eye': 0.2, 'gaze': False, 'latency': 3.0,
'head': False, 'hands': False},
{'name': '无响应', 'eye': 0.1, 'gaze': False, 'latency': 5.0,
'head': False, 'hands': False},
]
for scenario in scenarios:
dms = {
'eye_open_ratio': scenario['eye'],
'gaze_on_road': scenario['gaze'],
'response_latency': scenario['latency'],
'head_pose_normal': scenario['head'],
'hands_on_wheel': scenario['hands'],
}
result = monitor.check_readiness(dms, {})
action = monitor.get_action()
print(f"\n场景: {scenario['name']}")
print(f" 就绪度: {result.readiness.name}")
print(f" 动作: {action['description']}")
|
