Dapalm | IMS技术前沿

Mosaic ATM FATED：计算机视觉驱动的铁路驾驶员疲劳检测框架深度解析

本文深度解读美国联邦铁路管理局（FRA）资助的 FATED（Fatigue Assessment for Transportation Engineer Determination）项目 Phase I 成果，分析 Mosaic ATM 如何用计算机视觉+人因工程构建操作员准备度连续评估框架，并探讨其对 IMS 座舱监控系统的跨领域启示。

1. 项目背景

1.1 问题定义

疲劳不是开关量。它渐进发展、因人而异、在绩效开始下降前难以自我察觉。对于安全关键运输操作（铁路、航空、卡车、海事），这意味着一个严峻的技术挑战：

• 传统方法缺陷： 依赖单一指标阈值（如闭眼时长、头部角度）容易产生误报
• 个体差异： 不同人表达疲劳的方式不同，通用阈值难以可靠适用
• 信任问题： 过多误报会削弱用户信心，导致监控技术被排斥

1.2 FATED 项目概况

要素	详情
全称	Fatigue Assessment for Transportation Engineer Determination
资助方	美国联邦铁路管理局（FRA）
执行方	Mosaic ATM
类型	SBIR Small Business Innovation Research Phase I
目标	用非侵入式视频监控检测铁路驾驶员疲劳与警觉性下降
状态	Phase I 完成，可行性已验证

1.3 技术渊源

FATED 建立在 Mosaic ATM 此前为汽车领域开发的 VILMAS（Visual Integration of Language Models in Automotive Safety）项目之上：

• VILMAS 探索了 gaze estimation、眼态监控、计算机视觉、深度学习、视觉语言模型
• FATED 将技术从驾驶员分心/情境感知转向 疲劳进展与持续警觉性
• 这种技术延续性使 Phase I 能集中资源解决核心研究问题

2. 核心创新：操作员准备度连续模型

2.1 从二元检测到连续评估

传统疲劳检测系统的基本范式是”疲劳/未疲劳”的二元判断。FATED 的核心创新在于：

将操作员准备度（operator readiness）建模为连续光谱，而非二元状态。

这一思路的核心理念是：

graph LR
    A[清醒 Alert] --> B[轻度疲劳 Mild Fatigue]
    B --> C[困倦 Drowsy]
    C --> D[嗜睡 Sleepy]
    D --> E[睡眠 Sleep]
    
    style A fill:#4CAF50,color:#fff
    style B fill:#FFEB3B,color:#000
    style C fill:#FF9800,color:#fff
    style D fill:#f44336,color:#fff
    style E fill:#9E9E9E,color:#fff

2.2 三级信号架构

FATED 设计了三层分析架构，将原始视觉信号逐级抽象为可解释的准备度评估：

flowchart TB
    subgraph Low["低层视觉信号 (Low-Level)"]
        L1[眼部状态]
        L2[注视方向]
        L3[面部关键点]
        L4[头部姿态]
        L5[身体姿势]
    end
    
    subgraph Mid["中层行为模式 (Mid-Level)"]
        M1[眨眼频率/时长]
        M2[持续闭眼]
        M3[打哈欠]
        M4[点头]
        M5[注视变化]
        M6[姿势变化]
    end
    
    subgraph High["高层准备度评估 (High-Level)"]
        H1[时间序列融合]
        H2[个体基线校准]
        H3[准备度连续评分]
    end
    
    L1 --> M1
    L1 --> M2
    L3 --> M3
    L4 --> M4
    L2 --> M5
    L5 --> M6
    
    M1 --> H1
    M2 --> H1
    M3 --> H1
    M4 --> H1
    M5 --> H1
    M6 --> H1
    
    H1 --> H2
    H2 --> H3

低层信号（帧级）

信号类型	提取方法	数据来源
眼部状态	眼睛关键点检测	面部 landmark 模型
注视方向	Gaze estimation 模型	瞳孔+眼角
面部关键点	68/468 点面部 landmark	MediaPipe / FAN
头部姿态	PnP 求解	3D face model
身体姿势	骨骼姿态估计	OpenPose / HRNet

中层行为（时序模式）

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

class FatigueBehaviorAnalyzer:
    """
    FATED 中层行为分析器
    
    将低层帧级信号聚合为时序行为模式，
    实现 FATED Phase I 描述的中层分析能力。
    
    参考：Mosaic ATM FATED Phase I 框架
    """
    
    def __init__(self, window_sec: int = 60, fps: int = 30):
        """
        Args:
            window_sec: 滑动窗口秒数（FATED 建议 60s 捕捉疲劳模式）
            fps: 视频帧率
        """
        self.window_size = window_sec * fps
        self.fps = fps
        
        # 低层信号缓冲区
        self.eye_openness = deque(maxlen=self.window_size)
        self.gaze_directions = deque(maxlen=self.window_size)
        self.head_poses = deque(maxlen=self.window_size)
        self.mouth_states = deque(maxlen=self.window_size)
        self.body_poses = deque(maxlen=self.window_size)
        
        # 中层行为统计
        self.behaviors = {
            'blink_count': 0,
            'blink_durations': [],
            'prolonged_closures': [],  # 持续闭眼事件
            'yawn_count': 0,
            'yawn_durations': [],
            'head_nods': [],
            'gaze_shifts': 0,
            'posture_changes': 0,
        }
    
    def update(self, frame_signals: Dict) -> None:
        """
        更新一帧的低层信号
        
        Args:
            frame_signals: 包含以下键的字典:
                - 'eye_openness': float (0-1), 眼睑开度
                - 'gaze': Tuple[float, float], 注视方向 (pitch, yaw)
                - 'head_pose': Tuple[float, float, float], (roll, pitch, yaw)
                - 'mouth_open': float (0-1), 嘴部开度
                - 'body_keypoints': np.ndarray, 骨骼关键点
        """
        self.eye_openness.append(frame_signals['eye_openness'])
        self.gaze_directions.append(frame_signals['gaze'])
        self.head_poses.append(frame_signals['head_pose'])
        self.mouth_states.append(frame_signals['mouth_open'])
        self.body_poses.append(frame_signals['body_keypoints'])
        
        # 实时行为检测
        self._detect_blink()
        self._detect_yawn()
        self._detect_head_nod()
        self._detect_gaze_shift()
        self._detect_posture_change()
    
    def _detect_blink(self) -> None:
        """检测眨眼事件：开度从 >0.3 降到 <0.2 再回到 >0.3"""
        if len(self.eye_openness) < 3:
            return
        
        prev, curr, _ = (list(self.eye_openness)[-3], 
                         list(self.eye_openness)[-2], 
                         list(self.eye_openness)[-1])
        
        if prev > 0.3 and curr < 0.2:
            self.behaviors['blink_count'] += 1
            # 记录闭眼开始时间
            self._blink_start = len(self.eye_openness) - 2
            
        if hasattr(self, '_blink_start') and curr > 0.3:
            duration = (len(self.eye_openness) - self._blink_start) / self.fps
            self.behaviors['blink_durations'].append(duration)
            
            # 持续闭眼 >1.5s 记录为 prolonged closure
            if duration > 1.5:
                self.behaviors['prolonged_closures'].append({
                    'duration_sec': duration,
                    'timestamp': len(self.eye_openness) / self.fps,
                })
            del self._blink_start
    
    def _detect_yawn(self) -> None:
        """检测打哈欠：嘴部开度 >0.6 持续 >1s"""
        if len(self.mouth_states) < self.fps:
            return
        
        recent = np.array(list(self.mouth_states)[-self.fps:])
        if np.max(recent) > 0.6 and np.mean(recent > 0.5) > 0.5:
            if not hasattr(self, '_yawn_start'):
                self._yawn_start = len(self.mouth_states) - self.fps
        elif hasattr(self, '_yawn_start'):
            duration = (len(self.mouth_states) - self._yawn_start) / self.fps
            if duration > 1.0:
                self.behaviors['yawn_count'] += 1
                self.behaviors['yawn_durations'].append(duration)
            del self._yawn_start
    
    def _detect_head_nod(self) -> None:
        """检测点头：头部 pitch 在 2s 内下降再回升"""
        if len(self.head_poses) < self.fps * 2:
            return
        
        recent_pitch = np.array([p[1] for p in list(self.head_poses)[-self.fps*2:]])
        # 简化：检测 pitch 先降后升的模式
        diff = np.diff(recent_pitch)
        neg_count = np.sum(diff < -0.02)
        pos_count = np.sum(diff > 0.02)
        
        if neg_count > 5 and pos_count > 5:
            self.behaviors['head_nods'].append({
                'timestamp': len(self.head_poses) / self.fps,
                'pitch_range': float(np.max(recent_pitch) - np.min(recent_pitch)),
            })
    
    def _detect_gaze_shift(self) -> None:
        """检测注视转移：注视方向突变"""
        if len(self.gaze_directions) < 2:
            return
        
        prev_gaze = np.array(list(self.gaze_directions)[-2])
        curr_gaze = np.array(list(self.gaze_directions)[-1])
        angular_diff = np.arccos(np.clip(
            np.dot(prev_gaze, curr_gaze) / 
            (np.linalg.norm(prev_gaze) * np.linalg.norm(curr_gaze) + 1e-8),
            -1, 1
        ))
        
        if angular_diff > 0.3:  # >17° 突变
            self.behaviors['gaze_shifts'] += 1
    
    def _detect_posture_change(self) -> None:
        """检测姿势变化：身体关键点整体位移"""
        if len(self.body_poses) < 2:
            return
        
        prev_body = self.body_poses[-2]
        curr_body = self.body_poses[-1]
        displacement = np.mean(np.linalg.norm(
            curr_body - prev_body, axis=-1
        ))
        
        if displacement > 5.0:  # 像素阈值
            self.behaviors['posture_changes'] += 1
    
    def get_behavior_summary(self) -> Dict:
        """获取当前窗口的行为摘要"""
        return {
            'blink_rate': len(self.behaviors['blink_durations']) / 
                          (self.window_size / self.fps) * 60,  # 次/分钟
            'mean_blink_duration': float(np.mean(
                self.behaviors['blink_durations'])) if self.behaviors['blink_durations'] else 0,
            'prolonged_closure_count': len(self.behaviors['prolonged_closures']),
            'yawn_rate': self.behaviors['yawn_count'] / 
                        (self.window_size / self.fps) * 60,
            'head_nod_count': len(self.behaviors['head_nods']),
            'gaze_shift_rate': self.behaviors['gaze_shifts'] / 
                             (self.window_size / self.fps) * 60,
            'posture_change_count': self.behaviors['posture_changes'],
        }


# 测试代码
if __name__ == "__main__":
    import time
    np.random.seed(42)
    
    analyzer = FatigueBehaviorAnalyzer(window_sec=60, fps=30)
    
    # 模拟 60 秒视频（1800 帧）
    print("=== 模拟正常驾驶状态 ===")
    for i in range(900):  # 前 30 秒：正常
        signals = {
            'eye_openness': np.clip(np.random.normal(0.8, 0.05), 0, 1),
            'gaze': (np.random.normal(0, 0.05), np.random.normal(0, 0.05)),
            'head_pose': (0, np.random.normal(0, 0.01), 0),
            'mouth_open': np.clip(np.random.normal(0.1, 0.02), 0, 1),
            'body_keypoints': np.random.randn(17, 2) * 0.1,
        }
        analyzer.update(signals)
    
    summary_normal = analyzer.get_behavior_summary()
    print(f"正常状态行为摘要: {summary_normal}")
    
    # 模拟疲劳状态
    print("\n=== 模拟疲劳驾驶状态 ===")
    for i in range(900):  # 后 30 秒：疲劳
        # 偶尔闭眼
        eye = 0.1 if i % 90 == 0 else np.clip(np.random.normal(0.5, 0.15), 0, 1)
        signals = {
            'eye_openness': eye,
            'gaze': (np.random.normal(0.2, 0.15), np.random.normal(0.1, 0.1)),
            'head_pose': (0, 0.15 + np.sin(i * 0.1) * 0.05, 0),  # 头前倾+点头
            'mouth_open': 0.7 if 300 <= i <= 330 else 0.1,  # 打哈欠
            'body_keypoints': np.random.randn(17, 2) * 0.3,
        }
        analyzer.update(signals)
    
    summary_fatigue = analyzer.get_behavior_summary()
    print(f"疲劳状态行为摘要: {summary_fatigue}")
    
    print(f"\n眨眼率变化: {summary_normal['blink_rate']:.1f} → {summary_fatigue['blink_rate']:.1f} 次/分")
    print(f"持续闭眼事件: {summary_fatigue['prolonged_closure_count']} 次")
    print(f"打哈欠次数: {summary_fatigue['yawn_rate']:.1f} 次/分")
    print(f"点头次数: {summary_fatigue['head_nod_count']} 次")

高层评估（准备度连续评分）

class OperatorReadinessAssessor:
    """
    FATED 高层操作员准备度评估器
    
    将中层行为模式融合为连续的准备度评分 (0-100)
    采用加权融合 + 个体基线校准
    """
    
    def __init__(self):
        # 行为权重（基于 FATED 人因研究）
        self.weights = {
            'blink_rate': 0.15,
            'prolonged_closure': 0.25,  # 持续闭眼权重最高
            'yawn_rate': 0.15,
            'head_nod': 0.20,
            'gaze_shift': 0.10,
            'posture_change': 0.15,
        }
        
        # 个体基线
        self.baseline = None
        self.calibration_samples = []
        self.is_calibrated = False
    
    def calibrate(self, behavior_summaries: list) -> None:
        """
        用个体正常状态数据校准基线
        
        Args:
            behavior_summaries: 正常状态下的行为摘要列表
        """
        if len(behavior_summaries) < 10:
            print(f"警告：校准样本不足 ({len(behavior_summaries)}/10)")
            return
        
        self.baseline = {}
        for key in behavior_summaries[0]:
            values = [s[key] for s in behavior_summaries]
            self.baseline[key] = {
                'mean': float(np.mean(values)),
                'std': float(np.std(values)) + 1e-6,
            }
        self.is_calibrated = True
        print(f"基线校准完成（{len(behavior_summaries)} 样本）")
    
    def assess(self, behavior_summary: Dict) -> Dict:
        """
        评估准备度
        
        Returns:
            {
                'readiness_score': float,  # 0-100, 越高越警觉
                'fatigue_level': str,     # alert/mild/drowsy/sleepy
                'contributing_factors': Dict,  # 各行为偏离基线的程度
            }
        """
        if not self.is_calibrated:
            # 无基线时用固定阈值
            return self._assess_fixed(behavior_summary)
        
        # 计算各行为相对基线的偏离
        deviations = {}
        for key, weight in self.weights.items():
            if key in self.baseline:
                baseline_mean = self.baseline[key]['mean']
                baseline_std = self.baseline[key]['std']
                actual = behavior_summary.get(key, 0)
                # z-score
                z = (actual - baseline_mean) / baseline_std
                deviations[key] = {
                    'z_score': float(z),
                    'weight': weight,
                    'actual': actual,
                    'baseline': baseline_mean,
                }
        
        # 加权融合
        total_deviation = sum(
            max(0, d['z_score']) * d['weight'] 
            for d in deviations.values()
        )
        
        # 映射到 0-100 准备度评分
        readiness_score = max(0, 100 - total_deviation * 10)
        
        # 疲劳等级
        if readiness_score > 80:
            fatigue_level = 'alert'
        elif readiness_score > 60:
            fatigue_level = 'mild_fatigue'
        elif readiness_score > 40:
            fatigue_level = 'drowsy'
        else:
            fatigue_level = 'sleepy'
        
        return {
            'readiness_score': float(readiness_score),
            'fatigue_level': fatigue_level,
            'contributing_factors': deviations,
        }
    
    def _assess_fixed(self, behavior_summary: Dict) -> Dict:
        """无基线时的固定阈值评估"""
        # PERCLOS-like: 持续闭眼事件
        closure_score = min(100, behavior_summary.get('prolonged_closure_count', 0) * 20)
        # 眨眼率偏离
        blink_rate = behavior_summary.get('blink_rate', 0)
        blink_score = min(100, max(0, (blink_rate - 15) * 5))
        # 打哈欠
        yawn_score = min(100, behavior_summary.get('yawn_rate', 0) * 30)
        # 点头
        nod_score = min(100, behavior_summary.get('head_nod_count', 0) * 15)
        
        total_fatigue = (closure_score * 0.3 + blink_score * 0.2 + 
                         yawn_score * 0.2 + nod_score * 0.3)
        readiness = 100 - total_fatigue
        
        return {
            'readiness_score': float(readiness),
            'fatigue_level': 'alert' if readiness > 80 else 
                          'mild_fatigue' if readiness > 60 else
                          'drowsy' if readiness > 40 else 'sleepy',
            'contributing_factors': {
                'note': '使用固定阈值（未校准个体基线）',
            },
        }


# 测试
if __name__ == "__main__":
    assessor = OperatorReadinessAssessor()
    
    # 模拟校准
    np.random.seed(42)
    calib_data = []
    for _ in range(15):
        calib_data.append({
            'blink_rate': np.random.normal(12, 3),
            'prolonged_closure_count': np.random.poisson(0.5),
            'yawn_rate': np.random.normal(0.5, 0.2),
            'head_nod_count': np.random.poisson(1),
            'gaze_shift_rate': np.random.normal(8, 2),
            'posture_change_count': np.random.poisson(2),
        })
    assessor.calibrate(calib_data)
    
    # 评估疲劳状态
    fatigue_behavior = {
        'blink_rate': 22,
        'prolonged_closure_count': 3,
        'yawn_rate': 2.5,
        'head_nod_count': 5,
        'gaze_shift_rate': 15,
        'posture_change_count': 8,
    }
    
    result = assessor.assess(fatigue_behavior)
    print(f"\n准备度评分: {result['readiness_score']:.1f}/100")
    print(f"疲劳等级: {result['fatigue_level']}")
    print(f"贡献因子:")
    for k, v in result['contributing_factors'].items():
        if isinstance(v, dict) and 'z_score' in v:
            print(f"  {k}: z={v['z_score']:.2f} (基线={v['baseline']:.2f}, 实际={v['actual']:.2f})")

3. Phase I 实验验证

3.1 数据集

参数	数值
视频总时长	>30 小时
参与者数	60 人
条件	自报告清醒 vs 疲劳
数据类型	公开可用视频数据集

3.2 关键发现

1. 可观测差异： 在打哈欠频率、眨眼特征、闭眼事件、面部活动中，清醒与疲劳录制之间存在可测量差异
2. 个体变异性： 不同个体表达疲劳的方式有显著差异，验证了个体基线校准的必要性
3. 时序结构： 时间序列和时频分析揭示了简单阈值规则难以捕捉的行为模式

3.3 计算约束

Phase I 测试发现了关键的计算瓶颈：

问题	原因	解决方向
多模型并行在 CPU-only 硬件上性能不足	面部分析+姿态估计+注视模型同时运行	硬件加速（GPU/NPU）、模型调度、边缘计算优化

4. 模块化架构设计

flowchart LR
    subgraph Input
        CAM[摄像头输入]
    end
    
    subgraph Pipeline
        DET[检测与跟踪]
        TRK[目标跟踪]
        FEAT[特征提取]
        AGG[时序聚合]
        ANAL[高层分析]
    end
    
    subgraph Output
        ALERT[告警/可视化]
        API[API 接口]
        DASH[仪表盘]
    end
    
    CAM --> DET
    DET --> TRK
    TRK --> FEAT
    FEAT --> AGG
    AGG --> ANAL
    ANAL --> ALERT
    ANAL --> API
    ANAL --> DASH
    
    style DET fill:#2196F3,color:#fff
    style AGG fill:#FF9800,color:#fff
    style ANAL fill:#f44336,color:#fff

每个模块设计为独立服务，可单独替换算法而不需要重新设计整个系统：

模块	功能	可替换组件
检测与跟踪	人脸/人体检测+跟踪	YOLO/MTCNN/MediaPipe
特征提取	面部关键点+姿态	FAN/MediaPipe FaceMesh
时序聚合	滑动窗口行为统计	自定义分析器
高层分析	准备度评分	加权融合/时序模型

5. 跨领域启示：对 IMS 的指导价值

5.1 从二元到连续：IMS 疲劳检测的范式转移

当前 IMS 疲劳检测主要基于 PERCLOS 阈值（如 ≥30% 触发警告）。FATED 框架提供了更丰富的思路：

当前 IMS 方案	FATED 启示	改进方向
PERCLOS 单一阈值	多指标融合	眨眼+点头+哈欠+姿势联合评估
固定阈值	个体基线校准	首次 30 分钟建立基线
二元告警	连续准备度评分	0-100 连续评分+多级干预
帧级判断	时序模式分析	60s 窗口行为模式分析

5.2 铁路→汽车的技术迁移路径

技术能力	铁路场景	汽车场景适配
非侵入式视频	驾驶室固定摄像头	DMS 摄像头已有
长时间持续监控	数小时运行	驾驶全程监控
个体校准	排班固定驾驶员	驾驶员身份识别+个性化模型
多信号融合	面部+姿势	可增加方向盘传感器
边缘部署	驾驶室计算	高通/T