基于深度学习的多模态可穿戴传感器融合实现BFRB检测实战

1. 从无意识小动作说起BFRB 检测到底在解决什么问题身体聚焦重复行为Body-Focused Repetitive Behaviors简称 BFRB的自动检测是一个看起来简单、做起来却相当繁琐的工程问题。难点不在代码量而在多路传感器数据怎么对齐、怎么分段、怎么融合以及模型在真实佩戴场景下如何保持稳定。网上资料往往只讲某一个环节要么只谈传感器选型要么只贴一段分类网络代码缺少一条完整可跑的链路。本文就以“Deep Multimodal Wearable Sensor Fusion for Detection of Body-Focused Repetitive Behaviors”为主线从概念、数据、模型到训练验证带你把一套最小可运行的检测原型搭起来。1.1 什么是身体聚焦重复行为BFRB 是一类重复性的、难以自控的身体相关行为常见表现包括拔毛癖Trichotillomania反复拔扯头发、眉毛、睫毛抓挠皮肤Skin Picking / Excoriation反复抠抓皮肤直至破损咬指甲Onychophagia反复啃咬指甲或甲周皮肤咬嘴唇、咬口腔黏膜等其他习惯性动作。这类行为在人群中并不少见但在传统诊疗中主要依赖患者自述和临床问诊。问题在于很多患者在无意识状态下完成这些动作事后自己都记不清频率导致医生得到的主观数据误差很大。因此如何客观、连续地捕捉这些行为成为一个很实际的工程问题。1.2 为什么选择可穿戴传感器与摄像头监控相比可穿戴设备有几个明显优势不受场地限制可以在日常环境中长时间佩戴不侵犯视觉隐私用户更容易接受传感器直接接触身体对手部、手臂、头部等部位的动作信息捕捉更精确。常见的做法是把加速度计Accelerometer、陀螺仪Gyroscope、肌电传感器EMG等集成在手表、手环或定制贴片上。当用户做出抓、拔、咬等动作时传感器数据会出现特定的小幅度、短时、重复性模式。检测算法要做的就是把这些模式从大量日常动作中区分出来。1.3 Deep Multimodal 与 Sensor Fusion 的含义Multimodal多模态指的是数据来源不止一种。例如同时使用加速度计和陀螺仪或者同时采集 IMU 与 EMG 数据。Sensor Fusion传感器融合指把多个模态的信息按一定策略合并成统一表示从而比单一传感器获得更高的识别准确率。Deep 在这里指的是用深度神经网络自动学习特征而不是手工设计阈值或特征规则。一句话概括本文的核心就是“用深度学习把多路可穿戴传感器数据融合起来自动识别 BFRB 动作”。这是一条从数据采集到模型部署的完整技术链路。2. 检测系统整体架构2.1 端到端链路一个典型的 BFRB 检测系统可以拆成下面几个环节数据采集可穿戴设备按固定频率采集多路传感器原始数据数据清洗去除异常值、填补缺失段、对齐时间戳滑窗分段将连续数据切成固定时长的片段特征/表示学习用深度学习模型从每个片段中提取特征分类决策输出当前片段是否属于 BFRB 行为后处理与告警对连续片段做平滑减少误报必要时触发提示或干预。这个链路里数据清洗和滑窗分段往往决定上限。模型结构再先进如果输入数据里含有大量空洞、错位、未对齐的片段训练出来的效果也不会好。2.2 常见传感器与采样设置传感器作用典型采样率说明加速度计捕捉手部/肢体加速度变化50–100 Hz功耗低适合长时间佩戴陀螺仪捕捉旋转角速度50–100 Hz与加速度计互补EMG 肌电捕捉肌肉放电活动200–1000 Hz数据量大信息更直接PPG 光电脉搏反映心率和血流变化25–64 Hz通常用于辅助状态判断实际项目中不一定要全部传感器都用上。优先选择低功耗、用户佩戴舒适、对目标动作区分度高的模态组合常见配置是“加速度计 陀螺仪”进阶方案再加入 EMG。2.3 多模态数据的对齐与分段多路传感器数据首先需要按时间戳对齐。设备端通常会给每包数据打上时间戳接收端可以按时间插值到统一采样率。对齐之后就是滑窗分段。窗口长度建议结合动作特点选择BFRB 单次动作往往持续 0.5 到 3 秒因此窗口通常设为 2 到 4 秒步长可以为 1 秒。窗口太短会丢失动作上下文太长又会引入大量无关动作增大误报。3. 环境准备与依赖说明3.1 工具链选型文章后续实战以 Python 为主深度学习框架使用 PyTorch。示例环境如下Python 3.9 或 3.10PyTorch 2.xNumPy、Pandas 用于数据处理scikit-learn 用于评估指标计算Matplotlib 用于可视化可选。版本需要根据你的实际环境调整本文重点演示设计思路不绑定某个具体小版本。如果你的显卡支持 CUDA训练会更快没有 GPU 也没关系示例模型参数很少CPU 也能完成训练。3.2 项目目录结构一个最小但完整的多模态检测项目推荐这样组织brfb_detection/ ├── data/ │ ├── raw/ # 原始传感器 CSV │ └── processed/ # 预处理后的窗口数据 ├── src/ │ ├── preprocess.py # 清洗、对齐、滑窗 │ ├── dataset.py # PyTorch Dataset │ ├── model.py # 多模态融合模型 │ ├── train.py # 训练与评估 │ └── config.py # 全局配置 └── checkpoints/ # 模型权重保存这样拆分的目的是让每个模块职责单一数据处理、模型定义、训练逻辑互不耦合方便后续替换传感器组合或调整网络结构。4. 数据预处理与融合策略4.1 传感器数据清洗与“空洞检测”传感器在传输过程中经常出现丢包表现为某段时间内完全没有数据或数据稀疏。这种“空洞”如果直接送入模型会引入大量伪特征。因此第一步要检测空洞区域hole detection再决定是插值还是丢弃。下面给出一个基于 Pandas 的示例。假设原始数据是每个传感器一个 CSV字段包含 timestamp、ax、ay、az、gx、gy、gz。import numpy as np import pandas as pd def detect_hole_intervals(df, time_coltimestamp, expected_delta_ms20, max_gap_ms200): 检测传感器时间序列中的空洞区域。 参数 ----- df : pd.DataFrame 包含时间戳列和传感器列的原始数据。 time_col : str 时间戳列名单位为秒。 expected_delta_ms : float 正常情况下的采样间隔毫秒。 max_gap_ms : float 超过该间隔视为空洞。 返回 ----- holes : list[tuple[int, int]] 空洞区域的起始/结束行索引列表。 df df.sort_values(time_col).reset_index(dropTrue) diff_ms (df[time_col].diff() * 1000).to_numpy() diff_ms[0] expected_delta_ms holes [] start None for i, gap in enumerate(diff_ms): if gap max_gap_ms: if start is None: start i - 1 else: if start is not None: holes.append((start, i - 1)) start None if start is not None: holes.append((start, len(df) - 1)) return holes这个函数的核心思路是简单的连续差值判断相邻两行时间戳间隔明显偏大就认为是空洞起始直到间隔恢复正常。得到的空洞区间可以用于后续插值或直接切掉。4.2 滑窗分段清洗后的数据需要切成固定长度的窗口。窗口长度、步长在config.py中统一管理。# filepath: src/config.py SAMPLE_RATE 50 # 对齐后的统一采样率Hz WINDOW_SEC 3 # 窗口长度秒 STRIDE_SEC 1 # 滑动步长秒 NUM_SENSOR_CHANNELS 6 # 例如 ax ay az gx gy gz NUM_CLASSES 2 # 0: 非BFRB, 1: BFRB对应的滑窗函数import numpy as np def sliding_windows(data, window_len, stride_len): 将多通道传感器数据切成窗口。 参数 ----- data : np.ndarray 形状为 (T, C) 的对齐后数据。 window_len : int 窗口样本点数。 stride_len : int 步长样本点数。 返回 ----- windows : np.ndarray 形状为 (N, window_len, C)。 windows [] n data.shape[0] for start in range(0, n - window_len 1, stride_len): windows.append(data[start:start window_len]) return np.stack(windows)这段代码是“拿到窗口”的最小实现。窗口数量大约为(T - window_len) / stride_len 1例如 3 分钟数据采样率 50Hz约 9000 行按 150 点窗口、50 点步长能切出约 178 个窗口。4.3 早融合、晚融合与混合融合多传感器数据的融合策略直接影响模型结构和训练难度。早融合Early Fusion把多路数据在输入层就直接拼接例如把加速度计和陀螺仪合并成 6 通道输入后续统一用一个网络处理。优点是实现简单缺点是各模态特征在浅层就被混合难以区分不同传感器自身的噪声特性。晚融合Late Fusion每个模态各自走一个特征提取分支最后把特征拼接或相加后再分类。优点是各分支可以独立设计适合采样率和特性差异大的模态组合。缺点是参数更多训练数据不足时容易过拟合。混合融合Hybrid Fusion在多个层级上逐步融合例如浅层各模态独立提取局部特征中层通过注意力机制交互最后全局决策。这是当前论文里最常见的设计思路也是本文实战部分采用的方式。5. 模型实战基于注意力机制的多分支融合网络5.1 模型整体设计我们设计一个适合 3 秒窗口、6 通道输入的轻量级模型结构如下两个分支一个分支处理加速度计三轴一个分支处理陀螺仪三轴每个分支包含两层 1D CNN用于提取局部时序特征两个分支的特征用多头注意力层做交互融合融合后的特征经全局池化、全连接层输出二分类概率是否 BFRB。选择分支结构而不是单一网络是为了体现“多模态融合”的思路每个传感器先学自己的特征再用注意力机制学习模态之间的关系避免把不同物理含义的信号强行混在一起。5.2 PyTorch 模型实现# filepath: src/model.py import torch import torch.nn as nn class SensorBranch(nn.Module): 单个传感器分支两层一维卷积 残差连接 def __init__(self, in_channels: int, hidden: int 32): super().__init__() self.conv1 nn.Sequential( nn.Conv1d(in_channels, hidden, kernel_size5, padding2), nn.BatchNorm1d(hidden), nn.ReLU(), ) self.conv2 nn.Sequential( nn.Conv1d(hidden, hidden, kernel_size5, padding2), nn.BatchNorm1d(hidden), nn.ReLU(), ) self.proj nn.Conv1d(in_channels, hidden, kernel_size1) self.pool nn.AdaptiveAvgPool1d(1) def forward(self, x: torch.Tensor) - torch.Tensor: # x: (B, C, T) out self.conv2(self.conv1(x)) out out self.proj(x) out self.pool(out).squeeze(-1) # (B, hidden) return out class MultiModalFusionModel(nn.Module): 多模态融合模型 - 加速度计分支 3 通道 - 陀螺仪分支 3 通道 - 注意力融合层 def __init__(self, acc_channels: int 3, gyro_channels: int 3, hidden: int 32, num_classes: int 2): super().__init__() self.acc_branch SensorBranch(acc_channels, hidden) self.gyro_branch SensorBranch(gyro_channels, hidden) # 注意力融合 self.fusion_dim hidden * 2 self.attn nn.MultiheadAttention(embed_dimself.fusion_dim, num_heads2, batch_firstTrue) self.classifier nn.Sequential( nn.Linear(self.fusion_dim, 64), nn.ReLU(), nn.Dropout(0.3), nn.Linear(64, num_classes), ) def forward(self, acc: torch.Tensor, gyro: torch.Tensor) - torch.Tensor: # acc 和 gyro 均为 (B, C, T) f_acc self.acc_branch(acc) # (B, hidden) f_gyro self.gyro_branch(gyro) # (B, hidden) fused torch.cat([f_acc, f_gyro], dim-1) # (B, hidden*2) # 注意力机制需要 (B, L, E)L 这里用 1 放在序列维度 fused_seq fused.unsqueeze(1) # (B, 1, E) attn_out, _ self.attn(fused_seq, fused_seq, fused_seq) attn_out attn_out.squeeze(1) # (B, E) logits self.classifier(attn_out) return logits这段代码有几个地方要说明SensorBranch中做了残差连接避免网络加深后梯度消失注意力层放在两个传感器分支的融合向量之后可以让模型学到“当前片段里加速度计信息更重要还是陀螺仪信息更重要”分类头加了 Dropout缓解小样本数据上的过拟合。注意这是一个针对窗口数据的离线分类模型。如果要做实时检测通常需要在模型前后加上滑窗缓冲区和后处理逻辑这点在后面的工程建议中再展开。5.3 Dataset 与训练脚本接下来实现 PyTorch Dataset。数据集假设已经由预处理阶段生成了两个.npy文件train_acc.npy、train_gyro.npy以及对应的标签train_label.npy。# filepath: src/dataset.py import numpy as np import torch from torch.utils.data import Dataset class SensorWindowDataset(Dataset): 加速度计与陀螺仪窗口数据对 def __init__(self, acc_path, gyro_path, label_path): self.acc np.load(acc_path) # (N, T, 3) self.gyro np.load(gyro_path) # (N, T, 3) self.labels np.load(label_path) # (N,) def __len__(self): return len(self.labels) def __getitem__(self, idx): acc torch.as_tensor(self.acc[idx], dtypetorch.float32) gyro torch.as_tensor(self.gyro[idx], dtypetorch.float32) label torch.as_tensor(self.labels[idx], dtypetorch.long) # 模型输入要求 (C, T)需要转置 return acc.permute(1, 0), gyro.permute(1, 0), label训练脚本负责加载数据、构建模型、训练并输出评估指标。# filepath: src/train.py import torch import torch.nn as nn from torch.utils.data import DataLoader from sklearn.metrics import accuracy_score, f1_score, confusion_matrix from config import NUM_CLASSES from dataset import SensorWindowDataset from model import MultiModalFusionModel def train_one_epoch(model, loader, criterion, optimizer, device): model.train() total_loss 0.0 for acc, gyro, label in loader: acc, gyro, label acc.to(device), gyro.to(device), label.to(device) optimizer.zero_grad() logits model(acc, gyro) loss criterion(logits, label) loss.backward() optimizer.step() total_loss loss.item() * len(label) return total_loss / len(loader.dataset) torch.no_grad() def evaluate(model, loader, device): model.eval() preds, gts [], [] for acc, gyro, label in loader: acc, gyro acc.to(device), gyro.to(device) logits model(acc, gyro) pred torch.argmax(logits, dim1) preds.extend(pred.cpu().numpy().tolist()) gts.extend(label.numpy().tolist()) acc accuracy_score(gts, preds) f1 f1_score(gts, preds, zero_division0) return acc, f1, confusion_matrix(gts, preds) def main(): device torch.device(cuda if torch.cuda.is_available() else cpu) train_ds SensorWindowDataset(data/processed/train_acc.npy, data/processed/train_gyro.npy, data/processed/train_label.npy) test_ds SensorWindowDataset(data/processed/test_acc.npy, data/processed/test_gyro.npy, data/processed/test_label.npy) train_loader DataLoader(train_ds, batch_size64, shuffleTrue) test_loader DataLoader(test_ds, batch_size64, shuffleFalse) model MultiModalFusionModel(num_classesNUM_CLASSES).to(device) criterion nn.CrossEntropyLoss() optimizer torch.optim.Adam(model.parameters(), lr1e-3) for epoch in range(30): loss train_one_epoch(model, train_loader, criterion, optimizer, device) if (epoch 1) % 5 0: acc, f1, cm evaluate(model, test_loader, device) print(fEpoch {epoch 1:02d} | loss{loss:.4f} | facc{acc:.4f} | f1{f1:.4f}) print(cm) torch.save(model.state_dict(), checkpoints/fusion_model.pt) if __name__ __main__: main()6. 运行与验证6.1 准备示例数据由于真实受试者数据通常涉及隐私和伦理审批本地复现时可以先用合成数据验证流程。下面生成两段正弦波叠加噪声的模拟数据分别模拟“日常动作”和“重复抓取动作”的加速度计与陀螺仪信号。import numpy as np def make_synthetic_data(n_windows200, window_len150): rng np.random.default_rng(42) acc [] gyro [] labels [] for i in range(n_windows): t np.arange(window_len) / 50.0 if rng.random() 0.5: # 类别 1模拟重复抓取有较高频周期性抖动 freq rng.uniform(2, 4) a (0.3 * np.sin(2 * np.pi * freq * t) rng.normal(0, 0.05, window_len)) g (0.2 * np.sin(2 * np.pi * freq * t 1.0) rng.normal(0, 0.05, window_len)) label 1 else: # 类别 0日常低频小幅动作 a (0.05 * np.sin(2 * np.pi * 0.3 * t) rng.normal(0, 0.05, window_len)) g (0.03 * np.sin(2 * np.pi * 0.2 * t 0.5) rng.normal(0, 0.05, window_len)) label 0 # 每路传感器生成 3 轴并加入不同相位偏移 acc.append(np.stack([a, np.roll(a, 2), np.roll(a, -3)], axis1)) gyro.append(np.stack([g, np.roll(g, -1), np.roll(g, 4)], axis1)) labels.append(label) return (np.stack(acc), np.stack(gyro), np.array(labels)) acc, gyro, label make_synthetic_data() np.save(data/processed/train_acc.npy, acc[:160]) np.save(data/processed/train_gyro.npy, gyro[:160]) np.save(data/processed/train_label.npy, label[:160]) np.save(data/processed/test_acc.npy, acc[160:]) np.save(data/processed/test_gyro.npy, gyro[160:]) np.save(data/processed/test_label.npy, label[160:])把上面脚本保存为src/make_synthetic_data.py并运行就能得到 160 个训练窗口、40 个测试窗口。首次运行前记得创建data/processed和checkpoints目录。6.2 训练输出说明运行训练脚本cd brfb_detection mkdir -p data/processed checkpoints python src/make_synthetic_data.py python src/train.py预期输出类似Epoch 05 | loss0.5921 | acc0.7750 | f10.7419 [[17 5] [ 4 14]] Epoch 10 | loss0.4237 | acc0.8500 | f10.8378 [[18 4] [ 2 16]] Epoch 15 | loss0.3011 | acc0.8750 | f10.8710 [[19 3] [ 2 16]]不同运行环境、随机种子下数字会有差异这属于正常现象。关键看趋势loss 逐步下降acc 和 f1 逐步上升说明模型在合成数据上确实学到了区分两类模式的规律。6.3 评估指标怎么解读在二分类场景中不能只看准确率。因为实际采集数据里“非 BFRB”样本往往远多于“BFRB”样本模型即使把所有样本都判为负类准确率也可能很高。因此需要重点看Precision精确率模型判为 BFRB 的样本中真正是 BFRB 的比例Recall召回率所有真实的 BFRB 样本中模型检出多少F1-ScorePrecision 和 Recall 的调和平均Confusion Matrix混淆矩阵看错在哪个方向便于分析误报和漏报。对于 BFRB 检测一般更关注“漏报”的代价。如果漏掉一次行为可能错过一次干预机会而误报可以靠后处理阶段的连续性判断来过滤。所以早期版本可以把阈值调得偏“敏感”一些。7. 常见问题与排查思路7.1 数据类别不均衡真实场景中 BFRB 行为出现的时长占比通常很低可能只有 5% 到 10%导致训练集中正样本极少。解决思路使用类别权重nn.CrossEntropyLoss(weightpos_weight)给少数类更大权重过采样少数类窗口或者对少数类窗口做小幅时间扭曲、加噪声等数据增强评估时以 F1、召回率为主而不是只看准确率。7.2 传感器噪声与伪影可穿戴设备在佩戴过程中会有松动、皮肤出汗、电磁干扰等问题导致信号中出现尖峰或漂移。排查步骤先画出原始波形肉眼确认异常区域检查是不是真实动作还是设备异常例如拔掉设备甩动时数据是否有对应响应对噪声严重的样本考虑增加带通滤波或去除异常尖峰如果某个设备持续异常需要先检查硬件佩戴方式而不是只改算法。7.3 模型过拟合当训练样本很少而模型参数很多时训练 loss 很低但验证指标不佳。具体表现为训练集 acc 接近 1测试集 acc 明显低。处理方法增大 Dropout 比例减小模型 hidden 维度使用数据增强使用早停early stopping在验证 loss 不再下降时停止训练如果条件允许引入迁移学习先在公开的日常动作数据集如运动识别数据集上预训练再在 BFRB 数据上微调。7.4 检测延迟与误触发离线分类可以接受秒级延迟但可穿戴设备上的实时检测对延迟很敏感。常见问题是模型在手臂正常摆动时被误判为 BFRB。建议后处理使用“连续 k 个窗口中有 m 个判为正类”再触发称为 detection switch 思路能明显降低单窗口误报在时间维度上对预测概率做平滑例如指数移动平均结合用户场景做约束例如只在静止或特定姿态下启用检测。7.5 检测管线“有 bug 但看不出错”有时候模型跑起来不报错但结果明显不合理例如 loss 一直不降、训练 acc 乱跳。按下面顺序排查问题现象常见原因排查思路loss 不下降标签与数据不对齐