1D卷积神经网络在时序信号处理中的实战优化：从模型设计到推理加速

1次阅读

共计 3332 个字符，预计需要花费 9 分钟才能阅读完成。

在 ECG 分类、工业传感器监测等场景中，我们常遇到这些挑战：

特征工程复杂 ：手工设计时频域特征需要专业领域知识，且难以适应不同设备采集的信号差异
计算效率低下 ：传统方法（如 SVM+HMM）对长序列的处理时间复杂度呈指数增长
实时性要求高 ：边缘设备往往需要 10ms 内的响应速度，但原始信号可能包含 1000+ 采样点 / 秒

对比三种主流时序模型架构：

RNN/LSTM：
优势：天然适合序列建模
劣势：计算复杂度 O(n^2)，难以并行化；存在梯度消失问题
Transformer：
优势：强大的全局建模能力
劣势：自注意力机制带来 O(n^2) 内存消耗；对短序列过拟合
1D CNN：
平移不变性：相同模式在不同位置触发相同响应
层级感受野：浅层捕捉局部特征，深层整合全局信息
并行计算：卷积运算原生支持 GPU 加速

import torch
import torch.nn as nn

class ECGNet(nn.Module):
    def __init__(self, input_channels=12):
        super().__init__()
        # 输入形状：(batch, 12, 1000) 
        self.block1 = nn.Sequential(nn.Conv1d(input_channels, 32, kernel_size=7, padding=3),  # 保持长度不变
            nn.BatchNorm1d(32),
            nn.ReLU(),
            nn.MaxPool1d(2)  # -> (batch, 32, 500)
        )

        # 空洞卷积扩大感受野
        self.block2 = nn.Sequential(nn.Conv1d(32, 64, kernel_size=5, dilation=2, padding=4),  # 计算公式：padding = dilation*(kernel_size-1)//2
            nn.BatchNorm1d(64),
            nn.ReLU(),
            nn.MaxPool1d(2)  # -> (batch, 64, 250)
        )

        # 因果卷积（未来信息屏蔽）self.block3 = nn.Sequential(nn.Conv1d(64, 128, kernel_size=3, padding=1),
            nn.BatchNorm1d(128),
            nn.ReLU(),
            nn.MaxPool1d(2)  # -> (batch, 128, 125)
        )

        self.classifier = nn.Linear(128, 5)  # 假设 5 种心律分类

    def forward(self, x):
        x = self.block1(x)
        x = self.block2(x)
        x = self.block3(x)
        x = x.mean(dim=-1)  # 全局平均池化 -> (batch, 128)
        return self.classifier(x)

关键处理步骤：

滑动窗口分割 ：

def sliding_window(sequence, window_size, step):
    """
    参数：sequence: (n_samples, n_channels)
        window_size: 窗口长度（单位：采样点）step: 滑动步长
    返回：(n_windows, n_channels, window_size)
    """
    num_windows = (sequence.shape[0] - window_size) // step + 1
    windows = []
    for i in range(num_windows):
        start = i * step
        window = sequence[start:start+window_size].T  # 转置为通道优先
        windows.append(window)
    return np.stack(windows)

异常值处理 ：

def clip_outliers(data, threshold=3.0):
    median = np.median(data, axis=1, keepdims=True)
    mad = 1.4826 * np.median(np.abs(data - median), axis=1)  # 中位数绝对偏差
    upper = median + threshold * mad
    lower = median - threshold * mad
    return np.clip(data, lower, upper)

标准化 ：

class Normalizer:
    def __init__(self):
        self.mean = None
        self.std = None

    def fit(self, data):
        # data 形状：(n_samples, n_channels, n_timesteps)
        self.mean = np.mean(data, axis=(0, 2), keepdims=True)
        self.std = np.std(data, axis=(0, 2), keepdims=True)

    def transform(self, data):
        return (data - self.mean) / (self.std + 1e-8)

导出 ONNX 模型 ：

dummy_input = torch.randn(1, 12, 1000).to(device)
torch.onnx.export(
    model,
    dummy_input,
    "ecgnet.onnx",
    input_names=["input"],
    output_names=["output"],
    dynamic_axes={"input": {0: "batch", 2: "length"},
        "output": {0: "batch"}
    }
)

生成校准数据集 ：

class Calibrator(trt.IInt8EntropyCalibrator2):
    def __init__(self, data_loader):
        self.loader = data_loader
        self.current_index = 0

    def get_batch_size(self):
        return self.loader.batch_size

    def get_batch(self, names):
        if self.current_index >= len(self.loader):
            return None
        batch = next(iter(self.loader))
        self.current_index += 1
        return [batch[0].numpy().astype(np.float32)]

构建 TensorRT 引擎 ：

trtexec --onnx=ecgnet.onnx \
        --int8 \
        --calib=calibration.cache \
        --saveEngine=ecgnet.trt \
        --workspace=2048

测试环境：NVIDIA T4 GPU

指标	原始 PyTorch	TensorRT-FP16	TensorRT-INT8
延迟 (ms)	8.2	3.1	2.4
吞吐量 (样本 / 秒)	1219	3225	4166
GPU 内存 (MB)	890	510	320

卷积核设计原则 ：
第一层 kernel_size 建议覆盖 1 - 2 个典型波形周期（如 ECG 的 QRS 波群约 0.08-0.12 秒）
感受野计算公式：$RF_{l} = (RF_{l-1} – 1) \times stride + dilation \times (kernel_size – 1) + 1$
变长序列处理 ：

动态 padding 方案：

def pad_sequence(batch):
    max_len = max([x.shape[-1] for x in batch])
    padded = torch.zeros(len(batch), batch[0].shape[0], max_len)
    for i, x in enumerate(batch):
        padded[i, :, :x.shape[-1]] = x
    return padded