Dapalm | IMS技术前沿

驾驶员压力估计：面部视频+生物信号融合方法与IMS落地

论文与技术背景

• 核心论文： “Combining Facial Videos and Biosignals for Stress Estimation During Driving” (Springer, 2025, NMITCON)
• 关联研究： “Driver Emotion Recognition Involving Multimodal Signals” (2024, ResearchGate)
• 关键论文： “Multimodal Car Driver Stress Recognition” (ACM, 2019)
• 最新方法： “Enhancing Car Safety with Multimodal Emotion Recognition using CNN-LSTM” (Evergreen, 2025)

核心创新

Springer 2025论文提出将 面部视频rPPG（远程光电容积脉搏波）+ 接触式生物信号 融合，在真实驾驶条件下实现压力估计。rPPG从普通面部视频中提取心率信息，无需接触式传感器。

rPPG原理

面部皮肤 → 血液容积变化（心跳）→ 颜色微小变化（绿通道最敏感）
摄像头捕捉 → 信号处理 → 心率/HRV提取

方法	接触方式	精度	可用性	IMS兼容性
ECG胸带	接触	⭐⭐⭐⭐⭐	❌ 消费者不接受	低
PPG指夹	接触	⭐⭐⭐⭐	⚠️ 需戴设备	中
rPPG摄像头	非接触	⭐⭐⭐	✅ 复用DMS摄像头	高
mmWave雷达	非接触	⭐⭐⭐	✅ 已有CPD雷达	高

技术实现

rPPG心率提取算法

"""
远程光电容积脉搏波(rPPG)心率提取

基于 Green/CHROM/POS 算法
从DMS面部视频中提取心率，无需接触式传感器

依赖: pip install numpy scipy opencv-python
参考: Poh et al. (2010), Verkruysse et al. (2008)
"""

import numpy as np
from scipy.signal import butter, filtfilt, find_peaks
from scipy.fft import fft, fftfreq
from typing import Tuple, Dict

class rPPGHeartRate:
    """
    基于面部视频的rPPG心率提取
    
    算法: CHROM (Chrominance-based rPPG)
    参考: De Haan & Jeanne (2013)
    
    核心原理:
    - 皮肤血液容积随心跳变化
    - 绿色通道对血液容积变化最敏感
    - 通过色度变换消除运动伪影
    """
    
    def __init__(self, fs: int = 30):
        """
        Args:
            fs: 视频帧率 (Hz)
        """
        self.fs = fs
        
    def extract_face_roi(self, frame: np.ndarray, landmarks=None) -> np.ndarray:
        """
        提取面部ROI区域
        
        Args:
            frame: 视频帧 (H, W, 3) BGR格式
            landmarks: 面部关键点 (可选，否则用肤色检测)
            
        Returns:
            roi: 面部ROI图像
        """
        if landmarks is not None:
            # 使用关键点提取额头+脸颊区域
            # landmarks格式: [(x, y), ...]
            x_coords = [p[0] for p in landmarks]
            y_coords = [p[1] for p in landmarks]
            
            x1, x2 = int(min(x_coords)), int(max(x_coords))
            y1, y2 = int(min(y_coords) * 0.3), int(max(y_coords) * 0.7)
            
            return frame[y1:y2, x1:x2]
        else:
            # 肤色检测 (YCbCr空间)
            ycbcr = np.array(frame)
            ycbcr = cv2.cvtColor(ycbcr, cv2.COLOR_BGR2YCrCb) if frame.dtype == np.uint8 else frame
            
            # 简化肤色掩码
            if frame.dtype == np.uint8:
                hsv = cv2.cvtColor(frame, cv2.COLOR_BGR2HSV)
                lower = np.array([0, 30, 60])
                upper = np.array([20, 150, 255])
                mask = cv2.inRange(hsv, lower, upper)
                
                # 取最大连通域
                contours, _ = cv2.findContours(mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
                if contours:
                    largest = max(contours, key=cv2.contourArea)
                    x, y, w, h = cv2.boundingRect(largest)
                    return frame[y:y+h, x:x+w]
            
            # 回退到中心区域
            h, w = frame.shape[:2]
            return frame[int(h*0.2):int(h*0.6), int(w*0.3):int(w*0.7)]
    
    def chrom_method(self, rgb_sequence: np.ndarray) -> np.ndarray:
        """
        CHROM算法提取rPPG信号
        
        色度信号 X = 3*R - 2*G
        色度信号 Y = 1.5*R + G - 1.5*B
        rPPG信号 = X - α*Y
        
        Args:
            rgb_sequence: RGB均值序列, shape=(n_frames, 3)
            
        Returns:
            rppg_signal: 提取的rPPG信号
        """
        R = rgb_sequence[:, 0].astype(np.float64)
        G = rgb_sequence[:, 1].astype(np.float64)
        B = rgb_sequence[:, 2].astype(np.float64)
        
        # 归一化
        R_norm = R / (R + G + B + 1e-10)
        G_norm = G / (R + G + B + 1e-10)
        B_norm = B / (R + G + B + 1e-10)
        
        # CHROM色度
        X = 3 * R_norm - 2 * G_norm
        Y = 1.5 * R_norm + G_norm - 1.5 * B_norm
        
        # 自适应alpha
        alpha = np.std(X) / (np.std(Y) + 1e-10)
        
        rppg = X - alpha * Y
        
        # 带通滤波 0.7-3.5 Hz (42-210 bpm)
        nyq = self.fs / 2
        b, a = butter(4, [0.7/nyq, 3.5/nyq], btype='band')
        rppg_filtered = filtfilt(b, a, rppg)
        
        return rppg_filtered
    
    def estimate_heart_rate(self, rppg_signal: np.ndarray, 
                            window_sec: int = 10) -> Tuple[float, Dict]:
        """
        从rPPG信号估计心率
        
        Args:
            rppg_signal: 滤波后rPPG信号
            window_sec: 分析窗口秒数
            
        Returns:
            heart_rate: 心率 (bpm)
            details: 详细信息
        """
        window = window_sec * self.fs
        n = len(rppg_signal)
        
        if n < window:
            window = n
        
        # 取最近窗口
        signal = rppg_signal[-window:]
        
        # FFT频谱分析
        freqs = fftfreq(len(signal), 1/self.fs)
        spectrum = np.abs(fft(signal))
        
        # 只看心跳频率范围 (0.7-3.0 Hz = 42-180 bpm)
        mask = (freqs >= 0.7) & (freqs <= 3.0)
        pos_freqs = freqs[mask]
        pos_spectrum = spectrum[mask]
        
        # 峰值频率
        if len(pos_spectrum) > 0:
            peak_idx = np.argmax(pos_spectrum)
            hr_freq = pos_freqs[peak_idx]
            heart_rate = hr_freq * 60
            
            # 峰值显著性
            peak_power = pos_spectrum[peak_idx]
            mean_power = np.mean(pos_spectrum)
            snr = peak_power / (mean_power + 1e-10)
        else:
            heart_rate = 0
            snr = 0
        
        # 时域：峰值间隔法
        peaks, _ = find_peaks(signal, distance=self.fs*0.4)
        if len(peaks) > 2:
            intervals = np.diff(peaks) / self.fs
            hr_time = 60 / np.mean(intervals) if np.mean(intervals) > 0 else 0
        else:
            hr_time = 0
        
        # 融合估计
        if hr_freq > 0 and hr_time > 0:
            heart_rate = 0.6 * heart_rate + 0.4 * hr_time
        
        # HRV (RMSSD)
        if len(peaks) > 3:
            intervals = np.diff(peaks) / self.fs
            rmssd = np.sqrt(np.mean(intervals**2))
            sdnn = np.std(intervals)
        else:
            rmssd = 0
            sdnn = 0
        
        return heart_rate, {
            'hr_freq': heart_rate,
            'hr_time': hr_time,
            'snr': snr,
            'rmssd': rmssd,
            'sdnn': sdnn,
            'peak_count': len(peaks),
            'signal_quality': min(snr / 3.0, 1.0),
        }
    
    def process_video(self, frames: np.ndarray) -> Dict:
        """
        处理视频帧序列
        
        Args:
            frames: shape=(n_frames, H, W, 3) BGR
            
        Returns:
            {
                'heart_rate': 心率bpm,
                'hrv_rmssd': HRV,
                'signal_quality': 信号质量0-1,
                'stress_index': 压力指数
            }
        """
        n_frames = len(frames)
        rgb_means = np.zeros((n_frames, 3))
        
        for i, frame in enumerate(frames):
            roi = self.extract_face_roi(frame)
            rgb_means[i] = roi.mean(axis=(0, 1))  # BGR均值
        
        # 转为RGB
        rgb_means = rgb_means[:, ::-1]
        
        # CHROM算法
        rppg = self.chrom_method(rgb_means)
        
        # 估计心率
        hr, details = self.estimate_heart_rate(rppg)
        
        # 压力指数 (基于HRV)
        if details['rmssd'] > 0:
            # 低HRV = 高压力
            stress_index = 1.0 / (1.0 + details['rmssd'] * 10)
        else:
            stress_index = 0.5  # 未知
        
        return {
            'heart_rate': round(hr, 1),
            'hrv_rmssd': round(details['rmssd'], 3),
            'hrv_sdnn': round(details['sdnn'], 3),
            'signal_quality': round(details['signal_quality'], 2),
            'stress_index': round(stress_index, 2),
            'n_peaks': details['peak_count'],
        }


# === 测试 ===
import cv2

if __name__ == "__main__":
    np.random.seed(42)
    estimator = rPPGHeartRate(fs=30)
    
    # 模拟视频帧（生成带心跳脉冲的肤色帧）
    n_frames = 300  # 10秒@30fps
    frames = np.zeros((n_frames, 200, 200, 3), dtype=np.uint8)
    
    t = np.arange(n_frames) / 30.0
    hr = 72  # 1.2 Hz
    pulse = 5 * np.sin(2 * np.pi * hr/60 * t)  # 心跳脉冲
    
    for i in range(n_frames):
        base_color = np.array([120, 100, 90], dtype=np.float64)  # 肤色BGR
        modulated = base_color + pulse[i]
        frames[i] = np.clip(modulated, 0, 255).astype(np.uint8)
        frames[i] = np.random.randint(0, 10, (200, 200, 3), dtype=np.uint8) + frames[i]
    
    print("=== rPPG心率提取测试 ===")
    result = estimator.process_video(frames)
    
    for k, v in result.items():
        print(f"  {k}: {v}")
    
    print(f"\n真实心率: {hr} bpm")
    print(f"估计心率: {result['heart_rate']} bpm")
    print(f"误差: {abs(result['heart_rate'] - hr):.1f} bpm")
    
    print(f"\n信号质量: {result['signal_quality']:.0%}")
    print(f"压力指数: {result['stress_index']:.2f} (0=放松, 1=高压)")

多模态压力融合

面部rPPG + 驾驶行为 + 语音

class MultiModalStressFusion:
    """
    多模态压力估计融合
    
    信号源:
    1. rPPG心率/HRV (DMS摄像头)
    2. 驾驶行为激进度 (CAN总线)
    3. 语音压力特征 (麦克风)
    
    输出: 压力等级 0-100
    """
    
    def __init__(self):
        # 权重（来自论文经验值）
        self.weights = {
            'hrv': 0.35,        # HRV下降 → 压力升高
            'driving_aggr': 0.30, # 激进驾驶 → 压力
            'voice_stress': 0.20, # 语音特征 → 压力
            'facial_tension': 0.15, # 面部紧张度
        }
    
    def compute_stress_score(self, 
                             hrv_rmssd: float,
                             driving_metrics: Dict,
                             voice_features: Dict = None,
                             facial_features: Dict = None) -> Dict:
        """
        计算综合压力分数
        
        Args:
            hrv_rmssd: rPPG提取的RMSSD (ms)
            driving_metrics: {'steering_jerk', 'brake_force_var', 'speed_var'}
            voice_features: {'pitch_mean', 'pitch_var', 'energy'}
            facial_features: {'brow_furrow', 'jaw_clench', 'lip_tension'}
        """
        # HRV分量 (低HRV=高压力)
        # 正常RMSSD: 20-100ms, 压力下<20ms
        hrv_score = max(0, 1 - hrv_rmssd / 40)
        
        # 驾驶行为分量
        aggr = (
            min(driving_metrics.get('steering_jerk', 0) / 500, 1) +
            min(driving_metrics.get('brake_force_var', 0) / 1000, 1) +
            min(driving_metrics.get('speed_var', 0) / 50, 1)
        ) / 3
        
        # 语音分量
        if voice_features:
            # 压力下基频升高+变异增大
            pitch_score = min((voice_features.get('pitch_mean', 150) - 120) / 80, 1)
            pitch_var_score = min(voice_features.get('pitch_var', 0) / 50, 1)
            voice_score = (pitch_score + pitch_var_score) / 2
        else:
            voice_score = 0.5  # 默认中性
        
        # 面部分量
        if facial_features:
            face_score = (
                facial_features.get('brow_furrow', 0.5) +
                facial_features.get('jaw_clench', 0.5) +
                facial_features.get('lip_tension', 0.5)
            ) / 3
        else:
            face_score = 0.5
        
        # 加权融合
        stress_score = (
            self.weights['hrv'] * hrv_score +
            self.weights['driving_aggr'] * aggr +
            self.weights['voice_stress'] * voice_score +
            self.weights['facial_tension'] * face_score
        ) * 100
        
        # 等级
        if stress_score < 30:
            level = 'relaxed'
        elif stress_score < 60:
            level = 'moderate'
        elif stress_score < 80:
            level = 'elevated'
        else:
            level = 'high'
        
        return {
            'stress_score': round(stress_score, 1),
            'level': level,
            'components': {
                'hrv': round(hrv_score, 2),
                'driving': round(aggr, 2),
                'voice': round(voice_score, 2),
                'facial': round(face_score, 2),
            }
        }

IMS落地路线

部署方案

模块	传感器	算法	硬件	延迟
rPPG心率	DMS摄像头	CHROM	8255 NPU	100ms
驾驶行为	CAN总线	统计特征	DSP	50ms
语音压力	麦克风	基频+能量	DSP	200ms
面部紧张	DMS摄像头	关键点	NPU	30ms
融合	-	加权	CPU	10ms
总计	-	-	-	<300ms

Euro NCAP关联

Euro NCAP要求	压力估计贡献	说明
驾驶员状态监测	✅ 压力→分心	压力影响注意力
无响应驾驶员	✅ 压力→冻结反应	极高压力下
疲劳检测	⚠️ 间接关联	压力+疲劳常共现
情绪安全	🔵 未来扩展	情绪→驾驶质量

参考文献

1. “Combining Facial Videos and Biosignals for Stress Estimation During Driving”, Springer NMITCON, 2025
2. De Haan & Jeanne, “Robust Pulse Rate From Chrominance-Based rPPG”, IEEE TBME, 2013
3. “Driver Emotion Recognition Involving Multimodal Signals”, ResearchGate, 2024
4. “Enhancing Car Safety with Multimodal Emotion Recognition”, Evergreen, 2025
5. “Multimodal Car Driver Stress Recognition”, ACM, 2019