Dapalm | IMS技术前沿

ECG-驾驶行为因果融合四层疲劳评估：Pearson相关量化+自适应加权特征融合

论文信息

项目	内容
标题	A Study on the Correlation Between Driving Behavior and ECG Data in Driving Fatigue
期刊	Sensors (MDPI), Vol. 26, Issue 11, Article 13441
发表	2026年5月29日
链接	https://doi.org/10.3390/s26113441
PubMed	https://pubmed.ncbi.nlm.nih.gov/42280957/
核心方法	四层生理-行为融合+Pearson相关+自适应加权
准确率	97.8%（重度疲劳）
响应时间	≤0.5s

核心创新

1. 四层架构：输入→预处理→特征融合→决策，工业级管道设计
2. Pearson因果量化：用统计相关性量化ECG-驾驶行为的因果关系
3. 自适应加权融合：基于Pearson相关系数动态分配模态权重
4. 双模态交叉验证：ECG-HRV和驾驶行为互相验证，降低误报

问题定义

单模态局限

模态	指标	局限
ECG-HRV	LF/HF比值, RMSSD	需接触电极
驾驶行为	方向盘修正, 车道偏移	滞后认知疲劳
融合	ECG+行为	互补+提前检测

ECG-行为因果路径

flowchart TD
    A[ECG-HRV变化] -->|Pearson r| B[驾驶行为变化]
    B --> C[疲劳判定]
    A --> C
    D[驾驶时长] --> A
    D --> B
    C --> E[疲劳等级+置信度+风险]

方法详解

四层架构

import torch
import torch.nn as nn
import numpy as np
from scipy.stats import pearsonr
from dataclasses import dataclass

@dataclass
class ECGFeatures:
    """ECG-HRV特征"""
    # 时域
    mean_rr: float        # 平均RR间期
    sdnn: float            # RR标准差
    rmssd: float           # 相邻RR差值均方根
    pnn50: float           # 相邻RR差>50ms比例
    
    # 频域
    lf_power: float        # 低频功率(0.04-0.15Hz)
    hf_power: float        # 高频功率(0.15-0.4Hz)
    lf_hf_ratio: float     # LF/HF比值（交感/副交感平衡）
    total_power: float
    
    # 非线性
    sampen: float          # 样本熵
    dfa_alpha1: float      # 去趋势波动分析

@dataclass
class DrivingBehaviorFeatures:
    """驾驶行为特征"""
    steering_rate: float       # 方向盘修正频率(次/min)
    steering_var: float         # 方向盘修正方差
    lane_deviation: float       # 车道偏移均值
    lane_deviation_var: float   # 车道偏移方差
    speed_var: float            # 速度变异系数
    brake_count: int            # 制动次数
    accelerator_var: float      # 油门方差


class FourLayerFusionModel:
    """
    四层生理-行为融合疲劳评估模型
    
    Layer 1: 输入层 - ECG-HRV + 驾驶行为
    Layer 2: 预处理层 - 异常去除 + Min-Max归一化
    Layer 3: 特征融合层 - Pearson相关自适应加权
    Layer 4: 决策层 - 双模态交叉验证 + 分类
    """
    
    def __init__(self):
        self.correlation_matrix = None
        self.weights = None
        
        # 分类器
        from sklearn.ensemble import RandomForestClassifier
        self.classifier = RandomForestClassifier(
            n_estimators=100, max_depth=8
        )
    
    def fit(self, ecg_features: list, 
            behavior_features: list,
            labels: np.ndarray):
        """
        训练
        
        Layer 2: 预处理
        Layer 3: Pearson相关+自适应权重
        """
        # Layer 2: 预处理
        ecg_array = np.array([[f.mean_rr, f.sdnn, f.rmssd, f.pnn50,
                                f.lf_power, f.hf_power, f.lf_hf_ratio,
                                f.total_power, f.sampen, f.dfa_alpha1]
                               for f in ecg_features])
        
        behav_array = np.array([[f.steering_rate, f.steering_var,
                                  f.lane_deviation, f.lane_deviation_var,
                                  f.speed_var, f.brake_count,
                                  f.accelerator_var]
                                 for f in behavior_features])
        
        # Min-Max归一化
        ecg_norm = (ecg_array - ecg_array.min(axis=0)) / \
                   (ecg_array.max(axis=0) - ecg_array.min(axis=0) + 1e-8)
        behav_norm = (behav_array - behav_array.min(axis=0)) / \
                      (behav_array.max(axis=0) - behav_array.min(axis=0) + 1e-8)
        
        # Layer 3: Pearson相关分析
        self.correlation_matrix = self._compute_correlations(
            ecg_norm, behav_norm, labels
        )
        
        # 自适应权重（基于相关系数）
        self.weights = self._compute_weights(self.correlation_matrix)
        
        # 加权融合
        fused = np.concatenate([
            ecg_norm * self.weights['ecg'],
            behav_norm * self.weights['behavior']
        ], axis=1)
        
        # Layer 4: 训练分类器
        self.classifier.fit(fused, labels)
        
        return self
    
    def _compute_correlations(self, ecg, behav, labels):
        """计算各特征与疲劳的Pearson相关"""
        corr = {}
        ecg_names = ['mean_rr', 'sdnn', 'rmssd', 'pnn50',
                     'lf_power', 'hf_power', 'lf_hf_ratio',
                     'total_power', 'sampen', 'dfa_alpha1']
        behav_names = ['steering_rate', 'steering_var',
                       'lane_dev', 'lane_dev_var',
                       'speed_var', 'brake_count', 'accel_var']
        
        for i, name in enumerate(ecg_names):
            r, p = pearsonr(ecg[:, i], labels)
            corr[f'ecg_{name}'] = {'r': r, 'p': p}
        
        for i, name in enumerate(behav_names):
            r, p = pearsonr(behav[:, i], labels)
            corr[f'behav_{name}'] = {'r': r, 'p': p}
        
        # 模态间相关
        ecg_behav_corr = np.corrcoef(
            ecg.mean(axis=1), behav.mean(axis=1)
        )[0, 1]
        corr['ecg_behav_cross'] = ecg_behav_corr
        
        return corr
    
    def _compute_weights(self, corr):
        """基于相关系数计算自适应权重"""
        ecg_rs = [abs(v['r']) for k, v in corr.items() 
                  if k.startswith('ecg_')]
        behav_rs = [abs(v['r']) for k, v in corr.items() 
                    if k.startswith('behav_')]
        
        ecg_weight = np.mean(ecg_rs) / (np.mean(ecg_rs) + np.mean(behav_rs))
        behav_weight = 1 - ecg_weight
        
        return {'ecg': ecg_weight, 'behavior': behav_weight}
    
    def predict(self, ecg_feat: ECGFeatures,
                behav_feat: DrivingBehaviorFeatures) -> dict:
        """预测"""
        ecg_vec = np.array([[ecg_feat.mean_rr, ecg_feat.sdnn,
                              ecg_feat.rmssd, ecg_feat.pnn50,
                              ecg_feat.lf_power, ecg_feat.hf_power,
                              ecg_feat.lf_hf_ratio, ecg_feat.total_power,
                              ecg_feat.sampen, ecg_feat.dfa_alpha1]])
        
        behav_vec = np.array([[behav_feat.steering_rate,
                                behav_feat.steering_var,
                                behav_feat.lane_deviation,
                                behav_feat.lane_deviation_var,
                                behav_feat.speed_var,
                                behav_feat.brake_count,
                                behav_feat.accelerator_var]])
        
        # 加权
        fused = np.concatenate([
            ecg_vec * self.weights['ecg'],
            behav_vec * self.weights['behavior']
        ], axis=1)
        
        pred = self.classifier.predict(fused)[0]
        proba = self.classifier.predict_proba(fused)[0]
        
        # 双模态交叉验证
        ecg_only = ecg_vec * self.weights['ecg']
        behav_only = behav_vec * self.weights['behavior']
        
        # 置信度
        confidence = max(proba)
        
        # 风险等级
        if pred == 0:
            risk = 'Low'
        elif pred == 1:
            risk = 'Medium'
        else:
            risk = 'High'
        
        return {
            'fatigue_level': ['Normal', 'Mild', 'Severe'][pred],
            'confidence': confidence,
            'risk_level': risk,
            'ecg_weight': self.weights['ecg'],
            'behavior_weight': self.weights['behavior'],
        }


# 测试
if __name__ == "__main__":
    model = FourLayerFusionModel()
    
    # 模拟训练数据
    np.random.seed(42)
    n = 100
    ecg_data = [ECGFeatures(
        mean_rr=800 + np.random.randn()*50,
        sdnn=40 + np.random.randn()*10,
        rmssd=30 + np.random.randn()*8,
        pnn50=10 + np.random.randn()*3,
        lf_power=500 + np.random.randn()*100,
        hf_power=400 + np.random.randn()*80,
        lf_hf_ratio=1.2 + np.random.randn()*0.3,
        total_power=1000 + np.random.randn()*200,
        sampen=1.5 + np.random.randn()*0.2,
        dfa_alpha1=0.8 + np.random.randn()*0.1,
    ) for _ in range(n)]
    
    behav_data = [DrivingBehaviorFeatures(
        steering_rate=5 + np.random.randn()*2,
        steering_var=3 + np.random.randn()*1,
        lane_deviation=0.3 + np.random.randn()*0.1,
        lane_deviation_var=0.1 + np.random.randn()*0.03,
        speed_var=0.05 + np.random.randn()*0.02,
        brake_count=2 + int(np.random.randn()*2),
        accelerator_var=0.1 + np.random.randn()*0.03,
    ) for _ in range(n)]
    
    labels = np.random.randint(0, 3, n)
    
    # 训练
    model.fit(ecg_data, behav_data, labels)
    
    # 预测
    test_ecg = ECGFeatures(
        mean_rr=900, sdnn=25, rmssd=15, pnn50=3,
        lf_power=800, hf_power=200, lf_hf_ratio=4.0,
        total_power=1000, sampen=0.8, dfa_alpha1=0.5
    )
    test_behav = DrivingBehaviorFeatures(
        steering_rate=1, steering_var=0.5,
        lane_deviation=0.8, lane_deviation_var=0.3,
        speed_var=0.15, brake_count=0, accelerator_var=0.2
    )
    
    result = model.predict(test_ecg, test_behav)
    print("四层融合疲劳评估:")
    print(f"  等级: {result['fatigue_level']}")
    print(f"  置信度: {result['confidence']:.2f}")
    print(f"  风险: {result['risk_level']}")
    print(f"  ECG权重: {result['ecg_weight']:.2f}")
    print(f"  行为权重: {result['behavior_weight']:.2f}")

实验结果

指标	单ECG	单行为	融合
正常准确率	88.2%	82.5%	95.3%
轻度疲劳	82.1%	75.8%	92.1%
重度疲劳	89.5%	80.2%	97.8%
响应时间	1.2s	0.8s	≤0.5s

Pearson相关系数

特征对	r值	p值	相关性
LF/HF↔steering_rate	-0.72	<0.001	强负相关
RMSSD↔lane_deviation	-0.58	<0.01	中等负相关
SDNN↔speed_var	0.45	<0.05	中等正相关
meanRR↔brake_count	-0.38	<0.05	弱负相关

IMS开发启示

1. 四层架构的工程价值

层级	功能	IMS对应
输入	ECG+CAN总线	rPPG+ADAS信号
预处理	归一化+去噪	标准管道
融合	Pearson自适应权重	动态模态权重
决策	交叉验证+分级	报警策略

2. 与rPPG管道的集成

组件	ECG方案	rPPG替代	优势
心率	胸带ECG	DMS摄像头rPPG	无接触
HRV	ECG精确	rPPG估计	精度略低
质量评估	ECG可靠	#03质量分类器	保证可靠性
CAN信号	方向盘+车道	同	直接复用

3. 与因果KAN(#06)对比

维度	因果KAN(#06)	Pearson四层(本论文)	选择
因果建模	深度学习因果图	统计Pearson	KAN更深
模态	EEG+EOG+ECG	ECG+行为	不同覆盖
准确率	93.5%	97.8%	Pearson更高
响应	-	≤0.5s	实时性强
复杂度	285K参数	RF轻量	四层更轻

总结

1. 97.8%重度疲劳准确率+≤0.5s响应：融合方案显著优于单模态
2. Pearson因果量化：LF/HF↔steering_rate r=-0.72强相关
3. 自适应权重：基于相关性动态分配ECG/行为权重
4. 四层架构工业级：输入→预处理→融合→决策管道清晰
5. 与rPPG+ADAS集成：无接触rPPG替代ECG+CAN信号复用