Dapalm | IMS技术前沿

FATED：美国联邦铁路管理局（FRA）计算机视觉疲劳连续评估框架——铁路驾驶员监测的新范式

信息来源

项目	内容
项目名称	Fatigue Assessment for Transportation Engineer Determination (FATED)
资助机构	美国联邦铁路管理局（FRA）
执行方	Mosaic ATM
类型	Small Business Innovation Research (SBIR) Phase I
时间	2025-2026
链接	https://mosaicatm.com/2026/09/01/advancing-rail-safety-with-computer-vision/
数据	30小时视频，60名被试
技术基础	VILMAS项目（视觉+语言模型汽车安全）

核心创新

1. 疲劳连续体模型：非二元”疲劳/清醒”判定，而是将疲劳视为渐进连续过程
2. 三层分析架构：低层视觉信号→中层行为模式→高层准备度评估
3. 个体基线校准：不使用通用阈值，而是基于个体行为基线判定变化
4. 面部+骨骼双通道：面部特征点+身体姿态估计联合分析

问题定义

铁路驾驶员疲劳监测挑战

挑战	描述	FATED解决方案
个体差异	不同人表达疲劳的方式不同	个体基线校准
误报过多	单一指标阈值导致大量误报	多指标时序融合
环境干扰	驾驶室光照/振动影响检测	非侵入视频+鲁棒特征
信任问题	操作员不信任过多警报	连续体+可解释性

与汽车DMS的关键差异

维度	汽车DMS	铁路FATED
驾驶时长	1-4小时	6-12小时
环境变化	大（天气/隧道）	相对稳定（驾驶室）
操作类型	手动驾驶	监控+手动干预
法规	Euro NCAP	FRA法规
报警后果	警告驾驶员	调度中心通知
振动	中等	高（机车振动）

方法详解

三层分析架构

flowchart TD
    subgraph Layer1[低层视觉信号]
        A1[眼部状态]
        A2[凝视方向]
        A3[面部特征点]
        A4[头部姿态]
        A5[骨骼姿态]
    end
    
    subgraph Layer2[中层行为模式]
        B1[眨眼频率/时长]
        B2[持续闭眼]
        B3[打哈欠]
        B4[点头]
        B5[凝视变化]
        B6[姿态变化]
    end
    
    subgraph Layer3[高层准备度评估]
        C1[疲劳连续体]
        C2[操作准备度]
        C3[渐进趋势]
    end
    
    Layer1 --> Layer2
    Layer2 --> Layer3

核心实现

import numpy as np
from dataclasses import dataclass, field
from typing import List

@dataclass
class LowLevelSignals:
    """低层视觉信号（逐帧提取）"""
    eye_openness: float         # 眼睑开度 (0-1)
    gaze_x: float               # 凝视X (度)
    gaze_y: float               # 凝视Y (度)
    head_pitch: float            # 头部俯仰 (度)
    head_roll: float             # 头部滚转 (度)
    head_yaw: float              # 头部偏航 (度)
    mouth_open: float            # 嘴部开度 (0-1)
    body_posture: np.ndarray    # 骨骼关键点 [17, 2]

@dataclass
class MidLevelBehaviors:
    """中层行为模式（时序聚合）"""
    blink_rate: float            # 眨眼频率 (次/min)
    avg_blink_duration: float    # 平均眨眼时长 (s)
    perclos: float               # PERCLOS (%)
    yawn_count: int              # 哈欠次数
    nod_count: int               # 点头次数
    gaze_variance: float        # 凝视方差
    posture_shifts: int         # 姿态变化次数
    head_sway_freq: float        # 头部摇摆频率 (Hz)

class FATEDFramework:
    """
    FATED疲劳连续评估框架
    
    核心理念：疲劳不是on/off，而是渐进连续体
    评估从正常→疲劳→嗜睡→睡眠的渐进过程
    """
    
    def __init__(self, window_sec: int = 60):
        self.window_sec = window_sec
        self.individual_baseline = None
        
        # 疲劳连续体阶段
        self.stages = [
            'Alert',        # 清醒：正常操作
            'Fatigued',     # 疲劳：行为开始变化
            'Drowsy',       # 嗜睡：明显困倦
            'Sleepy',       # 困倦：接近睡眠
            'Sleep',        # 睡眠：需立即干预
        ]
    
    def calibrate_baseline(self, 
                           signals: List[LowLevelSignals],
                           calibration_min: int = 5) -> dict:
        """
        个体基线校准
        
        Args:
            signals: 清醒状态信号序列
            calibration_min: 最少校准分钟数
            
        Returns:
            baseline: 个体基线指标
        """
        if len(signals) < calibration_min * 30 * 60:
            return None
        
        baseline = {
            'eye_openness': np.mean([s.eye_openness for s in signals]),
            'blink_rate': self._calc_blink_rate(signals),
            'perclos': self._calc_perclos(signals),
            'gaze_variance': np.var([s.gaze_x**2 + s.gaze_y**2 
                                      for s in signals]),
            'head_sway': np.std([s.head_pitch for s in signals]),
            'posture_stability': np.std([s.body_posture[0, 0] 
                                          for s in signals]),
        }
        
        self.individual_baseline = baseline
        return baseline
    
    def assess_continuum(self, 
                          signals: List[LowLevelSignals]) -> dict:
        """
        评估疲劳连续体阶段
        
        Returns:
            {'stage': str, 'score': 0-1, 'trend': str}
        """
        if self.individual_baseline is None:
            return {'stage': 'Unknown', 'score': 0, 'trend': 'stable'}
        
        behaviors = self._extract_behaviors(signals)
        
        # 计算各维度偏离度
        deviations = {
            'eye': self._deviation(
                behaviors.perclos, 
                self.individual_baseline['perclos'],
                max_val=40
            ),
            'blink': self._deviation(
                behaviors.blink_rate,
                self.individual_baseline['blink_rate'],
                max_val=30
            ),
            'gaze': self._deviation(
                behaviors.gaze_variance,
                self.individual_baseline['gaze_variance'],
                max_val=100
            ),
            'head': self._deviation(
                behaviors.head_sway_freq,
                self.individual_baseline['head_sway'],
                max_val=0.5
            ),
            'yawn': min(behaviors.yawn_count / 5, 1.0),
            'nod': min(behaviors.nod_count / 3, 1.0),
            'posture': min(behaviors.posture_shifts / 10, 1.0),
        }
        
        # 加权综合
        weights = {
            'eye': 0.25, 'blink': 0.15, 'gaze': 0.15,
            'head': 0.15, 'yawn': 0.10, 'nod': 0.10,
            'posture': 0.10,
        }
        
        score = sum(weights[k] * deviations[k] for k in weights)
        
        # 映射到连续体阶段
        if score < 0.15:
            stage = 'Alert'
        elif score < 0.35:
            stage = 'Fatigued'
        elif score < 0.55:
            stage = 'Drowsy'
        elif score < 0.75:
            stage = 'Sleepy'
        else:
            stage = 'Sleep'
        
        return {
            'stage': stage,
            'score': score,
            'deviations': deviations,
            'trend': self._calc_trend(deviations),
        }
    
    def _extract_behaviors(self, 
                           signals: List[LowLevelSignals]) -> MidLevelBehaviors:
        """从低层信号提取中层行为"""
        eye_values = [s.eye_openness for s in signals]
        
        # PERCLOS
        threshold = np.mean(eye_values) * 0.7
        closed_count = sum(1 for e in eye_values if e < threshold)
        perclos = closed_count / len(eye_values) * 100
        
        # 眨眼（简化）
        crossings = sum(1 for i in range(1, len(eye_values))
                        if eye_values[i-1] > threshold and 
                        eye_values[i] <= threshold)
        blink_rate = crossings / (len(signals) / 30 / 60)  # 30fps
        
        # 哈欠（嘴部大开）
        yawns = sum(1 for s in signals if s.mouth_open > 0.6)
        
        return MidLevelBehaviors(
            blink_rate=blink_rate,
            avg_blink_duration=0.15,
            perclos=perclos,
            yawn_count=yawns,
            nod_count=sum(1 for s in signals if abs(s.head_pitch) > 20),
            gaze_variance=np.var([s.gaze_x**2 + s.gaze_y**2 
                                  for s in signals]),
            posture_shifts=0,
            head_sway_freq=np.std([s.head_pitch for s in signals]),
        )
    
    def _deviation(self, current, baseline, max_val):
        """计算偏离度 (0-1)"""
        diff = abs(current - baseline)
        return min(diff / max_val, 1.0)
    
    def _calc_trend(self, deviations):
        """计算趋势方向"""
        avg_dev = np.mean(list(deviations.values()))
        if avg_dev < 0.2:
            return 'stable'
        elif avg_dev < 0.4:
            return 'rising'
        else:
            return 'critical'


# 测试
if __name__ == "__main__":
    fated = FATEDFramework(window_sec=60)
    
    # 模拟清醒信号（5分钟，30fps）
    np.random.seed(42)
    alert_signals = [
        LowLevelSignals(
            eye_openness=0.85 + np.random.randn() * 0.05,
            gaze_x=np.random.randn() * 5,
            gaze_y=np.random.randn() * 3,
            head_pitch=np.random.randn() * 2,
            head_roll=np.random.randn() * 1,
            head_yaw=np.random.randn() * 3,
            mouth_open=np.random.rand() * 0.1,
            body_posture=np.random.randn(17, 2),
        )
        for _ in range(5 * 30 * 60)  # 5min
    ]
    
    # 校准基线
    baseline = fated.calibrate_baseline(alert_signals)
    print(f"基线 PERCLOS: {baseline['perclos']:.1f}%")
    print(f"基线 眨眼率: {baseline['blink_rate']:.1f}/min")
    
    # 模拟疲劳信号
    fatigue_signals = [
        LowLevelSignals(
            eye_openness=max(0, 0.6 + np.random.randn() * 0.1),
            gaze_x=np.random.randn() * 12,
            gaze_y=np.random.randn() * 8,
            head_pitch=np.random.randn() * 5 + (3 if i % 100 > 80 else 0),
            head_roll=np.random.randn() * 2,
            head_yaw=np.random.randn() * 6,
            mouth_open=0.5 if i % 200 > 190 else np.random.rand() * 0.1,
            body_posture=np.random.randn(17, 2),
        )
        for i in range(60 * 30)  # 1min
    ]
    
    result = fated.assess_continuum(fatigue_signals)
    print(f"\n疲劳评估:")
    print(f"  阶段: {result['stage']}")
    print(f"  分数: {result['score']:.2f}")
    print(f"  趋势: {result['trend']}")
    print(f"  各维度偏离: {result['deviations']}")

实验结果

Phase I 数据集

指标	数值
视频总时长	30+ 小时
参与者	60 名
条件	自评清醒 vs 疲劳
特征	面部特征点 + 骨骼姿态
分析方法	时序 + 时频

关键发现

1. 个体差异显著：不同被试表达疲劳的方式差异大，通用阈值不可靠
2. 时序模式重要：单帧特征不够，需时序分析捕获渐进趋势
3. 多信号互补：面部+骨骼联合优于单一通道
4. ** nuisance alert问题**：简单阈值方法误报率高，操作员失去信任

IMS开发启示

1. FATED框架对IMS的直接价值

FATED特性	IMS应用	实现方式
连续体模型	替代二元疲劳判定	渐进5阶段评分
个体基线	个性化疲劳检测	驾驶前5min校准
多层分析	丰富特征体系	低层→中层→高层管道
面部+骨骼	姿态+OOP检测复用	骨骼关键点双用途

2. 铁路→汽车的技术迁移

铁路特征	汽车适配	优势
长时程监测（6-12h）	长途驾驶疲劳	时序趋势分析
驾驶室稳定环境	车内光照变化	需增强光照鲁棒性
操作员信任优先	减少误报=用户体验	多信号降低误报
调度中心通知	云端车队管理	5G+边缘协同

3. 与现有IMS管道的集成

组件	现有IMS	FATED增强
疲劳检测	PERCLOS阈值	→连续体5阶段
分心检测	凝视偏离	→时序凝视方差
OOP检测	骨骼关键点	→姿态稳定性指标
个体化	无	→5min基线校准
报警策略	固定阈值	→渐进式干预

总结

FATED框架是铁路驾驶员疲劳检测的重要进展：

1. 连续体替代二元判定：Alert→Fatigued→Drowsy→Sleepy→Sleep渐进评估
2. 个体基线校准：解决”一人一态”的疲劳表达差异问题
3. 三层架构：低层视觉→中层行为→高层准备度，可解释性强
4. 30小时60人验证：Phase I证实多信号时序融合优于单一阈值
5. 跨域迁移：铁路→汽车DMS的技术迁移路径清晰