实时数据管道：从噪声中提取信号

工业传感器每秒推送上百条数据，直接展示会淹没在噪声里。本文介绍如何设计一个可插拔的实时数据管道，在数据流中透明地插入过滤器，让噪声在到达界面之前就被消灭。

---

问题：为什么不在 UI 层做过滤？

采集端 100Hz 推送的数据，每一个通道都可能携带：

• 高斯白噪声：ADC 量化误差、热噪声，幅度 ±30，每帧随机跳变
• 脉冲野值：传感器断线瞬间输出最大值，单帧异常但后续恢复
• 电源纹波：50Hz 工频干扰叠加在信号上

最直观的做法是在曲线的 paintEvent 里对渲染数据做平滑。但这会引入三个问题：

1. 数据库和 UI 不一致：曲线看起来平滑，但导出 CSV 和回放仍然是原始噪声数据
2. 重复计算：每帧渲染都要重新过滤全部数据（30s 窗口 = 3000 个点 × 25FPS = 每秒 75000 次过滤）
3. 逻辑耦合：过滤代码混在渲染逻辑里，换一种过滤器要改 UI 代码

结论：过滤应该发生在数据写入缓冲之前，而非渲染时。

---

设计：管道模式

原始数据 dp
    │
    ▼
┌──────────────┐
│  FilterPipeline  │  ← 管道开关（一键启用/禁用）
│                    │
│  ├─ IFilter_0     │  ← 滑动平均 (窗口=8)
│  ├─ IFilter_1     │  ← 中值滤波 (窗口=5)
│  ├─ IFilter_2     │  ← 阈值告警 (上限=900)
│  └─ ...           │
│                    │
│  串联执行，按添加顺序  │
└──────┬───────────┘
       │
       ▼
 DataBuffer → SQLite → UI

核心接口

struct IFilter {
    virtual DataPoint process(const DataPoint &dp) = 0;  // 纯虚：接收一个点，返回一个点
    virtual QVector<int> channels() const;                // 空 = 全部通道生效
    virtual bool isEnabled() const;                       // 开关
};

为什么用虚函数而不是回调函数？

虚函数给每个过滤器提供了独立的 this 指针和成员变量空间。滑动平均需要维护 m_history[channel][window] 环形缓冲，中值滤波需要维护历史队列——这些状态如果用回调 + 闭包实现，要么用全局变量（线程不安全），要么每次构造 lambda 传参（内存碎片）。虚函数让过滤器的生命周期与管道一致，最干净。

串行引擎

DataPoint FilterPipeline::process(const DataPoint &dp) const {
    if (!m_enabled) return dp;          // 管道关闭 → 零开销直通

    DataPoint result = dp;
    for (const auto &f : m_filters) {
        if (f && f->isEnabled())
            result = f->process(result);  // 上一级输出 = 下一级输入
    }
    return result;
}

为什么是串行而不是并行？

滑动平均 → 中值滤波 → 阈值检测 有严格的顺序依赖：先去噪，再去尖刺，最后判断是否越限。反过来先检测阈值再降噪 = 噪声触发误报。单次过滤耗时 ≤ 0.001ms，并行引入的线程调度开销（~0.05ms）反而更贵。

通道选择性

不是所有传感器都需要同一种滤波：

// 创建滑动平均过滤器，指定只对通道 0 和 2 生效
auto ma0 = std::make_unique<MovingAverageFilter>(8);
ma0->setChannels({0, 2});
pipeline->addFilter(std::move(ma0));

// 内部实现：未选中通道原值透传
for (int ch = 0; ch < dp.channels.size(); ++ch) {
    if (!channelActive(ch)) continue;   // 未选中通道 → 原值保留
    // ... 正常过滤
}

---

实现细节

滑动平均：环形缓冲 vs 队列

方案	写入	读取（求平均）	内存
QQueue + push/pop	O(1)	O(N) 遍历	动态分配
环形缓冲	O(1)	O(N) 遍历	固定

窗口 N 固定时环形缓冲更优：不需要每帧分配/释放，CPU 缓存命中率高。实测 N=5 时环形缓冲比 QQueue 快约 40%。

m_history[ch][m_heads[ch]] = cur;              // 写入
m_heads[ch] = (m_heads[ch] + 1) % windowSize;  // 指针前移

m_filled[ch] 记录已填槽数。前 N 帧未填满时只对有效槽求平均，避免冷启动偏差。

EMA：为什么还需要简单平均？

EMA:  alpha = 2/(N+1),   EMA_t = α·v_t + (1-α)·EMA_{t-1}
Simple: avg = sum(最近N个) / N

EMA 只需要 2 次乘法 + 1 次加法，不依赖窗口缓冲。对最近值更敏感——适合跟踪趋势变化的信号。

但 EMA 也有代价：对历史值有”记忆效应”，一次脉冲尖刺的影响需要多帧才能衰减。工业场景下操作员更信任”等权重”的简单平均——每一帧的重要性相同。

两者都保留，让用户选择。

中值滤波：nth_element 的妙用

QVector<double> sorted = m_history[ch];         // 拷贝窗口数据
std::nth_element(sorted.begin(),
                 sorted.begin() + N/2,
                 sorted.end());
result = sorted[N/2];                           // 中位数

为什么不直接排序？std::sort 是 O(N log N)，std::nth_element 是 O(N)。N=5 时差距不大，但语义更精确——我们要的只是中位数，不是完整有序序列。

脉冲野值剔除的核心原理：窗口 [10, 10, 10, 10, 999] 排序后中位数 = 10，999 被丢弃。野值概率 2% 时，N=5 窗口中出现 ≥2 个野值的概率 ≈ 0.08%，窗口 5 已足够稳健。

---

集成点：为什么在 push() 之前？

ParseWorker 的数据流只有一个插入点：

// onFrameDecoded()
DataPoint filtered = m_pipeline.process(dp);  // ← 这里
m_buffer.push(filtered);                       // 曲线、表格都看到过滤后的值
m_dbManager.insert(filtered);                  // 数据库也一致

在 push 之前过滤，所有下游消费者看到的是同一份数据。如果做在 UI 层，导出 CSV 和数据库回放都会恢复原始噪声——用户困惑”为什么导出数据和界面不一样”。

---

性能

100Hz 采集 + 3 个过滤器 + 3 通道：

操作	每帧耗时
MA(5) Simple	0.0003ms
Median(5)	0.001ms
ThresholdAlarm	0.0005ms
管道总计	~0.002ms

对比同帧的 QPainter 渲染（~2ms）和 SQLite 写入（~0.5ms），管道开销不到渲染的 0.1%。

---

优化方向

1. 管道禁用的快速路径：if (!m_enabled) return dp; 当前已实现，关闭管道时零开销
2. 管道无信号通知：addFilter/removeFilter 不发射信号，UI 需要手动轮询 — 可加 filterChanged() 信号
3. 配置持久化：关窗口后管道配置丢失 — T036 QSettings 可以记下来
4. 过滤器冲突检测：同通道两个 MA 叠加效果 = 双重平滑，可能不是用户本意 — 可加提示

---

总结

管道模式的本质是关注点分离：ParseWorker 只负责解码和缓冲，FilterPipeline 只负责数据变换，UI 只负责渲染。三层各司其职，每层可以独立演进。今天加一个中值滤波，明天换一个卡尔曼滤波，ParseWorker 和 UI 一行代码都不需要改。