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
| """
铁路驾驶员警觉力监测系统
基于 ATC 研究方法迁移
"""
import numpy as np
from typing import List
class LocomotiveVigilance:
"""铁路机车驾驶员警觉力监测"""
def __init__(self):
self.scenario = "locomotive_cab"
self.shift_hours = 8
self.monotony_level = "EXTREME"
self.challenges = {
"extreme_monotony": "直线轨道+无变化场景 → 认知脱节",
"long_signaling_gaps": "信号灯间隔10-30km → 持续等待中注意力衰退",
"night_shifts": "夜间值班 → 昼夜节律最低点",
"solo_operation": "单人驾驶 → 无社交刺激维持警觉",
"automation_paradox": "ATP系统 → 自动化消极效应",
"vibration": "持续振动 → 催眠效应",
"noise": "恒定噪音 → 听觉疲劳",
}
def compare_modalities(self):
"""对比三种交通方式的监测需求"""
modalities = {
"ATC空管": {
"primary_signal": "EEG θ/α比",
"secondary": "眨眼率+HRV",
"environment": "室内可控",
"sensors": ["EEG头带", "眼动追踪", "座椅压力"],
"intervention": "换班提醒",
"deployment": "操作台固定",
},
"汽车DMS": {
"primary_signal": "PERCLOS+视线",
"secondary": "头部姿态+方向盘行为",
"environment": "移动座舱",
"sensors": ["IR摄像头", "方向盘ECG", "座椅压力"],
"intervention": "音频+触觉警告",
"deployment": "量产车量产",
},
"铁路驾驶室": {
"primary_signal": "PERCLOS+眨眼率",
"secondary": "HRV+操作频率",
"environment": "振动+噪音",
"sensors": ["IR摄像头(防振)", "座椅ECG", "操作日志"],
"intervention": "语音提醒+紧急制动",
"deployment": "驾驶台固定",
},
}
print("=== 三种交通方式警觉力监测对比 ===\n")
for mode, spec in modalities.items():
print(f"【{mode}】")
for k, v in spec.items():
print(f" {k}: {v}")
print()
def detect_vigilance_decrement(self, features: dict) -> dict:
"""
警觉力衰退检测
铁路特有阈值调整：
- 单调环境降低触发阈值（更早报警）
- 振动环境提高信号滤波强度
- 单人操作增加主动检查频率
"""
score = 1.0
perclos = features.get("perclos", 0)
if perclos > 0.15:
score -= 0.3
elif perclos > 0.08:
score -= 0.15
blink_rate = features.get("blink_rate", 12)
if blink_rate > 25:
score -= 0.2
control_actions = features.get("control_actions_per_min", 5)
if control_actions < 1:
score -= 0.3
hrv = features.get("hrv_lf_hf", 3.0)
if hrv < 1.5:
score -= 0.2
return {
"vigilance_score": max(0, score),
"level": "CRITICAL" if score < 0.3 else "WARNING" if score < 0.5 else "CAUTION" if score < 0.7 else "NORMAL",
"recommended_action": self._get_action(score),
}
def _get_action(self, score: float) -> str:
if score < 0.3:
return "EMERGENCY_BRAKE: 触发紧急制动+通知调度"
elif score < 0.5:
return "VOICE_ALERT: 语音唤醒+15秒内必须操作"
elif score < 0.7:
return "AUDIO_STIMULUS: 播放提示音+调整座椅振动"
return "MONITOR: 继续监测"
loco = LocomotiveVigilance()
loco.compare_modalities()
print("=== 铁路驾驶员状态测试 ===")
normal = {"perclos": 0.05, "blink_rate": 15, "control_actions_per_min": 4, "hrv_lf_hf": 2.8}
print(f"正常: {loco.detect_vigilance_decrement(normal)}")
monotonous = {"perclos": 0.12, "blink_rate": 22, "control_actions_per_min": 1, "hrv_lf_hf": 1.8}
print(f"单调驾驶: {loco.detect_vigilance_decrement(monotonous)}")
microsleep = {"perclos": 0.25, "blink_rate": 35, "control_actions_per_min": 0, "hrv_lf_hf": 1.2}
print(f"微睡眠: {loco.detect_vigilance_decrement(microsleep)}")
|