FATED：铁路驾驶员疲劳监测计算机视觉框架——从单车预警到连续就绪度评估

Mosaic ATM 在联邦铁路局（FRA）SBIR 项目 FATED 中开发了基于计算机视觉的铁路驾驶员疲劳监测框架。核心创新：将疲劳从”是/否”二元判定转为”连续就绪度”评估，融合面部+体态多层级信号，30小时视频60人验证。本文深度解析其三层架构、个体基线校准及对 IMS 跨座舱部署的启示。

1 项目背景

1.1 FATED 概况

项目	内容
全称	Fatigue Assessment for Transportation Engineer Determination
赞助	联邦铁路局（FRA）
类型	SBIR Phase I
承包商	Mosaic ATM
目标	铁路驾驶员疲劳监测计算机视觉框架
数据	30+小时视频，60名参与者
前序项目	VILMAS（汽车 DMS 研究）

1.2 核心问题

铁路驾驶员疲劳监测面临三大挑战：

挑战	说明
个体差异	不同人疲劳表现不同，通用阈值不可靠
误报困扰	过多误报导致用户失去信任
渐进性	疲劳不是开关，是渐变过程

核心创新： 将疲劳从”是/否”二元判定转为连续就绪度评估。

2 三层架构设计

graph TD
    A[视频输入] --> B[第一层: 低级视觉信号]
    
    B --> B1[眼部状态]
    B --> B2[视线方向]
    B --> B3[面部关键点]
    B --> B4[头部姿态]
    B --> B5[体态估计]
    
    B1 --> C[第二层: 中级行为模式]
    B2 --> C
    B3 --> C
    B4 --> C
    B5 --> C
    
    C --> C1[眨眼频率+持续时间]
    C --> C2[持续闭眼]
    C --> C3[打哈欠]
    C --> C4[头部点头]
    C --> C5[视线变化]
    C --> C6[体态变化]
    
    C1 --> D[第三层: 高级就绪度评估]
    C2 --> D
    C3 --> D
    C4 --> D
    C5 --> D
    C6 --> D
    
    D --> E[连续就绪度评分]
    E --> F[疲劳进展趋势]
    E --> G[干预建议]

2.1 第一层：低级视觉信号（帧级）

信号	提取方法	帧率
眼部状态	面部关键点 → EAR	30fps
视线方向	瞳孔+头部姿态	30fps
面部关键点	98点 landmarks	30fps
头部姿态	参考点 → Pitch/Yaw/Roll	30fps
体态估计	骨架关键点	15fps

2.2 第二层：中级行为模式（时序）

import numpy as np
from collections import deque
from typing import List, Dict

class FATEDBehaviorAnalyzer:
    """
    FATED 中级行为分析器
    
    从帧级信号提取时序行为模式
    
    参考: Mosaic ATM FATED Phase I, FRA SBIR
    """
    
    def __init__(self, window_sec: int = 60):
        self.window_sec = window_sec
        self.window_frames = window_sec * 30  # 30fps
        
        # 滑动窗口存储
        self.ear_history = deque(maxlen=self.window_frames)
        self.gaze_history = deque(maxlen=self.window_frames)
        self.head_pose_history = deque(maxlen=self.window_frames)
        self.mouth_open_history = deque(maxlen=self.window_frames)
        self.body_pose_history = deque(maxlen=self.window_frames)
        
    def update(self, frame_features: Dict) -> Dict:
        """更新行为分析"""
        self.ear_history.append(frame_features.get('ear', 0.3))
        self.gaze_history.append(frame_features.get('gaze', (0, 0)))
        self.head_pose_history.append(frame_features.get('head_pose', (0, 0, 0)))
        self.mouth_open_history.append(frame_features.get('mar', 0.2))
        self.body_pose_history.append(frame_features.get('body_pose', (0, 0)))
        
        if len(self.ear_history) < self.window_frames // 2:
            return {'behaviors': {}, 'ready': False}
        
        behaviors = {
            'blink_frequency': self._compute_blink_freq(),
            'blink_duration_mean': self._compute_blink_duration(),
            'perclos_60s': self._compute_perclos(),
            'prolonged_eye_closure': self._detect_prolonged_closure(),
            'yawn_frequency': self._compute_yawn_freq(),
            'head_nodding': self._detect_head_nodding(),
            'gaze_deviation': self._compute_gaze_deviation(),
            'posture_change_rate': self._compute_posture_change(),
        }
        
        return {
            'behaviors': behaviors,
            'ready': True,
            'window_sec': self.window_sec,
        }
    
    def _compute_blink_freq(self) -> float:
        """眨眼频率（次/分钟）"""
        ear = np.array(self.ear_history)
        # 检测眨眼（EAR下降再上升）
        threshold = 0.2
        below = ear < threshold
        # 计算下降沿数量
        diff = np.diff(below.astype(int))
        blinks = np.sum(diff == 1)
        return blinks * (60.0 / self.window_sec)
    
    def _compute_blink_duration(self) -> float:
        """平均眨眼持续时间（ms）"""
        ear = np.array(self.ear_history)
        threshold = 0.2
        below = ear < threshold
        
        durations = []
        in_blink = False
        start = 0
        
        for i, b in enumerate(below):
            if b and not in_blink:
                in_blink = True
                start = i
            elif not b and in_blink:
                in_blink = False
                durations.append((i - start) / 30.0 * 1000)  # ms
        
        return np.mean(durations) if durations else 0
    
    def _compute_perclos(self) -> float:
        """PERCLOS（60秒窗口闭眼百分比）"""
        ear = np.array(self.ear_history)
        threshold = 0.2
        closed_ratio = np.sum(ear < threshold) / len(ear)
        return closed_ratio * 100
    
    def _detect_prolonged_closure(self) -> int:
        """检测持续闭眼事件（>2秒）"""
        ear = np.array(self.ear_history)
        threshold = 0.2
        below = ear < threshold
        
        count = 0
        in_closure = False
        start = 0
        
        for i, b in enumerate(below):
            if b and not in_closure:
                in_closure = True
                start = i
            elif not b and in_closure:
                in_closure = False
                duration = (i - start) / 30.0
                if duration > 2.0:
                    count += 1
        
        return count
    
    def _compute_yawn_freq(self) -> float:
        """打哈欠频率（次/分钟）"""
        mar = np.array(self.mouth_open_history)
        threshold = 0.5
        above = mar > threshold
        diff = np.diff(above.astype(int))
        yawns = np.sum(diff == 1)
        return yawns * (60.0 / self.window_sec)
    
    def _detect_head_nodding(self) -> bool:
        """检测头部点头"""
        head = np.array(self.head_pose_history)
        pitch = head[:, 0]  # Pitch
        
        # 检测低频周期性下降
        if len(pitch) < 60:
            return False
        
        # 简化：检测连续下降超过10度
        pitch_diff = np.diff(pitch)
        consecutive_drop = np.sum(pitch_diff < -0.5)
        return consecutive_drop > 10
    
    def _compute_gaze_deviation(self) -> float:
        """视线偏离前方的时间比"""
        gaze = np.array(self.gaze_history)
        forward = (np.abs(gaze[:, 0]) < 15) & (np.abs(gaze[:, 1]) < 15)
        return (1 - np.mean(forward)) * 100
    
    def _compute_posture_change(self) -> float:
        """体态变化率"""
        body = np.array(self.body_pose_history)
        if len(body) < 2:
            return 0
        changes = np.linalg.norm(np.diff(body, axis=0), axis=1)
        return np.mean(changes)


# 测试
if __name__ == "__main__":
    np.random.seed(42)
    analyzer = FATEDBehaviorAnalyzer(window_sec=60)
    
    # 模拟正常状态
    for _ in range(1800):
        analyzer.update({
            'ear': np.random.normal(0.30, 0.03),
            'gaze': (np.random.normal(0, 5), np.random.normal(0, 3)),
            'head_pose': (np.random.normal(5, 2), 0, 0),
            'mar': np.random.normal(0.15, 0.05),
            'body_pose': (0, 0),
        })
    
    result_normal = analyzer.update({
        'ear': 0.30, 'gaze': (0, 0), 'head_pose': (5, 0, 0),
        'mar': 0.15, 'body_pose': (0, 0)
    })
    
    # 模拟疲劳状态
    analyzer2 = FATEDBehaviorAnalyzer(window_sec=60)
    for _ in range(1800):
        analyzer2.update({
            'ear': np.random.choice([0.15, 0.30], p=[0.35, 0.65]),
            'gaze': (np.random.normal(15, 10), np.random.normal(5, 5)),
            'head_pose': (np.random.normal(15, 8), 0, 0),
            'mar': np.random.choice([0.6, 0.2], p=[0.1, 0.9]),
            'body_pose': (np.random.normal(0, 0.5), 0),
        })
    
    result_fatigue = analyzer2.update({
        'ear': 0.15, 'gaze': (20, 5), 'head_pose': (20, 0, 0),
        'mar': 0.6, 'body_pose': (0.5, 0)
    })
    
    print("=== 正常状态行为分析 ===")
    for k, v in result_normal['behaviors'].items():
        print(f"  {k}: {v:.2f}")
    
    print("\n=== 疲劳状态行为分析 ===")
    for k, v in result_fatigue['behaviors'].items():
        print(f"  {k}: {v:.2f}")

2.3 第三层：高级就绪度评估

FATED 的核心创新是将多个行为指标融合为连续就绪度评分：

就绪度等级	行为特征	干预策略
90-100（充分就绪）	眨眼正常，视线稳定，体态端正	无需干预
70-89（轻度下降）	眨眼频率略增，偶有视线偏离	记录，持续监测
50-69（中度下降）	PERCLOS 升高，偶有打哈欠	提醒，建议活动
30-49（显著下降）	持续闭眼，头部点头	警告，建议休息
0-29（严重不足）	频繁闭眼，体态垮塌	紧急，立即停车

3 个体基线校准

3.1 核心发现

FATED Phase I 最重要发现之一：不存在适用于所有人的通用阈值。

问题	通用阈值方案	FATED 个体基线方案
眨眼频率	固定 20次/分	个体均值±2σ
PERCLOS	固定 15%	个体基线×1.5
头部姿态	固定 15°	个体正常范围

3.2 基线校准算法

class IndividualBaseline:
    """
    个体基线校准
    
    FATED 核心策略: 个性化阈值而非通用阈值
    """
    
    def __init__(self, calibration_min: int = 5):
        self.calibration_frames = calibration_min * 60 * 30  # 5分钟@30fps
        self.baseline_features = {}
        self.is_calibrated = False
    
    def calibrate(self, feature_history: Dict[str, np.ndarray]) -> bool:
        """
        从历史数据建立个体基线
        """
        for feat_name, values in feature_history.items():
            if len(values) >= self.calibration_frames:
                self.baseline_features[feat_name] = {
                    'mean': np.mean(values),
                    'std': np.std(values),
                    'p5': np.percentile(values, 5),
                    'p95': np.percentile(values, 95),
                }
        
        if len(self.baseline_features) >= 4:
            self.is_calibrated = True
            return True
        return False
    
    def compute_deviation(self, current_value: float, 
                          feat_name: str) -> float:
        """计算当前值与基线的偏差"""
        if not self.is_calibrated or feat_name not in self.baseline_features:
            return 0.0
        
        baseline = self.baseline_features[feat_name]
        if baseline['std'] == 0:
            return 0.0
        
        # Z-score
        z = (current_value - baseline['mean']) / baseline['std']
        return z
    
    def compute_readiness(self, current_features: Dict) -> float:
        """计算就绪度评分 (0-100)"""
        if not self.is_calibrated:
            return 50.0  # 未校准返回中性值
        
        scores = []
        weights = {
            'ear': 0.25,
            'blink_freq': 0.20,
            'perclos': 0.25,
            'gaze_dev': 0.15,
            'head_pose': 0.15,
        }
        
        for feat, weight in weights.items():
            if feat in current_features and feat in self.baseline_features:
                z = self.compute_deviation(current_features[feat], feat)
                # Z-score → 0-100 分（z=0 → 100, z=3 → 0）
                score = max(0, min(100, 100 - z * 25))
                scores.append((score, weight))
        
        if not scores:
            return 50.0
        
        total = sum(s * w for s, w in scores)
        total_weight = sum(w for _, w in scores)
        return total / total_weight if total_weight > 0 else 50.0

4 跨座舱启示：汽车 → 铁路 → 航空

4.1 VILMAS → FATED 技术延续

项目	领域	核心技术	差异
VILMAS	汽车	视线+语言模型+环境上下文	分心+场景理解
FATED	铁路	面部+体态+时序疲劳	疲劳进展+就绪度

4.2 跨领域共性

技术模块	汽车DMS	铁路FATED	航空
眼部检测	EAR+PERCLOS	EAR+PERCLOS+时序	同
头部姿态	Pitch/Yaw/Roll	同+点头检测	同
体态估计	可选	必需（体态垮塌）	必需
个体基线	可选	核心	核心
就绪度评分	二元预警	连续评分	连续评分
误报控制	中等	极高要求	极高要求

4.3 对 IMS 的启示

启示	内容	优先级
连续就绪度	从二元预警→连续评分	P0
个体基线	每个驾驶员建立正常范围	P0
多层架构	帧级→行为级→就绪度级	P1
体态融合	增加骨架关键点	P1
时序分析	时频分析提取疲劳模式	P2
误报控制	个体基线+多指标确认	P0

5 与 NeuroUX PVT 的互补

NeuroUX 平台使用 PVT（精神运动警戒测试）进行班前检测：

维度	FATED（CV监测）	NeuroUX（PVT测试）
时机	实时持续	班前/班后
方式	非侵入视频	主动测试（手机/平板）
指标	面部/体态行为	反应时间/注意力
基线	个体视觉基线	个体PVT基线
部署	驾驶室内摄像头	移动设备

互补方案： PVT 班前筛查 + CV 实时监测 = 完整疲劳管理。

6 测试场景

编号	场景	预期就绪度	干预
R-01	班前休息充足	90-100	无
R-02	连续驾驶4小时	70-85	监测
R-03	连续驾驶8小时	50-70	提醒
R-04	夜间03:00-05:00	40-60	警告
R-05	微睡眠事件	20-40	紧急
R-06	个体差异	基线校准	自适应
R-07	眼镜遮挡	降级模式	体态主导
R-08	振动环境	鲁棒	滤波

7 总结

FATED 的核心贡献是将疲劳监测从”单指标阈值触发”升级为”多层级连续就绪度评估”。三层架构（帧级信号→行为模式→就绪度评分）和个体基线校准是两项关键技术。

对 IMS 跨座舱部署，建议：

1. 采用连续就绪度评分替代二元预警
2. 建立个体基线校准机制
3. 融合面部+体态多模态信号
4. 参考铁路经验降低误报率

---

参考来源：

• Mosaic ATM FATED
• FRA SBIR Program
• NeuroUX PVT Platform