ViezeVingertjes/watchpat-one

# watchpat-one 通过 Bluetooth LE 捕获并解码来自 Itamar/ZOLL WatchPAT ONE 的实时传感器数据。 WatchPAT ONE 是一款一次性家庭睡眠呼吸暂停记录仪。您佩戴它一晚，结果会发送给您的医生，然后就可以将其丢弃。其内部包含一个 nRF52832、128 Mbit 的 flash、一个 LIS3DH 加速度计和一个 TI AFE4404 光学前端。 该库负责连接设备、启动数据采集，并将每个通道以原始字节和 CSV 格式写入磁盘。它不进行任何分析。它的定位是作为一个供您在此基础上进行构建的基础工具。 | 通道 | 速率 | |---|---| | 血氧仪 A（红光），原始 PPG | 100 Hz | | 血氧仪 B（红外），原始 PPG | 100 Hz | | PAT（外周动脉张力） | 100 Hz | | 胸部运动（呼吸） | 100 Hz | | 加速度计 X/Y/Z | 5 Hz | | 胸部震动，两个通道 | 5 Hz | | 指标（有符号 32 位） | 1 Hz | ## 安装与运行 需要 Python 3.10+ 以及一个 BLE 适配器。 ``` git clone https://github.com/ViezeVingertjes/watchpat-one cd watchpat-one python3 -m venv .venv && .venv/bin/pip install bleak .venv/bin/python -m watchpat capture --minutes 10 ``` ``` session: sessions/20260802-104746 scanning ... connected to DE:07:24:FB:45:F4 -> IS_DEVICE_PAIRED -> START_SESSION -> START_ACQUISITION device: Itamar Medical Watch-PATOne fw 04.02.1234 recording for 10 min. Ctrl-C to stop early. 42 packets 4242 waveform 210 motion 24.1 KiB ``` 检查会话： ``` .venv/bin/python -m watchpat decode sessions/20260802-104746 ``` ## 您需要了解的设备行为 **请务必仅使用已耗尽的设备。** 启动采集会写入设备的 flash。请先完成您的睡眠监测。以 `N` 结尾的 BLE 名称（例如 `ITAMAR_087F0A87N`）表示该设备从未进行过配对。 **该设备仅在开机后的一段时间窗口内接受指令。** 当该窗口关闭后，它仍然会广播，仍然会接受 BLE 连接，仍然会响应普通的 GATT 读取，但会忽略所有指令。软重置无济于事，因为重置指令本身也会被忽略。请拔下电池并重新插入。如果采集程序报告“no DATA received”，原因几乎总是如此。 **已使用过的设备电池很可能已经没电。** 该电池的电量设计仅够维持一晚。当电压低于约 1.29 V 时，开机自检会失败，固件会停止为应用层提供服务，其表现与上述问题完全相同。在此状态下，设备的事件日志会显示 `BAT test Failed`。更换一节全新的 1.5 V 电池即可解决此问题，而且无论如何，您也需要通过接触电池来进行断电重启。 **全芯片擦除需要几分钟时间。** 如果您发送 `CLEAR_DATA`，设备会确认该指令，随后在 flash 擦除完毕之前将停止响应。 ## 输出 ``` sessions/20260802-104746/ ├── raw.bin every byte the device sent, unparsed ├── waveforms.csv t_s, channel, rate_hz, value ├── motion.csv t_s, field_a, field_b, accel_x, accel_y, accel_z, crc_ok ├── events.csv t_s, kind, detail └── session.json device identity, counts, channel summary ``` `raw.bin` 会在字节到达时立即写入，因此程序崩溃或连接断开不会造成任何数据丢失。CSV 文件是从原始文件派生出来的，可以随时重新生成： ``` .venv/bin/python -m watchpat decode sessions/20260802-104746/raw.bin ``` `t_s` 从采集开始时计数。波形样本是通过根据记录的采样率，从其数据包的到达时间向前反推来定位的，因此第一个数据包中的样本会带有微小的负时间戳。 ## 库的使用 ``` from watchpat import Reassembler, decode_data_payload, record_values, stitch packets = Reassembler().feed(open("raw.bin", "rb").read()) pat_records = [] for pkt in packets: frames, motion = decode_data_payload(pkt.payload) for f in frames: if f.kind == "PAT": pat_records.append(record_values(f)) for m in motion: print(m.accel_x, m.accel_y, m.accel_z, m.magnitude, m.crc_ok) pat = stitch(pat_records) ``` ## 协议 使用 Nordic UART Service。所有数据均以 20 字节的数据块进行传输，间隔 10 ms，即使协商了更大的 MTU 也是如此。 | UUID | 角色 | |---|---| | `6e400001-b5a3-f393-e0a9-e50e24dcca9e` | 服务 | | `6e400002-…` | TX，主机写入，无响应写入 | | `6e400003-…` | RX，设备通知 | ### 数据包头，24 字节，混合字节序 ``` off size field encoding 0 2 magic 0xBBBB big-endian 2 2 command id big-endian (0x2300 -> bytes 23 00) 4 8 timestamp big-endian, unix seconds (0 is accepted) 12 4 transaction id little-endian 16 2 length little-endian, header + payload 18 2 flags little-endian 20 2 zero always 0 22 2 crc little-endian, CRC-16/CCITT-FALSE ``` CRC 的多项式为 `0x1021`，初始值为 `0xFFFF`，无反射，无 xorout，校验范围为将 CRC 字段置零后的整个数据包。 事务 ID 在多次重启间必须保持唯一。设备会记住它曾见过的 ID，并通过 `ACK status 3 (NON_UNIQUE_ID)` 拒绝重复的 ID。 ACK 的 payload 为 `[被确认的命令 id: u16be][状态: u8][2 个保留字节]`，其中状态值代表 `0 OK`、`1 CRC_ERROR`、`2 ILLEGAL_OPCODE`、`3 NON_UNIQUE_ID`、`4 INVALID_PARAM`。 ### DATA payload `[记录数: u8][2 个保留字节]`，随后是一系列记录。每条记录的头部从其同步字开始算起为 12 字节： ``` off size field 0 u16 0xAAAA sync 2 u8 record id 3 u8 record type 4 u16le payload length, in BYTES 6 u16le sample rate, Hz 8 u32le flags 12 ... payload[payload_len] ``` 请使用 length 字段来遍历记录。以 `0xAAAA` 进行拆分是行不通的，因为该字节对也会出现在波形数据中。 | id/类型 | 速率 | 通道 | Payload | |---|---|---|---| | `01/11` | 100 Hz | 血氧仪 A（红光） | seed + zigzag deltas | | `02/11` | 100 Hz | 血氧仪 B（红外） | seed + zigzag deltas | | `03/11` | 100 Hz | PAT | seed + zigzag deltas | | `04/01` | 100 Hz | 胸部运动 | seed + nibble deltas | | `05/10` | 1 Hz | 指标 | `s32le` | | `06/00` | 5 Hz | 运动 | 5 × 16 字节带 CRC 的子帧 | | `0C/00` | 1 Hz | 事件代码 | `u16le` | | `0D/00` | 1 Hz | 事件 payload | 原始字节 | ### 波形编解码器 两者均以一个 2 字节的有符号 seed 开始，随后是增量编码，在 100 Hz 下每条记录大约可产生一秒钟的样本数据。 ``` # Codec A, zigzag byte deltas (01/11, 02/11, 03/11) acc = struct.unpack_from("> 1) ^ -(b & 1) # 0,1,2,3,4 -> 0,-1,+1,-2,+2 out.append(acc) ``` 编解码器 B（胸部运动，`04/01`）在 seed 之后跳过一个字节，然后将每个字节的低位作为一个 4 位有符号增量（delta）读取。高位用于标记重复的样本。 每条记录都携带一个绝对 seed，该 seed 不会延续上一条记录的值，因为设备会在记录之间应用其自身的 AGC。天真地拼接记录会产生锯齿状波形。 `stitch()` 通过将记录的第一个样本与上一条记录的最后一个样本对齐，从而消除每个边界处的阶跃。它不会消除记录*内部*的趋势，而且每条记录通常会产生几百个计数的漂移，因此经过长时间拼接的数据序列会累积这种漂移，并逐渐偏离真实的基线。对于持续时间超过几秒的分析，请按单条记录处理或对结果进行高通滤波。 ### 运动子帧，16 字节，每条记录包含五个 ``` 0 u32 0x57A3DDDD marker 4 u16 field_a chest vibration 6 u16 field_b chest vibration 8 i16 accel x 10 i16 accel y 12 i16 accel z 14 u16 CRC-16/CCITT-FALSE over bytes 0..13 ``` 加速度计的计数大约为每 g 1000。 `field_a` 和 `field_b` 是来自胸部传感器的震动和冲击能量。它们在 1010 左右达到饱和，既不响应空气传播的声音，也不响应呼吸。它们未校准至任何物理单位，厂商也未提供相关文档。 该 1 Hz 指标的含义尚未明确。它会在几十秒内发生漂移，具体方向取决于会话过程，并且对物理刺激没有任何反应。 ## 测试 ``` .venv/bin/python -m tests ``` 无需测试框架。这些检查是针对作为 `tests/data/sample_capture.bin` 提交的真实采集数据运行的，该数据是在设备放置在桌面上被敲击且未佩戴任何东西的情况下记录的，因此不包含任何生理或个人数据。 它们涵盖了数据包成帧和 CRC、考虑到每个 payload 字节的记录遍历、运动子帧 CRC、针对重力的加速度计幅度、两种波形编解码器以及记录拼接。 ## 未包含的内容 SpO2 和脉率不会被传输。设备发送原始光学波形，并在外部设备上推导出这些数值。利用本库提供的数据（基于红光/红外光的比值法、峰值检测）来计算这些指标，是一个合理的进阶开发方向。 ## 硬件 | 部件 | 功能 | |---|---| | Nordic nRF52832 | Cortex-M4 SoC + BLE，PCB 上有 SWD 焊盘 | | Micron MT25QL128ABA | 128 Mbit 串行 NOR flash | | ST LIS3DH | 3 轴加速度计 | | TI AFE4404 | 光学前端 | | 1 × 1.5 V 原电池 | 不可充电，但一旦打开外壳就很容易更换 | FCC ID [2APUBWPONE](https://fccid.io/2APUBWPONE)。基于固件版本 04.02.1234 进行了测试。 如果您打开了外壳，nRF52832 的 SWD 焊盘将是比电池寿命更持久的接入途径。请参阅 [atc1441/ESP32_nRF52_SWD](https://github.com/atc1441/ESP32_nRF52_SWD)。 ## 非医疗器械 本项目解码医疗器械的输出结果，其本身并不是医疗器械。此处的内容均未经过与参考多导睡眠图的对比验证，胸部震动字段未经校准，且由于 seed 的符号特性，偶尔会导致记录的 DC 偏移发生改变。 基于 MIT 许可协议。

标签：Python, 云资产清单, 传感器数据, 医疗设备, 无后门, 物联网, 蓝牙低功耗, 逆向工具, 逆向工程