1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153
| """ 车载多模态心跳检测融合算法 基于 Scientific Reports 2026 论文方法 """
import numpy as np from scipy.signal import butter, filtfilt, find_peaks from dataclasses import dataclass from typing import Optional, List from enum import Enum
class ModalityType(Enum): ECG_STEERING = "ecg_steering" PPG_STEERING = "ppg_steering" IPPG_RGB = "ippg_rgb" IPPG_IR = "ippg_ir" IPPG_THERMAL = "ippg_thermal"
@dataclass class ModalityResult: """单模态心率检测结果""" modality: ModalityType heart_rate: Optional[float] confidence: float signal_quality: float coverage: float
class MultiModalHeartbeatFusion: """ 多模态心跳检测融合 策略: 1. ECG 优先(精度最高) 2. PPG 补充(当 ECG 失败时) 3. iPPG 补充(当接触式都失败时) 4. 加权融合(当多模态都可用时) """ MODALITY_PRIORITY = { ModalityType.ECG_STEERING: 0.40, ModalityType.PPG_STEERING: 0.25, ModalityType.IPPG_IR: 0.15, ModalityType.IPPG_RGB: 0.12, ModalityType.IPPG_THERMAL: 0.08 } def __init__(self): self.min_confidence = 0.3 def fuse(self, results: List[ModalityResult]) -> dict: """ 多模态融合 Returns: {'heart_rate': float, 'confidence': float, 'primary_source': str, 'modalities_used': list} """ valid = [r for r in results if r.heart_rate is not None and r.confidence > self.min_confidence] if not valid: return { 'heart_rate': 0, 'confidence': 0, 'primary_source': 'none', 'modalities_used': [] } valid.sort(key=lambda r: -self.MODALITY_PRIORITY.get(r.modality, 0)) if len(valid) == 1: r = valid[0] return { 'heart_rate': r.heart_rate, 'confidence': r.confidence, 'primary_source': r.modality.value, 'modalities_used': [r.modality.value] } total_weight = 0 weighted_hr = 0 modalities_used = [] for r in valid: w = self.MODALITY_PRIORITY.get(r.modality, 0) * r.confidence weighted_hr += r.heart_rate * w total_weight += w modalities_used.append(r.modality.value) fused_hr = weighted_hr / total_weight if total_weight > 0 else 0 hrs = [r.heart_rate for r in valid] hr_std = np.std(hrs) if hr_std > 15: primary = valid[0] return { 'heart_rate': primary.heart_rate, 'confidence': primary.confidence * 0.7, 'primary_source': primary.modality.value, 'modalities_used': [primary.modality.value] } return { 'heart_rate': fused_hr, 'confidence': min(total_weight / len(valid), 1.0), 'primary_source': valid[0].modality.value, 'modalities_used': modalities_used }
if __name__ == "__main__": fusion = MultiModalHeartbeatFusion() results_day = [ ModalityResult(ModalityType.ECG_STEERING, 72, 0.9, 0.85, 0.95), ModalityResult(ModalityType.PPG_STEERING, 73, 0.85, 0.80, 0.90), ModalityResult(ModalityType.IPPG_RGB, 71, 0.7, 0.65, 0.80), ModalityResult(ModalityType.IPPG_IR, 72, 0.75, 0.70, 0.85), ModalityResult(ModalityType.IPPG_THERMAL, 70, 0.5, 0.45, 0.60), ] r = fusion.fuse(results_day) print(f"白天: HR={r['heart_rate']:.1f}, 置信={r['confidence']:.2f}, 源={r['primary_source']}") results_night = [ ModalityResult(ModalityType.ECG_STEERING, 65, 0.85, 0.80, 0.90), ModalityResult(ModalityType.PPG_STEERING, 66, 0.80, 0.75, 0.85), ModalityResult(ModalityType.IPPG_RGB, None, 0.1, 0.05, 0.0), ModalityResult(ModalityType.IPPG_IR, 64, 0.7, 0.65, 0.75), ModalityResult(ModalityType.IPPG_THERMAL, None, 0.2, 0.15, 0.0), ] r = fusion.fuse(results_night) print(f"夜间: HR={r['heart_rate']:.1f}, 置信={r['confidence']:.2f}, 源={r['primary_source']}") results_no_hands = [ ModalityResult(ModalityType.ECG_STEERING, None, 0.0, 0.0, 0.0), ModalityResult(ModalityType.PPG_STEERING, None, 0.0, 0.0, 0.0), ModalityResult(ModalityType.IPPG_RGB, 75, 0.65, 0.60, 0.75), ModalityResult(ModalityType.IPPG_IR, 74, 0.70, 0.65, 0.80), ModalityResult(ModalityType.IPPG_THERMAL, 73, 0.45, 0.40, 0.55), ] r = fusion.fuse(results_no_hands) print(f"手离方向盘: HR={r['heart_rate']:.1f}, 置信={r['confidence']:.2f}, 源={r['primary_source']}")
|