从零搭一套端到端 EEG 信号处理工具链

起点

学完信号处理、MNE-Python、PyQt6 之后，知识点是散的。滤波是滤波，ICA 是 ICA，GUI 是 GUI。需要一个项目把它们串起来——不是跑通一个 tutorial，而是做一个自己会用、别人能跑的工具。

目标：从原始 EEG 文件加载，一路走到解码结果输出，中间每一步都可配置、可替换。模块分层清晰，方便后续扩展。

整体架构

flowchart TB
    subgraph GUI["GUI (PyQt6)"]
        direction TB
        View["BatchTab (view)
4 步骤进度条
实现 IBatchView"]
        Presenter["BatchPresenter (controller)
RunState 状态机
worker 工厂可注入"]
        Worker["BatchWorker (QObject)
run in QThread"]
        View -->|signal/slot| Presenter
        Presenter --> Worker
    end

    App["Application
PipelineSession: load → preprocess
→ epochs → decode
含 _first_invalid 增量重执行"]

    subgraph Domain["Domain (纯函数)"]
        Preprocessor
        Epocher
        Dataset
        Config
    end

    Source["Source
FileSource + reader registry
EDF / FIF / EEGLAB / BrainVision"]

    Decoder["Decoder (6 种, 懒加载)
LDA / SSVEP / FBCCA
CSP / CNN / Transformer (GPT/BERT)"]

    Worker --> App
    App --> Domain
    App --> Source
    App --> Decoder

    style GUI fill:#e8f4ff,stroke:#1f6feb
    style App fill:#fff4e6,stroke:#d97706
    style Domain fill:#f0fdf4,stroke:#16a34a
    style Source fill:#fef2f2,stroke:#dc2626
    style Decoder fill:#fdf4ff,stroke:#a21caf

自上而下三层 + 两个横切关注点：

GUI（PyQt6）：BatchTab + BatchPresenter + BatchWorker，MVP 架构
Application：PipelineSession 编排 + _first_invalid 增量重执行状态机
Domain（纯函数）：preprocessor / epocher / dataset / config，没有 Qt 依赖
横切：source（4 种 EEG 格式 reader 注册表）、decoder（6 种注册式解码器）

逐层实现

1. 数据加载:4 种格式,一个注册表

EEG 数据格式碎片化严重:EDF、FIF、EEGLAB（.set/.fdt）、BrainVision（.vhdr）。用一个 reader 注册表 + 文件后缀分发:

# bci/source/readers.py
@register_reader('.edf')
def _read_edf(path, **kw):
    return mne.io.read_raw_edf(path, preload=True, **kw)

@register_reader('.fif')
def _read_fif(path, **kw):
    return mne.io.read_raw_fif(path, preload=True, **kw)

@register_reader('.set')
def _read_eeglab(path, **kw):
    return mne.io.read_raw_eeglab(path, preload=True, **kw)

@register_reader('.vhdr')
def _read_brainvision(path, **kw):
    return mne.io.read_raw_brainvision(path, preload=True, **kw)

注册的好处:新增格式只需写一个函数 + 一行 @register_reader(...),其他代码不动。GUI 通过下拉框列出所有支持格式,自动从注册表读取——list_readers() 是单一事实来源,不在 GUI 里硬编码。

加载完成后 GUI 的 Preprocess 步骤会立即画出该数据的功率谱密度——通道平均 + 0.5–40 Hz Passband 高亮（红色虚线为截止频率）：

实现上调用 MNE 的 raw.compute_psd()，按全通道 dB 平均后画出单根曲线 + 高亮滤波通带。Mu 节律（8-13 Hz）和 Beta 节律（13-30 Hz）的能量集中在通带内清晰可见。

如果要逐通道或单电极（如 C3/C4/Cz 运动皮层）的详细 PSD 分析，需要在脚本里直接用 MNE API（GUI 当前不暴露电极选择）。

2. 预处理:链式 Preprocessor

每个方法返回 self,可以一路点下去:

1 2	proc = Preprocessor(raw, FilterConfig(l_freq=1, h_freq=40)) proc.bandpass().notch([50]).set_reference('average')

底层全是 MNE 的 raw.filter() / raw.notch_filter() / raw.set_eeg_reference(),但包了一层 fluent API 让调用更自然。ICA 去伪迹也包了:

1 2	ica = proc.apply_ica(n_components=15) # 返回 MNE ICA 对象,可以手动选成分排除

ICA 在 Preprocessor 层可用，但当前 GUI 没有暴露 ICA 组件视图（只有 domain 函数能跑）。研究员可以在脚本里调用 ica.find_bads_eog() 自动识别眼电成分（前额分布 + 低频能量大是 EOG 典型特征），pipeline 不会自动剔除——伪迹剔除是研究决策，不是工程决策。

注意:所有预处理操作是离线的。ICA 需要完整数据做分解,所以这套工具链的去伪迹只能放在 preprocess 阶段;想做实时去伪迹只能改用幅值剔除或回归。

3. 数据源:FileSource 封装

FileSource 是 source 包的对外接口:

# bci/source/file_source.py
class FileSource:
    @classmethod
    def load(cls, filepath: Path | list[Path]) -> mne.io.Raw:
        if isinstance(filepath, list):
            return mne.concatenate_raws([cls._load_one(p) for p in filepath])
        return cls._load_one(filepath)

    @classmethod
    def _load_one(cls, path: Path) -> mne.io.Raw:
        suffix = path.suffix.lower()
        reader = READERS.get(suffix)
        if reader is None:
            raise ValueError(f"Unsupported format: {suffix}")
        return reader(path)

多文件加载（同一被试多个 run）直接用 MNE 的 concatenate_raws。这种"自动识别拼接"是真实场景的常见需求——PhysioNet 的运动想象数据一个被试有 14 个 run (S001R01.edf ~ S001R14.edf),需要合并成一条连续信号。

4. Epoch 提取:事件检测的兜底策略

MNE 提取事件有两种途径:stim 通道或 annotation。不一定哪个有数据,所以做兜底:

def find_events(raw, stim_channel=None):
    try:
        events = mne.find_events(raw, stim_channel=stim_channel)
    except ValueError:
        events, _ = mne.events_from_annotations(raw)
    return events

加了幅值剔除:reject 超过阈值的 epoch 自动丢弃。

Epoch 切分后立即可以看到 ERP（事件相关电位）形态——这是解码的输入特征。GUI 的 EpochPage 画跨类平均 ERP 对比（每个事件类一条 grand-average 曲线）+ trials-by-class 直方图：

上图是 PhysioNet EEGBCI S001R04 实跑结果（二分类 T1=Left vs T2=Right，把 T3/T4 取消勾选得到）：21 个 trial 跨 2 类（T1=Left 13, T2=Right 8），2 条 grand-average ERP 曲线（T1 绿 / T2 蓝）叠加显示——二分类的 ERP 对比比 3 类更清晰，分类依据一眼可见。点击 Run Pipeline 后 pipeline 跑完 create_epochs 步骤时，EpochPage 自动刷新。

N100/P200 等成分在 100-300ms 区间最显著，这是后续 Transformer 消融实验中"窗口长度选 85 而不是 106"的生理依据。

5. 解码器:插件注册,6 种可插拔

@register('lda')
class LDADecoder(Decoder): ...

@register('ssvep')
class SSVEPDecoder(Decoder): ...

@register('fbcca')
class FBCCADecoder(Decoder): ...

@register('csp')
class CSPDecoder(Decoder): ...      # MNE CSP + LDA,MI 经典 pipeline

@register('cnn')
class CNNDecoder(Decoder): ...

@register('transformer')
class TransformerDecoder(Decoder): ...        # GPT 风格

@register('transformer_bert')
class TransformerBertDecoder(Decoder): ...    # BERT 风格

Decoder 基类统一 fit/predict/predict_proba/save/load 接口。decode() 顶层函数自动做 StratifiedKFold 交叉验证（SSVEP/FBCCA 除外,它们不需要训练）。

懒加载:registry 里的解码器只保存 (module, cls_name) 引用,真正选中时才 import 模块:

def _lazy(module: str, cls_name: str):
    def factory():
        import importlib
        return getattr(importlib.import_module(module), cls_name)
    return factory

这避免了 import bci.decoder 触发 torch/sklearn/mne 的重量级导入——bci 命令启动 GUI 几乎瞬间。

6. Pipeline:PipelineSession + 增量重执行

class PipelineSession:
    def load_recordings(self, raws: list[mne.io.Raw]) -> 'PipelineSession':
        ...

    def run(self) -> PipelineResult:
        # 4 个步骤:load → preprocess → create_epochs → decode
        # 跳过 _first_invalid 之前的所有步骤
        ...

run() 不是无脑重跑所有步骤,而是从 _first_invalid 开始跑——哪一步的配置改了,就从那一步重跑,前面的步骤复用结果。这对调参迭代非常友好。

一行跑全链路:

from bci import PipelineConfig, PipelineSession, create_default_config
from bci.source import FileSource

raw = FileSource.load("data/S001R04.edf")
session = PipelineSession(create_default_config())
session.load_recordings([raw])
result = session.run(method="transformer")  # 自动 CV,输出 accuracy/std

7. GUI:MVP 架构 + RunState 状态机

PyQt6 最容易犯的错:在主线程做耗时操作,界面冻住。解决方案:所有 I/O 和计算都放 QThread。

但比"放 QThread"更重要的是架构分层——这是事后重构时学到的:

BatchTab（view）:~240 行,只负责建控件、接信号、把请求转发给 presenter
BatchPresenter（controller）:拥有所有状态,RunState 状态机（IDLE / LOADING / LOADED / RUNNING / COMPLETE / ERROR）,把状态变化通过 IBatchView 接口推回 view
IBatchView（ABC）:view 必须实现的接口,presenter 不知道 view 的具体控件,只调接口
BatchWorker（QObject）:在 QThread 中跑 pipeline,进度通过 Qt 信号推回 presenter

sequenceDiagram
    participant V as View
(BatchTab)
    participant P as Presenter
(BatchPresenter)
    participant W as Worker
(BatchWorker)

    V->>P: user click → on_run_clicked(config)
    P->>W: start_run(config)
    Note over W: run pipeline in QThread
    W-->>P: step_completed(step, idx)
    P->>V: view.on_step_done(step, idx)
    W-->>P: decode_completed(PipelineResult)
    P->>V: view.on_done(result)

这样的好处:

测试不需要真的起 QThread——把 worker 工厂注入同步 fake,presenter 用 FakeBatchView,验证状态机转移即可
view 改 UI 不影响 presenter 逻辑——IBatchView 是契约
presenter 改流程不影响 view 控件——presenter 只调接口

显示面板四个:波形、频谱、地形图、预测结果。

下面是加载 PhysioNet EEGBCI 运动想象数据（S001R04.edf, 64 通道 / 160 Hz / 125 秒）后的主界面截图：

主界面一屏展示了 MVP 架构的所有要素：4 步进度条（Main 激活）、加载信息（1 subject / 1 run / 64 ch / 160 Hz / 125s）、文件列表、Preprocess/Epoch/Events/Decode 配置面板，以及底部解码方法选择（默认 LDA）。

点击 Run Pipeline 触发完整流程，跑完后切换到 Decode 步骤看到结果——5-fold CV 准确率柱状图 + 平均线：

状态栏 Done! Accuracy: 0.820 ± 0.157 是当前 pipeline 在 PhysioNet S001R04（21 trial 二分类）上的实跑结果。21 trial 数据量小，结果有噪声（fold 间波动 0.6-1.0）——这是真实结果，不是演示数据。

遇到的坑

EDF 通道名陷阱:EDF 文件的通道名不规范（“EEG Fp1-Ref” vs “Fp1”）,加载后要做通道名清洗才能和 10-20 系统对齐
ICA 是离线操作:ICA 需要完整数据做分解,所以去伪迹只能放在 preprocess 阶段;想做实时去伪迹只能用幅值剔除或回归
懒加载破坏单例:解码器 registry 用字符串路径（"bci.decoder.deep" + "CNNDecoder"）而不是直接 import,避免 torch 重量级依赖污染 GUI 启动——但这意味着不能写"模块级 cache",每次选不同方法都得重新 import,小开销可接受
MVP 重构的代价:BatchTab 早期是 god object（800+ 行混杂 UI + 状态 + 后台逻辑）,拆成 MVP 后单文件缩到 240 行,但需要补 presenter 测试——重构前没测试,重构后才补上。教训:测试不是为了验证重构,是让重构成为可能
懒加载在 PyInstaller 打包时会断:字符串路径 _lazy('bci.decoder.deep', 'CNNDecoder') 在 frozen 环境里找不到模块,需要做 fallback。实际项目还没打包,这只是提前注意

最终效果

# 安装
pip install .

# 启动 GUI
bci

支持 4 种 EEG 格式、6 种解码器（LDA / SSVEP / FBCCA / CSP / CNN / Transformer-GPT + Transformer-BERT）、MVP 架构 PyQt6 离线分析工具,128 个 pytest 测试全过（约 2.6k LOC）。

从零到跑通用 8 周,但最花时间的不是写代码,是两件事:

理清"哪些模块要纯函数、哪些要有状态" —— preprocessor/epocher/decoder 都是纯函数（输入 raw 输出 raw）,状态只在 PipelineSession 和 BatchPresenter 里
学会分层后回看早期代码 —— BatchTab 早期是 god object,第一版 GUI 能跑但代码没法维护;后来重构成 MVP 才真的能用

小结

端到端工具链不是把 MNE/PyTorch/PyQt6 拼起来——是把它们各自的"领域概念"（信号处理、解码模型、用户交互）映射到自己的分层结构里。MNE 提供 raw/epoch 的数据模型,PyTorch 提供模型训练能力,PyQt6 提供控件——但"Pipeline 如何编排、解码器如何注册、状态机如何驱动"是自己的设计。

写完这套工具链,最大的收获不是技术,是意识到"分层"不是写代码前设计的,是写完后回看才发现的。一开始只想把功能跑通;功能跑通后回看,发现 god object 没法维护;重构后分层才清晰。