为什么从数据链路开始
做脑机接口，最容易出问题的不是算法，是数据从哪来、怎么来、有没有丢。
很多 BCI 项目直接在算法层埋头——滤波、特征、分类——却对输入数据的质量一无所知。采集设备偶尔丢几包，时间戳不准，换帧顺序乱掉，这些都会被算法默默吞掉，表现为”准确率忽高忽低”。
Omind ONE 的厂家 BLE 头带只有一份含糊的协议说明。为了把数据链路做扎实，只能逆向。这篇文章记录完整的解析过程。
帧布局逆向
厂家的 BLE 通知帧至少有 142 字节。通过抓包和对照厂家单通道示例，逐步确认布局：
通知帧结构（最少 142 字节）
┌────────────┬───────────────────────────┬──────┐
│ 0..1 │ 2..121 (120字节) │ 141 │
│ 头部/元数据 │ EEG 数据区 │ 序号 │
└────────────┴───────────────────────────┴──────┘
│
└── 120 字节拆成 5 组 × 24 字节
每 24 字节 = 8 个有符号大端 24 位槽位
厂家单通道程序发布每组第 0 槽关键常量：
PAYLOAD_OFFSET = 2 # EEG 数据起始
PAYLOAD_LENGTH = 120 # EEG 数据长度
SEQUENCE_OFFSET = 141 # 序号位置
FRAMES_PER_NOTIF = 5 # 每条通知 5 组
SLOTS_PER_FRAME = 8 # 每组 8 个槽位
BYTES_PER_SLOT = 3 # 每槽 24 位
SCALE_UV_PER_COUNT = 0.02235 # µV/count，待校准
SEQUENCE_MODULUS = 128 # 7 位序号解析实现
def parse_vendor_notification(data, scale_uv_per_count=0.02235):
if len(data) < 142:
raise VendorFrameError("notification too short")
# 取 120 字节 EEG 区
payload = bytes(data[2:122])
raw_frames = []
for frame_index in range(5):
start = frame_index * 24
frame_bytes = payload[start:start + 24]
slots = tuple(
int.from_bytes(
frame_bytes[off:off + 3],
byteorder="big", # 大端
signed=True, # 有符号
)
for off in range(0, 24, 3)
)
raw_frames.append(slots)
# 7 位序号
sequence = int(data[141])
if not 0 <= sequence < 128:
raise VendorFrameError("sequence out of 7-bit range")
# 输出每组第 0 槽，转 µV
samples = tuple(frame[0] * scale_uv_per_count for frame in raw_frames)
return sequence, samples, raw_frames要点：
- 24 位槽位用
int.from_bytes(..., signed=True)——有符号大端三字节。EEG 信号是双极性的，转成无符号会导致负半周全部错乱。 - 只取每组第 0 槽——厂家单通道程序明确只发布槽 0，其余 7 个槽位保留/未解释，不视为有效通道。
0.02235暂用厂家示例值——这个 µV/count 缩放系数仍需厂家文档或信号源校准确认，不能想当然。
7 位序号：丢包检测
序号是字节 141，7 位（0-127 回绕）。检测逻辑：
- 连续：
(seq - last_seq) % 128 == 1 - 丢包：
(seq - last_seq) % 128 > 1 - 重复：
seq == last_seq - 乱序：
(seq - last_seq) % 128出现异常跳变后回补
def classify_sequence(seq, last_seq):
if last_seq is None:
return "first"
delta = (seq - last_seq) % 128
if delta == 0:
return "duplicate"
if delta == 1:
return "contiguous"
if delta < 64:
return "gap" # 丢包 (delta-1) 包
return "out_of_order" # 回绕或乱序delta < 64 判定是假定的——超过序号空间一半的跳变视为乱序而非丢包。这个阈值是可配置的，避免把一次回绕误判成丢了一百多个包。
时间重建：主机时钟锚定
设备本身不带可靠时钟（或者没暴露时钟信息）。但采样率是固定的 250Hz，所以可以用主机单调时钟做锚：
- 每条通知包含 5 个样本，理论上每 20ms 一条（250Hz / 5）
- 以第一条通知到达的主机时间为基准，按
t = t0 + n * (1/250)重建每个样本时间 - 实际到达时间有抖动，但样本间相对时间用标称采样周期，避免抖动被当成信号
def reconstruct_timestamps(host_arrival, samples_per_notification, fs=250):
# 以主机到达时刻为锚，样本按标称采样周期均匀分布
base = host_arrival
dt = 1.0 / fs
return [base + i * dt for i in range(samples_per_notification)]这是一个主机锚定的估计，不等价于设备硬件时间戳。文档里明确标注了这一点——不糊弄，不假装是硬件时钟。
缓冲与重放
采集和消费要解耦。线程安全环形缓冲 + 异步 JSONL 记录：
生产者线程: BLE 通知 → parse → ring_buffer.push(样本)
消费线程: ring_buffer.pop() → 实时处理 / recorder.write()- 环形缓冲：容量可配，溢出策略明确（丢最旧，不阻塞采集）
- JSONL 记录：逐条追加，异步落盘，崩溃不丢已写数据
- 离线重放：读回 JSONL，按记录时间戳重放，用于算法复现和回归测试
# 重放
python -m omind_one.cli.replay runtime/sessions/example.jsonl测试与验证
50 个自动化测试覆盖：
- 协议解析：合法帧、短帧、槽位越界、序号越界
- 故障注入：丢包、重复、乱序、回绕、坏包
- 持久化：环形缓冲并发、记录-重放一致性、中断不丢数据
- 断连重连：100 次断连重连，100 个 epoch，状态全部正确
一个典型测试：注入 10 条含 3 个丢包的流，验证序号检测器报出且只报出那 3 个 gap。
未解决的坑
诚实记录还没闭环的部分：
- 缩放系数未确认：
0.02235还是厂家示例值，需要信号发生器或厂家文档做量程校准 - 槽位语义未知：每组 7 个保留槽位到底是什么，需要厂家协议完整文档
- 真实硬件未验证：当前全部测试基于
FakeBleTransport和抓包样本，真实 BLE 连接尚未完成 - Windows BLE 环境：本机 MSYS2 Python 不是受控 BLE 环境，需要官方 CPython + WinRT 才能跑真头带
做数据链路的一些体会
做这一层数据基础设施，最大的收获不是会解析 BLE 帧了，而是重新理解了”可靠”的定义。
可靠不是”代码能跑”，是”丢一包你知道丢了、乱一帧你知道乱了、时间戳你知道它只是个估计”。把这些不确定性显式地暴露出来，而不是让算法默默吞掉，才是做严肃工程的第一步。
下一增量：真实 BLE transport、设备服务诊断、刺激器与在线分类整合。数据的地基打好了，上面的楼才敢盖。