可解释驾驶员行为状态监测：MediaPipe面部关键点+SVM实现91%五状态分类

论文信息

项目	内容
论文标题	Interpretable Driver Behavioral State Monitoring using Facial Landmarks for Human-Autonomous Vehicle Interaction
作者	Candela, Mandujano-Carrillo, Vidal-Lopez, Ramírez-Moreno, Martínez, Lozoya-Santos
期刊	Frontiers in Robotics and AI, Section: Human-Robot Interaction
发表时间	2026-09-15 (Accepted)
DOI	10.3389/frobt.2026.1896859
测试车辆	Polaris GEM e6 自动驾驶车

核心创新

提出一个可解释、可校准的驾驶员行为状态监测框架，使用MediaPipe面部关键点 + 经典ML分类器，在受控数据集上实现91% F1分数。

特性	说明
可解释性	基于面部关键点几何特征 (EAR, MAR, 头部旋转)
个性化校准	每个驾驶员独立建立中性基线
五种状态	正常/闭眼/偏离区域/打哈欠/频繁眨眼
实时HMI	ROS + Foxglove Studio 可视化

方法详解

1. 系统架构

"""
可解释驾驶员行为状态监测系统
论文: Frontiers in Robotics and AI, 2026

硬件: Intel RealSense L515 + Polaris GEM e6
软件: ROS + Foxglove + MediaPipe + OpenCV

依赖: mediapipe, opencv, scikit-learn, numpy
"""

import numpy as np
import cv2
import mediapipe as mp
from sklearn.svm import SVC
from sklearn.ensemble import RandomForestClassifier
from dataclasses import dataclass
from typing import Tuple, Dict

@dataclass
class DriverStateConfig:
    """驾驶员状态监测配置"""
    # 五种行为状态
    states = ['neutral', 'eye_closure', 'out_of_safe_zone', 'yawning', 'excessive_blinking']
    
    # MediaPipe面部网格关键点索引
    # 左眼 landmarks (EAR计算)
    left_eye_indices = [33, 160, 158, 133, 153, 144]  # P1-P6
    # 右眼 landmarks
    right_eye_indices = [362, 385, 387, 263, 373, 380]
    # 嘴巴 landmarks (MAR计算)
    mouth_indices = [61, 291, 13, 14, 78, 308]  # 外唇+内唇
    
    # EAR 阈值 (闭眼检测)
    ear_threshold_closed: float = 0.21    # < 此值视为闭眼
    ear_threshold_blink: float = 0.25     # 眨眼检测下限
    
    # MAR 阈值 (打哈欠检测)
    mar_threshold_yawn: float = 0.6       # > 此值视为打哈欠
    
    # 头部旋转阈值 (偏离安全区域)
    pitch_threshold: float = 15.0         # 俯仰角 (度)
    yaw_threshold: float = 25.0           # 偏航角 (度)
    roll_threshold: float = 20.0          # 翻滚角 (度)
    
    # 眨眼频率 (频繁眨眼)
    blink_rate_threshold: float = 20.0    # 次/分钟
    blink_duration_range: Tuple[float, float] = (0.1, 0.4)  # 正常眨眼时长(s)


class FaceLandmarkExtractor:
    """MediaPipe面部关键点提取器"""
    
    def __init__(self):
        self.mp_face = mp.solutions.face_mesh
        self.face_mesh = self.mp_face.FaceMesh(
            static_image_mode=False,
            max_num_faces=1,
            refine_landmarks=True,
            min_detection_confidence=0.5,
            min_tracking_confidence=0.5,
        )
        self.config = DriverStateConfig()
        
    def process(self, frame: np.ndarray) -> dict:
        """
        处理一帧图像，提取面部特征
        
        Args:
            frame: (H, W, 3) BGR图像
            
        Returns:
            features: dict, 包含 EAR, MAR, 头部角度, 关键点
        """
        h, w = frame.shape[:2]
        rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
        results = self.face_mesh.process(rgb)
        
        if results.multi_face_landmarks is None:
            return None
        
        landmarks = results.multi_face_landmarks[0].landmark
        
        # 1. 计算EAR (Eye Aspect Ratio)
        left_ear = self._compute_ear(landmarks, self.config.left_eye_indices, w, h)
        right_ear = self._compute_ear(landmarks, self.config.right_eye_indices, w, h)
        ear = (left_ear + right_ear) / 2
        
        # 2. 计算MAR (Mouth Aspect Ratio)
        mar = self._compute_mar(landmarks, self.config.mouth_indices, w, h)
        
        # 3. 估计头部姿态
        pitch, yaw, roll = self._estimate_head_pose(landmarks, w, h)
        
        return {
            'ear': ear,
            'left_ear': left_ear,
            'right_ear': right_ear,
            'mar': mar,
            'pitch': pitch,
            'yaw': yaw,
            'roll': roll,
            'landmarks': landmarks,
        }
    
    def _compute_ear(self, landmarks, indices, w, h):
        """
        计算Eye Aspect Ratio (EAR)
        
        EAR = (|P2-P6| + |P3-P5|) / (2 × |P1-P4|)
        
        闭眼时EAR趋近0，睁眼时EAR约0.25-0.35
        """
        pts = [(landmarks[i].x * w, landmarks[i].y * h) for i in indices]
        
        # 垂直距离
        v1 = np.linalg.norm(np.array(pts[1]) - np.array(pts[5]))
        v2 = np.linalg.norm(np.array(pts[2]) - np.array(pts[4]))
        
        # 水平距离
        h_dist = np.linalg.norm(np.array(pts[0]) - np.array(pts[3]))
        
        if h_dist < 1e-6:
            return 0
        
        return (v1 + v2) / (2 * h_dist)
    
    def _compute_mar(self, landmarks, indices, w, h):
        """
        计算Mouth Aspect Ratio (MAR)
        
        MAR = (|上唇-下唇|) / (|左嘴角-右嘴角|)
        
        打哈欠时MAR显著增大
        """
        pts = [(landmarks[i].x * w, landmarks[i].y * h) for i in indices]
        
        # 垂直: 上唇(13)到下唇(14)
        v = np.linalg.norm(np.array(pts[2]) - np.array(pts[3]))
        
        # 水平: 左嘴角(0)到右嘴角(1)
        h_dist = np.linalg.norm(np.array(pts[0]) - np.array(pts[1]))
        
        if h_dist < 1e-6:
            return 0
        
        return v / h_dist
    
    def _estimate_head_pose(self, landmarks, w, h):
        """
        从面部关键点估计头部姿态 (俯仰/偏航/翻滚)
        
        简化版: 使用关键点几何关系
        论文使用MediaPipe的transform matrix
        """
        # 鼻尖 (1), 左眼 (33), 右眼 (263), 左耳 (234), 右耳 (454)
        nose = np.array([landmarks[1].x, landmarks[1].y])
        left_eye = np.array([landmarks[33].x, landmarks[33].y])
        right_eye = np.array([landmarks[263].x, landmarks[263].y])
        
        # 偏航角 (左右转头): 基于鼻尖与两眼的位置
        eye_mid = (left_eye + right_eye) / 2
        eye_dist = np.linalg.norm(right_eye - left_eye)
        nose_offset = (nose[0] - eye_mid[0]) / (eye_dist + 1e-6)
        yaw = np.degrees(np.arcsin(np.clip(nose_offset, -1, 1)))
        
        # 俯仰角 (点头): 基于鼻尖与眼睛的垂直关系
        pitch = np.degrees(np.arctan2(
            nose[1] - eye_mid[1], eye_dist
        ))
        
        # 翻滚角 (歪头): 两眼连线角度
        roll = np.degrees(np.arctan2(
            right_eye[1] - left_eye[1],
            right_eye[0] - left_eye[0]
        ))
        
        return pitch, yaw, roll


class IndividualCalibrator:
    """
    个性化校准器
    
    论文核心: 每个驾驶员建立独立中性基线
    - 收集正常驾驶时EAR, MAR, 头部角度
    - 计算各特征的均值和标准差
    - 后续检测基于相对偏差
    """
    
    def __init__(self, calibration_duration_sec=30, fps=30):
        self.duration = calibration_duration_sec
        self.fps = fps
        self.n_samples = calibration_duration_sec * fps
        
        self.baselines = {}
        self.is_calibrated = False
        
    def collect_baseline(self, features_list: list):
        """
        收集基线特征
        
        Args:
            features_list: list of feature dicts from FaceLandmarkExtractor
        """
        if len(features_list) < self.n_samples * 0.5:
            return False
        
        ears = [f['ear'] for f in features_list if f is not None]
        mars = [f['mar'] for f in features_list if f is not None]
        pitches = [f['pitch'] for f in features_list if f is not None]
        yaws = [f['yaw'] for f in features_list if f is not None]
        rolls = [f['roll'] for f in features_list if f is not None]
        
        self.baselines = {
            'ear_mean': np.mean(ears),
            'ear_std': np.std(ears),
            'mar_mean': np.mean(mars),
            'mar_std': np.std(mars),
            'pitch_mean': np.mean(pitches),
            'pitch_std': np.std(pitches),
            'yaw_mean': np.mean(yaws),
            'yaw_std': np.std(yaws),
            'roll_mean': np.mean(rolls),
            'roll_std': np.std(rolls),
        }
        
        self.is_calibrated = True
        return True
    
    def get_normalized_features(self, features: dict) -> np.ndarray:
        """
        获取相对基线的归一化特征
        
        Returns:
            feat_vector: (10,) 归一化特征
        """
        if not self.is_calibrated:
            return np.zeros(10)
        
        # Z-score 归一化
        feat = np.array([
            (features['ear'] - self.baselines['ear_mean']) / (self.baselines['ear_std'] + 1e-6),
            (features['mar'] - self.baselines['mar_mean']) / (self.baselines['mar_std'] + 1e-6),
            (features['pitch'] - self.baselines['pitch_mean']) / (self.baselines['pitch_std'] + 1e-6),
            (features['yaw'] - self.baselines['yaw_mean']) / (self.baselines['yaw_std'] + 1e-6),
            (features['roll'] - self.baselines['roll_mean']) / (self.baselines['roll_std'] + 1e-6),
            features['ear'],  # 原始值也保留
            features['mar'],
            features['pitch'],
            features['yaw'],
            features['roll'],
        ])
        
        return feat


class DriverStateClassifier:
    """
    驾驶员状态分类器
    
    论文对比: SVM vs RF vs CNN vs RNN
    
    SVM最优: 91% F1 (室内) / 稳定 (室外)
    RF: 91% F1 (室内) / 70% F1 (室外) - 跨场景退化
    CNN: ~80% F1 / RNN: ~80% F1
    """
    
    def __init__(self, model_type='svm'):
        self.config = DriverStateConfig()
        self.calibrator = IndividualCalibrator()
        
        if model_type == 'svm':
            self.classifier = SVC(
                kernel='rbf', C=10.0, gamma='scale',
                probability=True, random_state=42
            )
        elif model_type == 'rf':
            self.classifier = RandomForestClassifier(
                n_estimators=100, max_depth=10, random_state=42
            )
        else:
            raise ValueError(f"Unknown model type: {model_type}")
            
        self.model_type = model_type
        self.is_trained = False
        
        # 眨眼追踪
        self.blink_counter = 0
        self.blink_timestamps = []
        
    def train(self, X_train, y_train):
        """训练分类器"""
        self.classifier.fit(X_train, y_train)
        self.is_trained = True
        
    def predict(self, features: dict) -> dict:
        """
        预测驾驶员状态
        
        Args:
            features: from FaceLandmarkExtractor
            
        Returns:
            result: dict with state, confidence, features
        """
        if not self.is_trained or not self.calibrator.is_calibrated:
            return {'state': 'unknown', 'confidence': 0}
        
        # 获取归一化特征
        feat_vec = self.calibrator.get_normalized_features(features).reshape(1, -1)
        
        # 预测
        state_idx = self.classifier.predict(feat_vec)[0]
        proba = self.classifier.predict_proba(feat_vec)[0]
        confidence = proba[state_idx]
        
        state_name = self.config.states[state_idx]
        
        # 混淆分析: out_of_safe_zone 与 neutral 易混淆
        if state_name == 'out_of_safe_zone' and confidence < 0.6:
            # 重新检查头部角度
            if abs(features['yaw']) < self.config.yaw_threshold * 0.7:
                state_name = 'neutral'  # 降级
        
        return {
            'state': state_name,
            'state_index': state_idx,
            'confidence': confidence,
            'ear': features['ear'],
            'mar': features['mar'],
            'pitch': features['pitch'],
            'yaw': features['yaw'],
            'roll': features['roll'],
        }


# ========== 完整测试 ==========
if __name__ == "__main__":
    print("=" * 60)
    print("可解释驾驶员行为状态监测系统")
    print("=" * 60)
    
    config = DriverStateConfig()
    
    print(f"\n五种行为状态:")
    for i, state in enumerate(config.states):
        print(f"  {i}: {state}")
    
    print(f"\n关键阈值:")
    print(f"  EAR闭眼阈值: < {config.ear_threshold_closed}")
    print(f"  MAR打哈欠阈值: > {config.mar_threshold_yawn}")
    print(f"  头部偏航阈值: > {config.yaw_threshold}°")
    print(f"  眨眼频率阈值: > {config.blink_rate_threshold}/min")
    
    # 模拟特征提取
    extractor = FaceLandmarkExtractor()
    
    # 模拟一帧图像
    frame = np.random.randint(0, 255, (480, 640, 3), dtype=np.uint8)
    features = extractor.process(frame)
    
    if features:
        print(f"\n模拟特征提取:")
        print(f"  EAR: {features['ear']:.4f}")
        print(f"  MAR: {features['mar']:.4f}")
        print(f"  Pitch: {features['pitch']:.2f}°")
        print(f"  Yaw: {features['yaw']:.2f}°")
        print(f"  Roll: {features['roll']:.2f}°")
    
    # 校准器测试
    calibrator = IndividualCalibrator(calibration_duration_sec=10, fps=30)
    
    # 模拟基线特征 (30个样本)
    np.random.seed(42)
    baseline_features = []
    for _ in range(300):
        baseline_features.append({
            'ear': np.random.normal(0.30, 0.03),
            'mar': np.random.normal(0.15, 0.02),
            'pitch': np.random.normal(-5, 3),
            'yaw': np.random.normal(0, 5),
            'roll': np.random.normal(0, 2),
        })
    
    calibrator.collect_baseline(baseline_features)
    print(f"\n校准结果:")
    for k, v in calibrator.baselines.items():
        print(f"  {k}: {v:.4f}")
    
    # 分类器训练 (模拟数据)
    from sklearn.model_selection import train_test_split, cross_val_score
    
    # 生成模拟训练数据 (5类, 每类100样本)
    n_per_class = 100
    X_sim = []
    y_sim = []
    
    for class_idx in range(5):
        for _ in range(n_per_class):
            # 每类不同的特征模式
            if class_idx == 0:  # neutral
                ear = np.random.normal(0.30, 0.03)
                mar = np.random.normal(0.15, 0.02)
                pitch = np.random.normal(0, 3)
                yaw = np.random.normal(0, 3)
                roll = np.random.normal(0, 2)
            elif class_idx == 1:  # eye_closure
                ear = np.random.normal(0.15, 0.02)
                mar = np.random.normal(0.15, 0.02)
                pitch = np.random.normal(0, 3)
                yaw = np.random.normal(0, 3)
                roll = np.random.normal(0, 2)
            elif class_idx == 2:  # out_of_safe_zone
                ear = np.random.normal(0.30, 0.03)
                mar = np.random.normal(0.15, 0.02)
                pitch = np.random.normal(0, 3)
                yaw = np.random.normal(35, 5)
                roll = np.random.normal(0, 2)
            elif class_idx == 3:  # yawning
                ear = np.random.normal(0.28, 0.03)
                mar = np.random.normal(0.75, 0.05)
                pitch = np.random.normal(-10, 3)
                yaw = np.random.normal(0, 3)
                roll = np.random.normal(0, 2)
            elif class_idx == 4:  # excessive_blinking
                ear = np.random.normal(0.22, 0.05)
                mar = np.random.normal(0.15, 0.02)
                pitch = np.random.normal(0, 3)
                yaw = np.random.normal(0, 3)
                roll = np.random.normal(0, 2)
            
            X_sim.append([ear, mar, pitch, yaw, roll,
                         ear, mar, pitch, yaw, roll])
            y_sim.append(class_idx)
    
    X_sim = np.array(X_sim)
    y_sim = np.array(y_sim)
    
    # 训练SVM
    classifier = DriverStateClassifier(model_type='svm')
    X_train, X_test, y_train, y_test = train_test_split(
        X_sim, y_sim, test_size=0.3, stratify=y_sim, random_state=42
    )
    classifier.train(X_train, y_train)
    
    # 评估
    from sklearn.metrics import classification_report, f1_score
    
    y_pred = classifier.classifier.predict(X_test)
    f1 = f1_score(y_test, y_pred, average='macro')
    print(f"\nSVM 分类器性能:")
    print(f"  Macro F1: {f1:.3f}")
    print(f"\n分类报告:")
    print(classification_report(y_test, y_pred,
                               target_names=config.states))
    
    # 对比RF
    rf_classifier = DriverStateClassifier(model_type='rf')
    rf_classifier.train(X_train, y_train)
    y_pred_rf = rf_classifier.classifier.predict(X_test)
    f1_rf = f1_score(y_test, y_pred_rf, average='macro')
    print(f"\nRF 分类器 F1: {f1_rf:.3f}")
    
    # 论文报告对比
    print(f"\n{'='*60}")
    print(f"论文报告性能对比:")
    print(f"{'模型':<15} {'室内F1':<12} {'室外F1':<12} {'说明'}")
    print(f"{'-'*60}")
    print(f"{'SVM (RBF)':<15} {'91%':<12} {'~88%':<12} {'最稳定'}")
    print(f"{'Random Forest':<15} {'91%':<12} {'~70%':<12} {'跨场景退化'}")
    print(f"{'CNN':<15} {'~80%':<12} {'~75%':<12} {'特征学习'}")
    print(f"{'RNN':<15} {'~80%':<12} {'~75%':<12} {'时序建模'}")
    print(f"{'启发式规则':<15} {'~81%':<12} {'-':<12} {'基线模型'}")

2. 论文关键发现

SVM vs RF 跨场景表现

指标	SVM (RBF)	Random Forest	说明
室内F1	91%	91%	相当
室外F1	~88%	~70%	RF严重退化
混淆矩阵	out_of_safe_zone ↔ neutral	同上但更严重	RF对头部角度敏感
稳定性	✅ 最佳	❌ 跨场景不稳定	SVM更适合车载

统计显著性 (Kruskal-Wallis)

特征	p值	显著性	说明
EAR (闭眼)	<0.001	✅***	闭眼检测可靠
MAR (打哈欠)	<0.001	✅***	打哈欠检测可靠
眨眼频率	<0.01	✅**	频繁眨眼有效
头部旋转 (out_of_safe_zone)	>0.05	❌	需要改进

3. HMI系统

"""
ROS + Foxglove HMI 集成
论文使用Polaris GEM e6自动驾驶车
"""

# ROS节点伪代码
class DriverMonitoringROS:
    """
    ROS节点: 驾驶员状态监测 + HMI可视化
    
    话题:
    - /driver/state (std_msgs/String): 当前状态
    - /driver/features (custom): EAR, MAR, 头部角度
    - /driver/calibration (std_msgs/Bool): 校准状态
    - /vehicle/state (autoware): 车辆状态
    - /perception/objects (autoware): 感知结果
    
    可视化: Foxglove Studio面板
    - 实时视频流 + 面部关键点
    - EAR/MAR 实时曲线
    - 状态指示灯 (5种颜色)
    - 车辆位置地图
    """
    
    def __init__(self):
        # 初始化ROS节点
        # self.node = rclpy.Node('driver_monitor')
        # 订阅传感器话题
        # 发布状态话题
        # 初始化MediaPipe + 分类器
        pass
    
    def run(self):
        """主循环: 30fps"""
        while True:
            # 1. 获取摄像头帧
            # 2. 提取面部特征
            # 3. 分类状态
            # 4. 发布到ROS话题
            # 5. Foxglove可视化
            pass

IMS 开发启示

1. 核心采纳建议

论文发现	IMS采纳建议	优先级
SVM优于RF	使用SVM作为分类后端	🔴 P0
个性化校准	前30秒自动建立基线	🔴 P0
EAR+MAR足够	不需复杂深度模型	🟡 P1
out_of_safe_zone弱	用头部yaw+pitch组合改进	🟡 P1
MediaPipe轻量	可在QCS8255上实时运行	✅

2. MediaPipe vs 商业方案

方案	精度	速度	成本	部署
MediaPipe	98.5%检测率	30fps (CPU)	免费	开源
Smart Eye	更高	60fps	$$$$	商业SDK
Tobii Pro	专业级	120Hz	$$$$$	研究用
自研CNN	可定制	需NPU	$$	研发成本

3. IMS集成方案

4. 与IMS现有模块的协同

IMS模块	论文方法补充	融合方式
疲劳检测	EAR + MAR + 眨眼频率	增强现有PERCLOS
分心检测	头部yaw/pitch	补充视线偏离
OOP检测	头部姿态异常	辅助压力垫
酒驾检测	❌ 不支持	需BiFuseNet

5. 部署性能估算

平台	模型	帧率	内存	CPU占用
QCS8255	MediaPipe+SVM	30fps	<200MB	~15%
QCS8290	MediaPipe+SVM	60fps	<200MB	~10%
Raspberry Pi 4	MediaPipe+SVM	15fps	<200MB	~60%

参考文献

1. Candela et al. “Interpretable Driver Behavioral State Monitoring using Facial Landmarks for Human-Autonomous Vehicle Interaction.” Frontiers in Robotics and AI, 2026. DOI: 10.3389/frobt.2026.1896859
2. MediaPipe Face Mesh: https://google.github.io/mediapipe/solutions/face_mesh
3. Soukupova, T., Cech, J. “Real-Time Eye Blink Detection using Facial Landmarks.” CVPR Workshop 2016.

---

论文的核心贡献是证明了”简单可解释方法(几何特征+SVM)可以媲美深度学习”，且跨场景鲁棒性更好。对于IMS开发，这意味着可以用MediaPipe+SVM快速实现MVP，无需复杂模型。个性化校准是关键创新——每个驾驶员的前30秒自动建立基线，后续所有检测基于相对偏差，解决了个体差异问题。