UTD-MHAD 多模态数据集实战：基于 PyTorch 的 4 种数据加载与预处理完整代码

UTD-MHAD 多模态数据集实战：基于 PyTorch 的 4 种数据加载与预处理完整代码

多模态数据融合已成为人体动作识别领域的重要研究方向。UTD-MHAD 作为同时包含 RGB 视频、深度视频、骨骼关节点和惯性传感器数据的公开数据集，为研究者提供了丰富的多模态实验基础。本文将深入探讨如何利用 PyTorch 实现四种模态数据的高效加载与预处理，并提供可直接集成到实际项目中的完整代码解决方案。

1. 数据集解析与环境准备

UTD-MHAD 数据集由德克萨斯大学达拉斯分校发布，包含 8 名受试者执行的 27 类日常动作，每个动作重复 4 次，共计 861 个有效序列。数据集采用多设备同步采集，确保了不同模态间的时间对齐。

关键文件结构：

UTD-MHAD/ ├── RGB/ # RGB视频(.avi) ├── Depth/ # 深度视频(.mat) ├── Skeleton/ # 骨骼数据(.mat) └── Inertial/ # 惯性传感器数据(.mat)

环境配置要求：

# 基础依赖库 pip install torch==1.12.0 torchvision==0.13.0 pip install scipy==1.9.0 matplotlib==3.5.2 opencv-python==4.6.0 pip install scikit-learn==1.1.1 pandas==1.4.3

硬件建议配置：

• GPU: NVIDIA GTX 1080 Ti 或更高性能显卡
• 内存: 16GB 以上
• 存储: SSD 硬盘以获得更好的数据加载性能

2. 多模态 Dataset 类实现

我们设计统一的 PyTorch Dataset 类处理四种模态数据，核心在于实现__getitem__方法的高效多模态数据返回。

import torch from torch.utils.data import Dataset import scipy.io as sio import cv2 import os class UTD_MHAD_Dataset(Dataset): def __init__(self, root_dir, modalities=['rgb', 'depth', 'skeleton', 'inertial'], transform=None, split='train', subjects_split=[1,3,5,7]): """ 参数: root_dir: 数据集根目录 modalities: 使用的模态列表 transform: 数据增强变换 split: 数据划分(train/test) subjects_split: 训练集受试者编号 """ self.root_dir = root_dir self.modalities = modalities self.transform = transform self.samples = self._build_sample_list(split, subjects_split) def _build_sample_list(self, split, subjects_split): samples = [] action_classes = sorted(os.listdir(os.path.join(self.root_dir, 'RGB'))) for action_idx, action in enumerate(action_classes): action_files = sorted(os.listdir(os.path.join(self.root_dir, 'RGB', action))) for f in action_files: parts = f.split('_') subject_id = int(parts[2][1:]) # 提取受试者编号 # 根据split参数筛选数据 if (split == 'train' and subject_id in subjects_split) or \ (split == 'test' and subject_id not in subjects_split): sample = { 'action': action_idx, 'subject': subject_id, 'base_name': '_'.join(parts[:3]) # 示例: a1_s1_t1 } samples.append(sample) return samples def _load_rgb(self, base_name): video_path = os.path.join(self.root_dir, 'RGB', f'a{base_name.split("_")[0][1:]}', f'{base_name}_color.avi') cap = cv2.VideoCapture(video_path) frames = [] while cap.isOpened(): ret, frame = cap.read() if not ret: break frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) frames.append(frame) cap.release() return torch.stack([torch.from_numpy(f).permute(2,0,1) for f in frames]) def _load_depth(self, base_name): mat_path = os.path.join(self.root_dir, 'Depth', f'{base_name}_depth.mat') depth_data = sio.loadmat(mat_path)['d_depth'] return torch.from_numpy(depth_data).unsqueeze(1) # 添加通道维度 def _load_skeleton(self, base_name): mat_path = os.path.join(self.root_dir, 'Skeleton', f'{base_name}_skeleton.mat') skeleton_data = sio.loadmat(mat_path)['d_skel'] return torch.from_numpy(skeleton_data) def _load_inertial(self, base_name): mat_path = os.path.join(self.root_dir, 'Inertial', f'{base_name}_inertial.mat') inertial_data = sio.loadmat(mat_path)['d_iner'] return torch.from_numpy(inertial_data) def __len__(self): return len(self.samples) def __getitem__(self, idx): sample = self.samples[idx] base_name = sample['base_name'] data = {} if 'rgb' in self.modalities: data['rgb'] = self._load_rgb(base_name) if 'depth' in self.modalities: data['depth'] = self._load_depth(base_name) if 'skeleton' in self.modalities: data['skeleton'] = self._load_skeleton(base_name) if 'inertial' in self.modalities: data['inertial'] = self._load_inertial(base_name) label = sample['action'] if self.transform: data = self.transform(data) return data, label

3. 多模态数据预处理技术

不同模态数据需要特定的预处理方法以确保模型输入的一致性。我们实现三种核心预处理技术：

3.1 时序对齐与归一化

跨模态时序同步方案：

def temporal_align(data_dict, target_frames=32): """ 将所有模态数据统一到相同的时间长度 参数: data_dict: 包含各模态数据的字典 target_frames: 目标帧数 返回: 时序对齐后的数据字典 """ aligned_data = {} for modality, data in data_dict.items(): # 获取当前模态的时间维度长度 T = data.shape[0] if len(data.shape) > 1 else 1 if T == 1: # 惯性数据等单时间序列 aligned_data[modality] = data.repeat(target_frames, 1) else: # 线性插值到目标长度 if modality in ['rgb', 'depth']: # 视觉数据使用双线性插值 aligned_data[modality] = F.interpolate( data.unsqueeze(0), size=(target_frames, *data.shape[2:]), mode='trilinear' if len(data.shape)==5 else 'bilinear' ).squeeze(0) else: # 其他数据使用线性插值 aligned_data[modality] = F.interpolate( data.unsqueeze(0).unsqueeze(0), size=(target_frames, data.shape[1]), mode='linear' ).squeeze(0).squeeze(0) return aligned_data

模态特定归一化方法：

def modality_specific_normalize(data_dict): """ 模态特定的归一化处理 """ normalized = {} if 'rgb' in data_dict: # RGB: [0,255] -> [0,1] normalized['rgb'] = data_dict['rgb'].float() / 255.0 if 'depth' in data_dict: # Depth: 假设Kinect深度范围[0,8000]mm depth = data_dict['depth'].float() normalized['depth'] = torch.clamp(depth / 8000.0, 0, 1) if 'skeleton' in data_dict: # Skeleton: 以髋关节(索引0)为中心，归一化到[-1,1] skeleton = data_dict['skeleton'].float() hip_joint = skeleton[:, 0:1, :] centered = skeleton - hip_joint max_val = torch.max(torch.abs(centered)) normalized['skeleton'] = centered / (max_val + 1e-6) if 'inertial' in data_dict: # Inertial: 各通道Z-score标准化 inertial = data_dict['inertial'].float() mean = inertial.mean(dim=0, keepdim=True) std = inertial.std(dim=0, keepdim=True) normalized['inertial'] = (inertial - mean) / (std + 1e-6) return normalized

3.2 数据增强策略

针对视觉模态(RGB和深度)设计时空数据增强：

class MultiModalAugment: def __init__(self, augment_prob=0.5): self.prob = augment_prob def __call__(self, data_dict): if random.random() < self.prob: # 时空随机裁剪 if 'rgb' in data_dict and 'depth' in data_dict: data_dict = self._spatial_temporal_crop(data_dict) # 颜色抖动(仅RGB) if 'rgb' in data_dict: data_dict['rgb'] = self._color_jitter(data_dict['rgb']) # 随机水平翻转 if random.random() < self.prob: data_dict = self._horizontal_flip(data_dict) return data_dict def _spatial_temporal_crop(self, data_dict, crop_size=(224,224), temporal_ratio=0.9): # 统一的空间裁剪 _, T, H, W, C = data_dict['rgb'].shape new_H, new_W = crop_size start_x = random.randint(0, W - new_W) start_y = random.randint(0, H - new_H) # 统一的时间裁剪 new_T = int(T * temporal_ratio) start_t = random.randint(0, T - new_T) for modality in ['rgb', 'depth']: if modality in data_dict: # 空间裁剪 data_dict[modality] = data_dict[modality][ :, start_t:start_t+new_T, start_y:start_y+new_H, start_x:start_x+new_W, : ] return data_dict def _color_jitter(self, rgb_data, brightness=0.2, contrast=0.2, saturation=0.2): # 在批次维度应用颜色抖动 jitter = torchvision.transforms.ColorJitter( brightness=brightness, contrast=contrast, saturation=saturation ) return torch.stack([jitter(img) for img in rgb_data]) def _horizontal_flip(self, data_dict): if 'rgb' in data_dict: data_dict['rgb'] = torch.flip(data_dict['rgb'], dims=[3]) if 'depth' in data_dict: data_dict['depth'] = torch.flip(data_dict['depth'], dims=[3]) if 'skeleton' in data_dict: # 翻转骨骼x坐标并交换左右关节 skeleton = data_dict['skeleton'] skeleton[:, :, 0] = -skeleton[:, :, 0] # 翻转x坐标 # 左右关节交换(根据UTD-MHAD骨骼结构) left_joints = [4,5,6, 10,11,12, 16,17,18, 22,23] right_joints = [7,8,9, 13,14,15, 19,20,21, 24,25] for l, r in zip(left_joints, right_joints): skeleton[:, [l,r], :] = skeleton[:, [r,l], :] data_dict['skeleton'] = skeleton return data_dict

3.3 多模态数据批处理

自定义 collate_fn 处理变长序列和不同模态：

def multimodal_collate_fn(batch): """ 处理不同模态数据的批处理 输入: batch是__getitem__返回的(data_dict, label)列表 输出: 批处理后的字典和标签张量 """ batch_data = {} batch_labels = [] # 收集所有存在的模态 modalities = set() for data, _ in batch: modalities.update(data.keys()) # 初始化批处理容器 for modality in modalities: if modality in ['rgb', 'depth']: # 视觉数据: (B,T,C,H,W) batch_data[modality] = [] elif modality == 'skeleton': # 骨骼数据: (B,T,J,D) batch_data[modality] = [] elif modality == 'inertial': # 惯性数据: (B,T,S) batch_data[modality] = [] # 填充数据 for data, label in batch: batch_labels.append(label) for modality in modalities: if modality in data: batch_data[modality].append(data[modality]) else: # 对于不存在的模态填充零 shape = infer_modality_shape(batch_data, modality) batch_data[modality].append(torch.zeros(shape)) # 转换为张量 for modality in batch_data: batch_data[modality] = torch.stack(batch_data[modality]) return batch_data, torch.tensor(batch_labels) def infer_modality_shape(batch_data, modality): """ 根据已有样本推断缺失模态的形状 """ for data in batch_data.values(): if len(data) > 0 and data[0] is not None: if modality == 'rgb': return (len(data[0]), 3, 224, 224) # 假设RGB形状 elif modality == 'depth': return (len(data[0]), 1, 224, 224) # 假设深度形状 elif modality == 'skeleton': return (len(data[0]), 20, 3) # 假设20个关节 elif modality == 'inertial': return (len(data[0]), 6) # 6轴惯性数据 # 默认形状 return { 'rgb': (32, 3, 224, 224), 'depth': (32, 1, 224, 224), 'skeleton': (32, 20, 3), 'inertial': (32, 6) }[modality]

4. 训练框架与多模态融合

基于上述组件构建完整的训练流程，支持灵活的多模态组合：

import torch.nn as nn from torch.utils.data import DataLoader from torch.optim import Adam from tqdm import tqdm class MultimodalActionRecognizer(nn.Module): def __init__(self, modalities, num_classes=27): super().__init__() self.modalities = modalities self.feature_extractors = nn.ModuleDict() self.classifiers = nn.ModuleDict() # 为每个模态初始化特征提取器 if 'rgb' in modalities: self.feature_extractors['rgb'] = RGBFeatureExtractor() if 'depth' in modalities: self.feature_extractors['depth'] = DepthFeatureExtractor() if 'skeleton' in modalities: self.feature_extractors['skeleton'] = SkeletonFeatureExtractor() if 'inertial' in modalities: self.feature_extractors['inertial'] = InertialFeatureExtractor() # 融合分类器 total_features = sum([extractor.output_dim for extractor in self.feature_extractors.values()]) self.fusion_classifier = nn.Linear(total_features, num_classes) def forward(self, x): features = [] for modality in self.modalities: if modality in x: mod_features = self.feature_extractors[modality](x[modality]) features.append(mod_features) # 特征级融合 fused = torch.cat(features, dim=1) return self.fusion_classifier(fused) def train_model(modalities=['rgb', 'depth'], num_epochs=50, batch_size=16): # 初始化数据集 train_dataset = UTD_MHAD_Dataset( root_dir='path/to/UTD-MHAD', modalities=modalities, transform=torchvision.transforms.Compose([ temporal_align, modality_specific_normalize, MultiModalAugment() ]), split='train' ) test_dataset = UTD_MHAD_Dataset( root_dir='path/to/UTD-MHAD', modalities=modalities, transform=torchvision.transforms.Compose([ temporal_align, modality_specific_normalize ]), split='test' ) # 数据加载器 train_loader = DataLoader( train_dataset, batch_size=batch_size, shuffle=True, collate_fn=multimodal_collate_fn, num_workers=4 ) test_loader = DataLoader( test_dataset, batch_size=batch_size, shuffle=False, collate_fn=multimodal_collate_fn, num_workers=4 ) # 初始化模型 device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') model = MultimodalActionRecognizer(modalities).to(device) criterion = nn.CrossEntropyLoss() optimizer = Adam(model.parameters(), lr=0.001) # 训练循环 best_acc = 0.0 for epoch in range(num_epochs): model.train() running_loss = 0.0 for inputs, labels in tqdm(train_loader, desc=f'Epoch {epoch+1}'): # 移动数据到设备 inputs = {k: v.to(device) for k, v in inputs.items()} labels = labels.to(device) # 前向传播 outputs = model(inputs) loss = criterion(outputs, labels) # 反向传播 optimizer.zero_grad() loss.backward() optimizer.step() running_loss += loss.item() # 验证集评估 model.eval() correct = 0 total = 0 with torch.no_grad(): for inputs, labels in test_loader: inputs = {k: v.to(device) for k, v in inputs.items()} labels = labels.to(device) outputs = model(inputs) _, predicted = torch.max(outputs.data, 1) total += labels.size(0) correct += (predicted == labels).sum().item() accuracy = 100 * correct / total print(f'Epoch {epoch+1}, Loss: {running_loss/len(train_loader):.4f}, ' f'Test Acc: {accuracy:.2f}%') # 保存最佳模型 if accuracy > best_acc: best_acc = accuracy torch.save(model.state_dict(), f'best_model_{"_".join(modalities)}.pth') print(f'Training completed. Best accuracy: {best_acc:.2f}%')

5. 实际应用与性能优化

5.1 多模态组合性能对比

我们在相同超参数设置下比较不同模态组合的性能表现：

5.2 工程优化技巧

内存优化：

# 使用Dataloader的persistent_workers减少进程创建开销 DataLoader(..., persistent_workers=True, prefetch_factor=2) # 使用混合精度训练 scaler = torch.cuda.amp.GradScaler() with torch.cuda.amp.autocast(): outputs = model(inputs) loss = criterion(outputs, labels) scaler.scale(loss).backward() scaler.step(optimizer) scaler.update()

磁盘IO优化：

# 使用LMDB数据库加速小文件读取 import lmdb class LMDB_UTD_MHAD: def __init__(self, lmdb_path): self.env = lmdb.open(lmdb_path, readonly=True) def get_data(self, key): with self.env.begin() as txn: data = txn.get(key.encode()) return pickle.loads(data)

多模态特征融合策略对比：

1. 早期融合：在输入层直接拼接原始数据
2. 中期融合：在各模态网络中间层进行特征交互
3. 晚期融合：独立处理各模态后融合分类结果

实验表明，针对UTD-MHAD数据集，中期融合在准确率和计算效率间取得了最佳平衡。