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
| """
铁路驾驶员疲劳监测系统架构
基于汽车DMS技术迁移
"""
class RailwayDriverMonitor:
"""
铁路驾驶室驾驶员监测系统
技术迁移自汽车DMS，针对性优化：
1. 信号灯扫视检测（铁路特有）
2. 推杆操作频率监测（替代方向盘）
3. 长时间值班疲劳累积模型
"""
def __init__(self):
self.camera_config = {
'sensor': 'OV2311 2MP RGB-IR',
'position': '操作台前方，距驾驶员~60cm',
'fov': 55,
'fps': 30,
'nir': '940nm补光（隧道场景）',
}
self.modules = {
'fatigue': FatigueDetector(),
'distraction': DistractionDetector(),
'signal_scan': SignalScanDetector(),
'operation': OperationMonitor(),
'vigilance': VigilanceAssessor(),
}
def process(self, frame, train_signals):
"""实时处理"""
results = {}
results['fatigue'] = self.modules['fatigue'].detect(frame)
results['distraction'] = self.modules['distraction'].detect(frame)
results['signal_scan'] = self.modules['signal_scan'].detect(
frame, train_signals['approaching_signal']
)
results['operation'] = self.modules['operation'].monitor(
train_signals['controls']
)
results['vigilance'] = self.modules['vigilance'].assess(results)
return results
class SignalScanDetector:
"""
信号灯扫视检测（铁路特有）
原理：
- 接近信号灯时，驾驶员应将视线移向信号灯方向
- 未扫视信号灯 = 高风险（可能错过信号）
检测方法：
1. GPS/轨道数据库获取前方信号灯位置
2. 接近信号灯时启动扫视检测窗口
3. 检测驾驶员视线是否朝向信号灯方向
4. 记录扫视频率和持续时间
"""
SCAN_WINDOW_SEC = 30
MIN_SCAN_COUNT = 1
MIN_SCAN_DURATION = 0.5
def detect(self, frame, approaching_signal):
"""
Args:
frame: 当前视频帧
approaching_signal: dict, 前方信号灯信息
{
'distance_m': 500,
'aspect': 'green/yellow/red',
'position': (x, y, z) # 信号灯3D位置
}
"""
if approaching_signal is None:
return {'status': 'no_signal_approaching'}
distance = approaching_signal['distance_m']
if distance > 500:
return {'status': 'too_far'}
signal_screen_pos = self._project_to_image(
approaching_signal['position']
)
gaze_direction = self._estimate_gaze(frame)
looking_at_signal = self._check_gaze_towards(
gaze_direction, signal_screen_pos
)
return {
'signal_approaching': True,
'distance': distance,
'signal_aspect': approaching_signal['aspect'],
'scan_detected': looking_at_signal,
'gaze_direction': gaze_direction,
'risk_level': 0 if looking_at_signal else 3,
}
class OperationMonitor:
"""
操作台操作监测（替代方向盘监测）
铁路驾驶员操作：
- 推杆（牵引/制动）
- 按钮（信号确认/警报）
- 阀门（制动）
疲劳指标：
- 操作频率下降 → 疲劳
- 反应时间延迟 → 疲劳
- 操作遗漏 → 高风险
"""
NORMAL_OPERATION_RATE = 5
FATIGUE_THRESHOLD = 2
def monitor(self, train_controls):
"""
Args:
train_controls: CAN总线信号
{
'throttle_position': 0.5,
'brake_pressure': 0,
'dead_man_button': True,
'alert_button': False,
}
"""
operation = self._detect_operation_event(train_controls)
self.operation_history.append(operation)
recent_ops = self._count_recent_ops(window_hours=1)
dmv_active = train_controls.get('dead_man_button', False)
return {
'recent_operation_rate': recent_ops,
'fatigue_risk': recent_ops < self.FATIGUE_THRESHOLD,
'dead_man_handle': dmv_active,
'last_operation_time': self.operation_history[-1]['timestamp'],
}
class VigilanceAssessor:
"""
综合警惕性评估
输入：疲劳+分心+信号扫视+操作
输出：警惕性等级(0-3)
"""
def assess(self, results):
fatigue = results.get('fatigue', {})
distraction = results.get('distraction', {})
signal = results.get('signal_scan', {})
operation = results.get('operation', {})
risk = 0
if fatigue.get('perclos', 0) > 0.3:
risk += 2
if distraction.get('off_road_ratio', 0) > 0.3:
risk += 1
if signal.get('scan_detected') == False:
risk += 3
if operation.get('fatigue_risk'):
risk += 1
level = min(risk, 3)
return {
'vigilance_level': level,
'level_name': ['正常', '轻度疲劳', '中度疲劳', '高风险'][level],
'components': {
'fatigue': fatigue,
'distraction': distraction,
'signal_scan': signal,
'operation': operation,
}
}
|