UL-DD多模态疲劳数据集：3D视频+生理信号+行为数据的完整采集方案

Nature Scientific Data 2026论文解读 | 深度学习疲劳检测数据基准

一、研究背景与数据集意义

1.1 疲劳驾驶安全挑战

疲劳驾驶是全球道路安全的重要威胁。据美国国家公路交通安全管理局（NHTSA）统计：

• 每年约10万起警察报告的疲劳驾驶事故
• 造成约1,550人死亡、71,000人受伤
• 疲劳驾驶事故的死亡率是普通事故的4倍

EU GSR法规要求：

• 2024年7月起，欧盟所有新车必须配备疲劳预警系统
• Euro NCAP 2026协议将疲劳检测纳入DSM评分（最高25分）

1.2 现有数据集局限

数据集	模态	时长	标注方式	局限性
NTHU-DDD	IR视频	~6小时	演绎式动作	非真实疲劳
DROZY	EEG+IR	7小时	KSS	样本量小
YawDD	RGB视频	~18小时	打哈欠标注	仅视频
SEED-VIG	EEG+EOG	~28小时	警觉度标注	无视频
UL-DD	多模态	95小时	KSS+生理	需采集设备

1.3 UL-DD核心优势

多模态完整性：

• 3D面部视频（深度+RGB+IR）
• 8种生理信号（HR, EDA, BVP, SPO2等）
• 行为信号（转向、握力、姿态）
• 总计19个传感器同步采集

科学标注：

• 每4分钟采集KSS主观评分
• 专家一致性检验（Cronbach’s α > 0.85）
• 捕捉疲劳渐进过程

flowchart TB
    subgraph 数据采集层
        A[19名受试者] --> B[模拟驾驶舱]
        
        subgraph 视觉传感器
            B --> C1[RGB摄像头]
            B --> C2[IR红外摄像头]
            B --> C3[3D深度摄像头]
        end
        
        subgraph 生理传感器
            B --> D1[Empatica E4腕带<br/>EDA/BVP/HR/ST]
            B --> D2[脉搏血氧仪<br/>SPO2]
            B --> D3[呼吸传感器<br/>RR]
        end
        
        subgraph 行为传感器
            B --> E1[方向盘力矩]
            B --> E2[握力传感器]
            B --> E3[姿态摄像头]
            B --> E4[加速度计]
        end
    end
    
    subgraph 数据同步
        C1 & C2 & C3 & D1 & D2 & D3 & E1 & E2 & E3 & E4 --> F[时间戳对齐<br/>PTP协议]
        F --> G[多模态融合数据集]
    end

二、实验设计与采集协议

2.1 受试者招募

项目	数值
受试者人数	19人
年龄范围	21-45岁
性别分布	13男 / 6女
驾驶经验	2-20年
筛选标准	无睡眠障碍、无心血管疾病

实验前准备：

• 睡眠剥夺：受试者在实验前需保持清醒16小时
• 对照实验：同一受试者在不同日期进行”清醒”和”疲劳”两次实验
• 知情同意：通过IRB伦理审查

2.2 数据采集协议

实验流程：

时间轴（分钟）:
0 ---- 10 ---- 20 ---- 30 ---- ... ---- 90 ---- 100
 |      |       |       |              |       |
 安装   预适应   开始    每4分钟         结束    卸载
 传感器  驾驶    正式    KSS采样        驾驶    传感器
                驾驶

KSS评分表（Karolinska Sleepiness Scale）：

1 = 极度清醒
3 = 清醒
5 = 既不清醒也不困倦
7 = 困倦，但无需努力保持清醒
9 = 极度困倦，需要极大努力保持清醒

2.3 传感器配置详解

"""
UL-DD传感器配置参数
"""

SENSOR_CONFIG = {
    # 视觉传感器
    'rgb_camera': {
        'resolution': '1920x1080',
        'fps': 30,
        'view': '面部正面',
        'format': 'MP4 (H.264)'
    },
    
    'ir_camera': {
        'resolution': '640x480',
        'fps': 30,
        'wavelength': '850nm',
        'purpose': '夜间/墨镜场景'
    },
    
    'depth_camera': {
        'model': 'Intel RealSense D435',
        'resolution': '1280x720',
        'fps': 30,
        'range': '0.3-3m',
        'depth_accuracy': '±2mm'
    },
    
    # 生理传感器（Empatica E4）
    'eda': {
        'sampling_rate': 4,  # Hz
        'unit': 'μS',
        'position': '手腕'
    },
    
    'bvp': {
        'sampling_rate': 64,  # Hz
        'type': 'PPG光电容积脉搏波',
        'purpose': 'HRV分析'
    },
    
    'hr': {
        'sampling_rate': 1,  # Hz
        'derivation': 'BVP峰值检测',
        'unit': 'BPM'
    },
    
    'skin_temp': {
        'sampling_rate': 4,  # Hz
        'unit': '°C',
        'position': '手腕'
    },
    
    'spo2': {
        'sampling_rate': 1,  # Hz
        'unit': '%',
        'device': '指尖血氧仪'
    },
    
    'respiration': {
        'sampling_rate': 1,  # Hz
        'type': '胸带传感器',
        'unit': '呼吸次数/分钟'
    },
    
    # 行为传感器
    'steering_torque': {
        'sampling_rate': 100,  # Hz
        'unit': 'Nm',
        'purpose': '转向微修正检测'
    },
    
    'grip_pressure': {
        'sampling_rate': 50,  # Hz,
        'position': '方向盘握把'
    },
    
    'posture_video': {
        'resolution': '640x480',
        'fps': 15,
        'view': '侧面姿态',
        'purpose': '坐姿变化检测'
    },
    
    'accelerometer': {
        'sampling_rate': 32,  # Hz
        'axes': '3轴',
        'position': '座椅下方',
        'purpose': '车身震动/移动检测'
    }
}


def generate_sync_timeline(duration_minutes=90, kss_interval=4):
    """生成数据采集时间线"""
    import pandas as pd
    
    timeline = {
        'timestamp': [],
        'event': [],
        'kss_score': []
    }
    
    start_time = pd.Timestamp('2024-01-01 09:00:00')
    
    # 初始10分钟：传感器安装和校准
    for t in range(0, 10):
        timeline['timestamp'].append(start_time + pd.Timedelta(minutes=t))
        timeline['event'].append('setup' if t < 5 else 'adaptation')
        timeline['kss_score'].append(None)
    
    # 正式驾驶：每4分钟KSS评分
    for t in range(10, duration_minutes, kss_interval):
        timeline['timestamp'].append(start_time + pd.Timedelta(minutes=t))
        timeline['event'].append('driving_with_kss')
        timeline['kss_score'].append(None)  # 实际评分需填入
    
    return pd.DataFrame(timeline)


# 打印传感器配置
if __name__ == "__main__":
    print("=" * 60)
    print("UL-DD Dataset Sensor Configuration")
    print("=" * 60)
    for sensor, config in SENSOR_CONFIG.items():
        print(f"\n{sensor}:")
        for key, value in config.items():
            print(f"  {key}: {value}")
    
    print("\n" + "=" * 60)
    print("Acquisition Timeline")
    print("=" * 60)
    timeline = generate_sync_timeline()
    print(timeline.head(15))

三、数据集结构与文件组织

3.1 目录结构

UL-DD/
├── metadata/
│   ├── participants.csv        # 受试者信息
│   ├── sessions.csv            # 实验会话记录
│   └── sensor_sync.csv         # 传感器同步信息
│
├── participants/
│   ├── P01/                    # 受试者01
│   │   ├── session_01/         # 会话01（疲劳状态）
│   │   │   ├── video/
│   │   │   │   ├── rgb/
│   │   │   │   │   └── rgb_video.mp4
│   │   │   │   ├── ir/
│   │   │   │   │   └── ir_video.mp4
│   │   │   │   └── depth/
│   │   │   │       ├── depth_00001.npy
│   │   │   │       └── ...
│   │   │   ├── biometric/
│   │   │   │   ├── eda.csv
│   │   │   │   ├── bvp.csv
│   │   │   │   ├── hr.csv
│   │   │   │   ├── temp.csv
│   │   │   │   ├── spo2.csv
│   │   │   │   └── respiration.csv
│   │   │   ├── behavioral/
│   │   │   │   ├── steering.csv
│   │   │   │   ├── grip.csv
│   │   │   │   ├── posture/
│   │   │   │   └── accelerometer.csv
│   │   │   └── labels/
│   │   │       ├── kss_scores.csv
│   │   │       └── events.csv
│   │   └── session_02/         # 会话02（清醒状态）
│   ├── P02/
│   └── ...
│
├── code/
│   ├── data_loader.py          # 数据加载工具
│   ├── preprocess.py           # 预处理脚本
│   └── visualization.py        # 可视化工具
│
└── docs/
    ├── data_descriptor.pdf     # 数据说明文档
    └── user_guide.pdf          # 使用指南

3.2 数据文件格式

"""
UL-DD数据加载示例
"""

import numpy as np
import pandas as pd
import cv2
from pathlib import Path


class ULDatasetLoader:
    """UL-DD数据集加载器"""
    
    def __init__(self, data_root):
        """
        Args:
            data_root: 数据集根目录
        """
        self.data_root = Path(data_root)
        self.participants = self._load_participants()
        
    def _load_participants(self):
        """加载受试者元数据"""
        meta_path = self.data_root / 'metadata' / 'participants.csv'
        return pd.read_csv(meta_path)
    
    def load_session(self, participant_id, session_id):
        """
        加载单个会话的所有数据
        
        Args:
            participant_id: 受试者ID (e.g., 'P01')
            session_id: 会话ID (e.g., 'session_01')
        
        Returns:
            session_data: dict 包含所有模态数据
        """
        session_path = self.data_root / 'participants' / participant_id / session_id
        
        session_data = {
            'video': self._load_video(session_path / 'video'),
            'biometric': self._load_biometric(session_path / 'biometric'),
            'behavioral': self._load_behavioral(session_path / 'behavioral'),
            'labels': self._load_labels(session_path / 'labels')
        }
        
        return session_data
    
    def _load_video(self, video_path):
        """加载视频数据"""
        video_data = {}
        
        # RGB视频
        rgb_path = video_path / 'rgb' / 'rgb_video.mp4'
        if rgb_path.exists():
            video_data['rgb'] = self._read_video_frames(rgb_path)
        
        # IR视频
        ir_path = video_path / 'ir' / 'ir_video.mp4'
        if ir_path.exists():
            video_data['ir'] = self._read_video_frames(ir_path)
        
        # 深度帧
        depth_path = video_path / 'depth'
        if depth_path.exists():
            depth_files = sorted(depth_path.glob('depth_*.npy'))
            video_data['depth'] = [np.load(f) for f in depth_files[:100]]  # 示例：取前100帧
        
        return video_data
    
    def _read_video_frames(self, video_path, max_frames=None):
        """读取视频帧"""
        frames = []
        cap = cv2.VideoCapture(str(video_path))
        
        frame_count = 0
        while cap.isOpened():
            ret, frame = cap.read()
            if not ret:
                break
            frames.append(frame)
            frame_count += 1
            if max_frames and frame_count >= max_frames:
                break
        
        cap.release()
        return frames
    
    def _load_biometric(self, bio_path):
        """加载生理信号"""
        biometric_data = {}
        
        signal_files = {
            'eda': 'eda.csv',
            'bvp': 'bvp.csv',
            'hr': 'hr.csv',
            'temp': 'temp.csv',
            'spo2': 'spo2.csv',
            'respiration': 'respiration.csv'
        }
        
        for name, filename in signal_files.items():
            file_path = bio_path / filename
            if file_path.exists():
                df = pd.read_csv(file_path)
                biometric_data[name] = df
        
        return biometric_data
    
    def _load_behavioral(self, behav_path):
        """加载行为数据"""
        behavioral_data = {}
        
        # 转向数据
        steering_path = behav_path / 'steering.csv'
        if steering_path.exists():
            behavioral_data['steering'] = pd.read_csv(steering_path)
        
        # 握力数据
        grip_path = behav_path / 'grip.csv'
        if grip_path.exists():
            behavioral_data['grip'] = pd.read_csv(grip_path)
        
        # 加速度计
        acc_path = behav_path / 'accelerometer.csv'
        if acc_path.exists():
            behavioral_data['accelerometer'] = pd.read_csv(acc_path)
        
        return behavioral_data
    
    def _load_labels(self, label_path):
        """加载标注数据"""
        labels = {}
        
        # KSS评分
        kss_path = label_path / 'kss_scores.csv'
        if kss_path.exists():
            labels['kss'] = pd.read_csv(kss_path)
        
        # 事件标注
        events_path = label_path / 'events.csv'
        if events_path.exists():
            labels['events'] = pd.read_csv(events_path)
        
        return labels


# 使用示例
if __name__ == "__main__":
    loader = ULDatasetLoader('/path/to/UL-DD')
    
    # 加载单个会话
    session_data = loader.load_session('P01', 'session_01')
    
    print("Loaded data modalities:")
    for modality, data in session_data.items():
        print(f"  {modality}: {list(data.keys())}")
    
    # 示例：分析KSS评分分布
    if 'kss' in session_data['labels']:
        kss_scores = session_data['labels']['kss']
        print("\nKSS Score Distribution:")
        print(kss_scores['score'].value_counts().sort_index())

3.3 时间同步方案

多模态数据的关键挑战是时间同步。UL-DD采用PTP（Precision Time Protocol）实现微秒级同步：

"""
多模态数据时间同步算法
"""

import numpy as np
from scipy import signal


class MultiModalSynchronizer:
    """多模态数据同步器"""
    
    def __init__(self, reference_sampling_rate=100):
        """
        Args:
            reference_sampling_rate: 参考采样率（Hz）
        """
        self.ref_rate = reference_sampling_rate
    
    def align_to_reference(self, signal_data, signal_rate, ref_timestamps):
        """
        将信号对齐到参考时间戳
        
        Args:
            signal_data: [N] 原始信号
            signal_rate: 信号原始采样率（Hz）
            ref_timestamps: 参考时间戳数组
        
        Returns:
            aligned_signal: 对齐后的信号
        """
        # 计算重采样比例
        resample_ratio = self.ref_rate / signal_rate
        
        # 重采样
        num_samples = int(len(signal_data) * resample_ratio)
        resampled = signal.resample(signal_data, num_samples)
        
        # 对齐到参考时间戳
        aligned = np.interp(
            ref_timestamps,
            np.linspace(0, 1, len(resampled)),
            resampled
        )
        
        return aligned
    
    def create_unified_timeline(self, session_data):
        """
        创建统一时间线
        
        Args:
            session_data: 包含多模态数据的字典
        
        Returns:
            unified_data: 时间对齐后的数据
        """
        # 以最高采样率信号为参考（如BVP 64Hz）
        # 创建统一时间戳（每秒100个采样点）
        duration_seconds = 90 * 60  # 90分钟
        unified_timestamps = np.linspace(0, duration_seconds, duration_seconds * 100)
        
        unified_data = {
            'timestamps': unified_timestamps,
            'modalities': {}
        }
        
        # 对齐各模态
        for modality, data in session_data.items():
            if modality == 'labels':
                continue
            
            for signal_name, signal_values in data.items():
                if isinstance(signal_values, np.ndarray):
                    aligned = self.align_to_reference(
                        signal_values,
                        signal_rate=4,  # 示例采样率
                        ref_timestamps=np.linspace(0, duration_seconds, len(unified_timestamps))
                    )
                    unified_data['modalities'][f"{modality}_{signal_name}"] = aligned
        
        return unified_data


# 示例：可视化时间对齐
def visualize_alignment(unified_data, start_time=0, duration=60):
    """可视化对齐后的多模态信号"""
    import matplotlib.pyplot as plt
    
    timestamps = unified_data['timestamps']
    start_idx = int(start_time * 100)
    end_idx = int((start_time + duration) * 100)
    
    time_slice = timestamps[start_idx:end_idx]
    
    fig, axes = plt.subplots(4, 1, figsize=(14, 10), sharex=True)
    
    # EDA
    if 'biometric_eda' in unified_data['modalities']:
        axes[0].plot(time_slice, unified_data['modalities']['biometric_eda'][start_idx:end_idx])
        axes[0].set_ylabel('EDA (μS)')
        axes[0].set_title('Electrodermal Activity')
    
    # HR
    if 'biometric_hr' in unified_data['modalities']:
        axes[1].plot(time_slice, unified_data['modalities']['biometric_hr'][start_idx:end_idx])
        axes[1].set_ylabel('HR (BPM)')
        axes[1].set_title('Heart Rate')
    
    # Steering
    if 'behavioral_steering' in unified_data['modalities']:
        axes[2].plot(time_slice, unified_data['modalities']['behavioral_steering'][start_idx:end_idx])
        axes[2].set_ylabel('Steering Torque (Nm)')
        axes[2].set_title('Steering Behavior')
    
    # Grip
    if 'behavioral_grip' in unified_data['modalities']:
        axes[3].plot(time_slice, unified_data['modalities']['behavioral_grip'][start_idx:end_idx])
        axes[3].set_ylabel('Grip Pressure')
        axes[3].set_title('Grip Pressure')
    
    axes[3].set_xlabel('Time (s)')
    plt.tight_layout()
    plt.savefig('multimodal_aligned.png', dpi=150)

四、深度学习疲劳检测模型

4.1 多模态融合架构

基于UL-DD数据集，可以构建多种疲劳检测模型：

"""
多模态疲劳检测模型
基于Transformer融合架构
"""

import torch
import torch.nn as nn
import torch.nn.functional as F
from torchvision.models import resnet18


class MultimodalFatigueDetector(nn.Module):
    """
    多模态疲劳检测模型
    融合视频、生理、行为信号
    """
    
    def __init__(self, num_classes=3, d_model=256):
        """
        Args:
            num_classes: 分类数（清醒/轻度疲劳/严重疲劳）
            d_model: 融合特征维度
        """
        super().__init__()
        
        # 视频编码器（基于ResNet）
        self.video_encoder = VideoEncoder(output_dim=d_model)
        
        # 生理信号编码器（基于Transformer）
        self.biometric_encoder = BiometricEncoder(
            input_dim=6,  # EDA, BVP, HR, TEMP, SPO2, RR
            d_model=128,
            output_dim=d_model
        )
        
        # 行为信号编码器
        self.behavioral_encoder = BehavioralEncoder(
            input_dim=5,  # Steering, Grip, Acc(x,y,z)
            d_model=128,
            output_dim=d_model
        )
        
        # 多模态融合Transformer
        self.fusion_transformer = nn.TransformerEncoder(
            nn.TransformerEncoderLayer(d_model=d_model, nhead=8, batch_first=True),
            num_layers=3
        )
        
        # 分类头
        self.classifier = nn.Sequential(
            nn.Linear(d_model, 128),
            nn.ReLU(),
            nn.Dropout(0.3),
            nn.Linear(128, num_classes)
        )
        
    def forward(self, video, biometric, behavioral):
        """
        Args:
            video: [B, T, C, H, W] 视频帧序列
            biometric: [B, T, 6] 生理信号
            behavioral: [B, T, 5] 行为信号
        
        Returns:
            logits: [B, num_classes]
        """
        # 编码各模态
        video_feat = self.video_encoder(video)        # [B, T, d_model]
        bio_feat = self.biometric_encoder(biometric)   # [B, T, d_model]
        behav_feat = self.behavioral_encoder(behavioral)  # [B, T, d_model]
        
        # 拼接为多模态序列
        multi_seq = torch.stack([video_feat, bio_feat, behav_feat], dim=1)  # [B, 3, T, d_model]
        B, M, T, D = multi_seq.shape
        multi_seq = multi_seq.view(B, M * T, D)  # [B, 3*T, d_model]
        
        # Transformer融合
        fused = self.fusion_transformer(multi_seq)  # [B, 3*T, d_model]
        
        # 全局池化
        fused = fused.mean(dim=1)  # [B, d_model]
        
        # 分类
        logits = self.classifier(fused)
        
        return logits


class VideoEncoder(nn.Module):
    """视频特征编码器"""
    
    def __init__(self, output_dim=256):
        super().__init__()
        
        # 使用预训练ResNet提取帧特征
        resnet = resnet18(pretrained=True)
        self.feature_extractor = nn.Sequential(*list(resnet.children())[:-1])
        
        # 时序建模
        self.temporal_encoder = nn.LSTM(512, output_dim, batch_first=True)
        
    def forward(self, x):
        """
        Args:
            x: [B, T, C, H, W]
        Returns:
            feat: [B, T, output_dim]
        """
        B, T, C, H, W = x.shape
        
        # 提取每帧特征
        x = x.view(B * T, C, H, W)
        frame_feat = self.feature_extractor(x)  # [B*T, 512, 1, 1]
        frame_feat = frame_feat.view(B, T, -1)   # [B, T, 512]
        
        # 时序编码
        feat, _ = self.temporal_encoder(frame_feat)  # [B, T, output_dim]
        
        return feat


class BiometricEncoder(nn.Module):
    """生理信号编码器"""
    
    def __init__(self, input_dim=6, d_model=128, output_dim=256):
        super().__init__()
        
        # 输入嵌入
        self.input_proj = nn.Linear(input_dim, d_model)
        
        # 位置编码
        self.pos_encoding = PositionalEncoding(d_model)
        
        # Transformer编码器
        encoder_layer = nn.TransformerEncoderLayer(
            d_model=d_model,
            nhead=4,
            dim_feedforward=256,
            dropout=0.1,
            batch_first=True
        )
        self.encoder = nn.TransformerEncoder(encoder_layer, num_layers=2)
        
        # 输出投影
        self.output_proj = nn.Linear(d_model, output_dim)
        
    def forward(self, x):
        """
        Args:
            x: [B, T, input_dim]
        Returns:
            feat: [B, T, output_dim]
        """
        # 嵌入
        x = self.input_proj(x)
        x = self.pos_encoding(x)
        
        # Transformer编码
        x = self.encoder(x)
        
        # 输出投影
        x = self.output_proj(x)
        
        return x


class BehavioralEncoder(nn.Module):
    """行为信号编码器"""
    
    def __init__(self, input_dim=5, d_model=128, output_dim=256):
        super().__init__()
        
        # 1D卷积提取局部特征
        self.conv_layers = nn.Sequential(
            nn.Conv1d(input_dim, 64, kernel_size=3, padding=1),
            nn.BatchNorm1d(64),
            nn.ReLU(),
            nn.Conv1d(64, d_model, kernel_size=3, padding=1),
            nn.BatchNorm1d(d_model),
            nn.ReLU()
        )
        
        # LSTM时序建模
        self.lstm = nn.LSTM(d_model, output_dim, batch_first=True)
        
    def forward(self, x):
        """
        Args:
            x: [B, T, input_dim]
        Returns:
            feat: [B, T, output_dim]
        """
        # Conv1d需要 [B, C, T]
        x = x.transpose(1, 2)  # [B, input_dim, T]
        x = self.conv_layers(x)  # [B, d_model, T]
        x = x.transpose(1, 2)    # [B, T, d_model]
        
        # LSTM
        x, _ = self.lstm(x)  # [B, T, output_dim]
        
        return x


class PositionalEncoding(nn.Module):
    """位置编码"""
    
    def __init__(self, d_model, max_len=5000):
        super().__init__()
        
        pe = torch.zeros(max_len, d_model)
        position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1)
        div_term = torch.exp(torch.arange(0, d_model, 2).float() * (-torch.log(torch.tensor(10000.0)) / d_model))
        
        pe[:, 0::2] = torch.sin(position * div_term)
        pe[:, 1::2] = torch.cos(position * div_term)
        
        self.register_buffer('pe', pe.unsqueeze(0))
        
    def forward(self, x):
        return x + self.pe[:, :x.size(1)]


# 模型测试
if __name__ == "__main__":
    model = MultimodalFatigueDetector(num_classes=3)
    
    # 模拟输入
    batch_size = 4
    seq_len = 300  # 3秒 @ 100Hz
    
    video = torch.randn(batch_size, seq_len, 3, 224, 224)
    biometric = torch.randn(batch_size, seq_len, 6)
    behavioral = torch.randn(batch_size, seq_len, 5)
    
    # 前向传播
    output = model(video, biometric, behavioral)
    
    print(f"Input video shape: {video.shape}")
    print(f"Input biometric shape: {biometric.shape}")
    print(f"Input behavioral shape: {behavioral.shape}")
    print(f"Output shape: {output.shape}")
    print(f"Model parameters: {sum(p.numel() for p in model.parameters()):,}")

4.2 训练流程

"""
UL-DD数据集训练流程
"""

import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import Dataset, DataLoader
import numpy as np
from sklearn.model_selection import train_test_split


class ULDD_Dataset(Dataset):
    """UL-DD数据集PyTorch封装"""
    
    def __init__(self, data_path, participant_list, mode='train'):
        """
        Args:
            data_path: 数据路径
            participant_list: 受试者列表
            mode: 'train', 'val', 'test'
        """
        self.data_path = data_path
        self.participant_list = participant_list
        
        # 加载并预处理数据
        self.samples = self._load_samples()
        
        # 划分训练/验证/测试
        train_ratio = 0.7
        val_ratio = 0.15
        
        n_samples = len(self.samples)
        if mode == 'train':
            self.samples = self.samples[:int(n_samples * train_ratio)]
        elif mode == 'val':
            self.samples = self.samples[int(n_samples * train_ratio):int(n_samples * (train_ratio + val_ratio))]
        else:
            self.samples = self.samples[int(n_samples * (train_ratio + val_ratio)):]
    
    def _load_samples(self):
        """加载样本"""
        samples = []
        # 实际实现需要加载真实数据
        # 这里用模拟数据演示
        for pid in self.participant_list:
            for session in range(2):
                for segment in range(20):
                    sample = {
                        'video': np.random.randn(30, 3, 224, 224),  # 30帧
                        'biometric': np.random.randn(30, 6),
                        'behavioral': np.random.randn(30, 5),
                        'label': np.random.randint(0, 3),
                        'kss': np.random.uniform(1, 9)
                    }
                    samples.append(sample)
        return samples
    
    def __len__(self):
        return len(self.samples)
    
    def __getitem__(self, idx):
        sample = self.samples[idx]
        
        video = torch.FloatTensor(sample['video'])
        biometric = torch.FloatTensor(sample['biometric'])
        behavioral = torch.FloatTensor(sample['behavioral'])
        label = torch.LongTensor([sample['label']])[0]
        
        return {
            'video': video,
            'biometric': biometric,
            'behavioral': behavioral,
            'label': label
        }


def train_multimodal_detector():
    """训练多模态疲劳检测模型"""
    
    # 配置
    config = {
        'data_path': '/path/to/UL-DD',
        'batch_size': 8,
        'epochs': 50,
        'lr': 1e-4,
        'weight_decay': 1e-4,
        'device': 'cuda' if torch.cuda.is_available() else 'cpu'
    }
    
    print(f"Using device: {config['device']}")
    
    # 数据集
    all_participants = [f'P{i:02d}' for i in range(1, 20)]
    
    train_dataset = ULDD_Dataset(config['data_path'], all_participants, mode='train')
    val_dataset = ULDD_Dataset(config['data_path'], all_participants, mode='val')
    test_dataset = ULDD_Dataset(config['data_path'], all_participants, mode='test')
    
    train_loader = DataLoader(train_dataset, batch_size=config['batch_size'], shuffle=True)
    val_loader = DataLoader(val_dataset, batch_size=config['batch_size'], shuffle=False)
    test_loader = DataLoader(test_dataset, batch_size=config['batch_size'], shuffle=False)
    
    # 模型
    model = MultimodalFatigueDetector(num_classes=3).to(config['device'])
    
    print(f"Model parameters: {sum(p.numel() for p in model.parameters()):,}")
    
    # 损失函数
    criterion = nn.CrossEntropyLoss()
    
    # 优化器
    optimizer = optim.AdamW(model.parameters(), lr=config['lr'], weight_decay=config['weight_decay'])
    
    # 训练循环
    best_val_acc = 0
    
    for epoch in range(config['epochs']):
        model.train()
        train_loss = 0
        train_correct = 0
        train_total = 0
        
        for batch in train_loader:
            video = batch['video'].to(config['device'])
            biometric = batch['biometric'].to(config['device'])
            behavioral = batch['behavioral'].to(config['device'])
            labels = batch['label'].to(config['device'])
            
            optimizer.zero_grad()
            outputs = model(video, biometric, behavioral)
            loss = criterion(outputs, labels)
            loss.backward()
            optimizer.step()
            
            train_loss += loss.item()
            _, predicted = outputs.max(1)
            train_total += labels.size(0)
            train_correct += predicted.eq(labels).sum().item()
        
        # 验证
        model.eval()
        val_correct = 0
        val_total = 0
        
        with torch.no_grad():
            for batch in val_loader:
                video = batch['video'].to(config['device'])
                biometric = batch['biometric'].to(config['device'])
                behavioral = batch['behavioral'].to(config['device'])
                labels = batch['label'].to(config['device'])
                
                outputs = model(video, biometric, behavioral)
                _, predicted = outputs.max(1)
                val_total += labels.size(0)
                val_correct += predicted.eq(labels).sum().item()
        
        train_acc = train_correct / train_total
        val_acc = val_correct / val_total
        
        print(f"Epoch {epoch+1}/{config['epochs']} | "
              f"Train Loss: {train_loss/len(train_loader):.4f} Acc: {train_acc:.4f} | "
              f"Val Acc: {val_acc:.4f}")
        
        if val_acc > best_val_acc:
            best_val_acc = val_acc
            torch.save(model.state_dict(), 'best_multimodal_fatigue_model.pth')
    
    print(f"\nBest validation accuracy: {best_val_acc:.4f}")


if __name__ == "__main__":
    train_multimodal_detector()

五、技术验证与基准结果

5.1 数据质量验证

论文对数据质量进行了严格验证：

生理信号验证：

• HR信噪比（SNR）> 20dB
• EDA基线漂移 < 5%
• BVP峰值检测准确率 > 98%

KSS与生理指标相关性：

HRV (LF/HF比值) vs KSS: r = -0.42, p < 0.001
EDA均值 vs KSS: r = 0.38, p < 0.001
皮肤温度 vs KSS: r = -0.31, p < 0.01

5.2 基准模型性能

模型	输入模态	准确率	F1分数
ResNet-18	RGB视频	78.3%	0.74
ResNet-18	IR视频	82.1%	0.79
LSTM	生理信号	71.5%	0.68
LSTM	行为信号	65.2%	0.61
Transformer	视频+生理	85.7%	0.83
多模态融合	全部	89.4%	0.87

六、数据集应用场景

6.1 研究方向

1. 早期疲劳检测：探索疲劳发生前的生理前兆
2. 跨个体泛化：开发受试者无关的疲劳检测模型
3. 实时系统集成：边缘部署与延迟优化
4. 个性化阈值：基于个体差异的疲劳阈值学习

6.2 产业应用

flowchart LR
    A[UL-DD数据集] --> B[模型训练]
    B --> C[验证测试]
    C --> D{部署方式}
    
    D -->|云端| E1[车联网平台<br/>疲劳监控服务]
    D -->|边缘| E2[T-Box车载终端<br/>实时预警]
    D -->|混合| E3[端云协同<br/>个性化更新]
    
    E1 --> F1[车队管理]
    E1 --> F2[保险定价]
    
    E2 --> F3[驾驶员预警]
    E2 --> F4[ADAS联动]
    
    E3 --> F5[持续学习]
    E3 --> F6[OTA升级]

七、数据获取与使用

7.1 数据访问

• 官方存储库：Nature Scientific Data
• 数据大小：约2.5TB（包含所有模态）
• 许可协议：CC BY 4.0（需引用论文）
• 下载方式：通过Figshare或作者提供的FTP

7.2 使用建议

"""
快速入门：UL-DD数据集使用示例
"""

# 1. 下载数据集后，解压到指定目录
DATA_ROOT = '/path/to/UL-DD'

# 2. 加载单个受试者数据
loader = ULDatasetLoader(DATA_ROOT)
session = loader.load_session('P01', 'session_01')

# 3. 提取关键特征
# 生理特征：HRV分析
def extract_hrv_features(bvp_signal):
    """从BVP提取HRV特征"""
    from scipy.signal import find_peaks
    peaks, _ = find_peaks(bvp_signal, distance=30)  # 假设采样率64Hz
    
    # RR间期
    rr_intervals = np.diff(peaks) / 64  # 转换为秒
    
    # HRV指标
    hrv = {
        'mean_rr': np.mean(rr_intervals),
        'std_rr': np.std(rr_intervals),
        'rmssd': np.sqrt(np.mean(np.diff(rr_intervals)**2)),
        'pnn50': np.sum(np.abs(np.diff(rr_intervals)) > 0.05) / len(rr_intervals)
    }
    return hrv

# 4. 视频特征：面部关键点
def extract_facial_features(video_frame):
    """提取面部关键点"""
    import mediapipe as mp
    
    face_mesh = mp.solutions.face_mesh.FaceMesh()
    results = face_mesh.process(video_frame)
    
    if results.multi_face_landmarks:
        landmarks = results.multi_face_landmarks[0]
        # 提取眼睛、嘴巴关键点
        left_eye = [landmarks.landmark[i] for i in [33, 160, 158, 133]]
        right_eye = [landmarks.landmark[i] for i in [362, 385, 387, 263]]
        
        # 计算EAR（Eye Aspect Ratio）
        ear = compute_ear(left_eye, right_eye)
        return ear
    
    return None

def compute_ear(left_eye, right_eye):
    """计算眼睛纵横比"""
    # 简化实现
    left_vertical = np.linalg.norm(np.array(left_eye[1].y) - np.array(left_eye[2].y))
    left_horizontal = np.linalg.norm(np.array(left_eye[0].x) - np.array(left_eye[3].x))
    ear = left_vertical / (left_horizontal + 1e-8)
    return ear

八、总结

UL-DD数据集为疲劳驾驶研究提供了宝贵资源：

核心优势：

1. 多模态完整性：19个传感器同步采集，覆盖视觉、生理、行为
2. 科学标注：KSS主观评分+生理验证，捕捉疲劳渐进过程
3. 高质量数据：严格质量控制，适合深度学习训练
4. 开放获取：Nature Scientific Data发表，CC BY 4.0许可

应用价值：

• 疲劳检测算法研发基准
• 多模态融合方法验证
• Euro NCAP DSM系统开发参考
• 跨学科研究（神经科学、交通安全）

未来方向：

• 扩充受试者样本（目标100+）
• 增加EEG/EOG参考信号
• 真实道路数据补充
• 持续更新与维护

---

本文为Nature Scientific Data论文解读，数据集使用请引用原文。