PRE-M8-T03: finalize WarmteLink hardware validation

dev-requirements.txt

@@ -88,6 +88,8 @@ pyproject-hooks==1.2.0
     # via
     #   build
     #   pip-tools
+pyserial==3.5
+    # via -r requirements.in
 pytest==8.4.2
     # via -r dev-requirements.in
 python-dotenv==1.2.2

docs/design/README.md

@@ -9,8 +9,8 @@
 - [`m5-iot-energy.md`](./m5-iot-energy.md) — IoT 集成与能耗采集（Modbus/Energy + MQTT/HA Discovery + 前端侧边栏）
 - [`m6-tibber-dynamic-energy.md`](./m6-tibber-dynamic-energy.md) — 通用电价层 + DSMR 实时电表接入 + 实时买卖电费计算 + HA Energy 反哺
 - [`m7-meter-epochs-archival.md`](./m7-meter-epochs-archival.md) — 电表生命周期 / 换表归档（Meter epochs）
-- [`pre-m8-warmtelink-p1-poc.md`](./pre-m8-warmtelink-p1-poc.md) — WarmteLink P1 真机概念验证（等待线到货）
-- [`m8-warmtelink-energy.md`](./m8-warmtelink-energy.md) — WarmteLink P1 与多数据源 Meter（Planning 占位）
+- [`pre-m8-warmtelink-p1-poc.md`](./pre-m8-warmtelink-p1-poc.md) — WarmteLink P1 真机概念验证（已完成；正式 CLI 长测与供暖变化均经物理表复核）
+- [`m8-warmtelink-energy.md`](./m8-warmtelink-energy.md) — WarmteLink P1 与多数据源 Meter（Planning 已解锁；架构仍开放）
 
 本文件定义**所有任务共用的格式与协作规则**，各个里程碑文档不再重复这些约定。
 

docs/design/m8-warmtelink-energy.md

@@ -1,13 +1,15 @@
 # M8 — WarmteLink P1 与多数据源 Meter（Planning 占位）
 
-> **状态：Planning 占位；尚未拆解任务卡，尚未锁定架构。** M8 必须等待 [Pre-M8 真机概念验证](./pre-m8-warmtelink-p1-poc.md)完成后再进入正式设计。
+> **状态：Planning 已解锁；尚未拆解任务卡，尚未锁定架构。** [Pre-M8 真机概念验证](./pre-m8-warmtelink-p1-poc.md)
+> 已完成正式 probe、测试和 10 分钟复验；这只满足进入
+> Planning 的证据门，不授权 schema、API、worker 或前端实现。
 
 ## 1. 候选目标
 
 把 Vattenfall WarmteLink 的 P1 数据接入现有 Energy 模块，至少支持：
 
 - 区域供暖累计热量（GJ）。
-- 真机 telegram 确认存在时的生活热水累计量（预计为 m³，最终以实测为准）。
+- 生活热水累计量（真机已确认为 m³）。
 - 历史读数、当前状态以及按需暴露给 Home Assistant。
 - 与现有 Meter epoch/换表归档语义兼容。
 
@@ -18,6 +20,15 @@
 - 当前 electricity Meter 与 `dsmr_reading` 之间没有显式 source FK/binding；电费计算通过代码约定直接查询 DSMR 电力寄存器。
 - `Meter.commodity` 后端已为 `heating` 等品类预留，但“增加 commodity”本身不会自动获得相应数据源或解析能力。
 - 当前 Devices UI/模型是 Modbus 专用，不能直接假设 WarmteLink 应复用 `modbus_device`。
+- 正式 CLI 复验中，一个 WarmteLink serial source 在 60/60 帧均输出两个累计 channel：channel 1
+  为生活热水 `m³`（device type `006`、`5.900 m³`），channel 2 为区域供暖 `GJ`（device type
+  `012`、基线 `0.017 GJ`）；同日与物理表复核一致。最终人工走查开启供暖后，channel 2 又从
+  `0.017 GJ` 增至 `0.018 GJ`，物理热量表同步显示 `0.018 GJ`。
+- 设备 timestamp 严格每 10 秒递进；frame 平均 256 bytes、范围 237–275，不能假定固定帧长。
+  当前 P1 telegram 没有瞬时流量、热功率、供水温度或回水温度字段。
+- 当前线材/设备组合以 `115200 7N1` 才能稳定解析正文。60/60 frame 缺少标准 `/`，故均为
+  `unverifiable`；即使部分 footer 恰为四位十六进制也不能验证 CRC。正式 ingestion 必须显式
+  处理该数据质量状态，不能将数值与表盘相符视为 CRC valid。
 
 ## 3. 下一轮 Planning 必须讨论的问题
 
@@ -28,23 +39,31 @@
 3. “Device”与“Data Source”是否为同一概念；前端 Devices 是否需要改名或分组。
 4. 一个 P1 source 暴露多个 measurement channel 时，如何映射到一个或多个 Meter。
 5. 直接 P1 读数是否使用独立存储，还是将现有 `dsmr_reading` 泛化；如何保证多 source 去重和隔离。
-6. heating GJ 与可选 hot-water m³ 的 commodity、单位、累计/换表语义。
+6. heating GJ 与 hot-water m³ 的 commodity、单位、累计/换表语义。
 7. M8 是否只做采集与展示；区域供暖合同、价格和成本计算是否留到后续里程碑。
 8. 串口 worker 的重连、停止、配置热更新、Docker device mapping 与权限边界。
 
 ## 4. Planning 入口条件
 
-正式编写 M8 目标架构、数据模型和原子任务卡前，至少需要：
+进入 M8 Planning 的证据门已满足：
+
+- [x] Pre-M8 通过并留下脱敏字段清单。
+- [x] 正式 probe 复验两个累计量、单位、channel、device type 与更新时间；equipment identifier
+  保持脱敏。
+- [x] 记录 parser 适配结论和异常 framing/CRC 状态；10 分钟样本的稳定累计值不构成消费更新、
+  reset/wrap 或长期稳定性的证明。
+- [x] 最终人工走查确认供暖消费时累计量按 `0.001 GJ` 更新并与物理表一致；精确更新延迟、
+  reset/wrap 和长期接纳策略仍留给 Planning。
+
+正式编写 M8 目标架构、数据模型和原子任务卡前，仍需：
 
-- Pre-M8 通过并留下脱敏字段清单。
-- 确认实际存在几个累计量及其单位、equipment id/channel 和更新时间。
-- 确认原始 telegram 的稳定性与 parser 适配方式。
 - 重新走查现有 DSMR ingest、Meter epoch、Modbus device、expose/HA 和 Energy 前端边界。
 - 与用户讨论并锁定 Meter ↔ source 的配置体验后，再决定 migration/API/UI 方案。
 
 ## 5. 当前明确不做
 
 - 本占位不创建 implementation task，不授权 schema/API/frontend 变更。
-- 不假设生活热水 m³ 一定可读，也不承诺可拆分“空间供暖 GJ”和“生活热水 GJ”。
+- 不承诺当前 P1 未提供的瞬时流量、热功率或温度，也不承诺可拆分“空间供暖 GJ”和
+  “生活热水 GJ”。
 - 不提前把 WarmteLink 塞进 `modbus_device` 或现有 `dsmr_reading`。
 - 不在缺少真机证据时设计通用 telemetry framework。

docs/design/pre-m8-warmtelink-p1-poc.md

@@ -1,24 +1,28 @@
 # Pre-M8 — WarmteLink P1 真机概念验证
 
-> **状态：等待 USB→P1 线到货后执行。** 本文只定义验证边界与证据要求；当前仓库尚未实现下文所示的 probe 命令。
+> **状态：已完成。** 2026-08-22 已完成临时脚本真机 bring-up；随后仓库内正式 probe 以
+> `115200 7N1` 连续运行 10 分钟，形成脱敏的可重复验收证据；最终人工走查开启供暖后又观察到
+> 区域供暖累计量从 `0.017 GJ` 增至 `0.018 GJ`，并与物理表一致。Pre-M8 现已解除
+> **M8 Planning** 的入口限制；M8 的架构和实现范围仍未锁定。
 
 ## 1. 目的
 
-在进入 M8 正式设计和实现前，先用新家的 Vattenfall WarmteLink 做一次只读真机验证，回答以下问题：
+在进入 M8 正式设计和实现前，用新家的 Vattenfall WarmteLink 建立一条只读 P1 验证链，回答：
 
-1. 当前 USB→P1 线、主机串口权限和 WarmteLink P1 端口能否稳定输出完整 telegram。
-2. telegram 的 framing、CRC、时间戳、OBIS/M-Bus channel 和单位能否被 parser 正确识别。
-3. 实际能够读取哪些累计量：区域供暖热量（GJ）、生活热水体积（m³）或其它字段。
-4. 读数的精度、更新频率和累计语义，是否与热力表/水表面板上的数字一致。
+1. USB→P1 线、主机串口权限和 WarmteLink P1 端口能否稳定输出可解析数据。
+2. 真机使用什么串口参数，telegram 的 framing、CRC、时间戳和 OBIS/M-Bus channel 有何特征。
+3. 实际能够读取哪些累计量，以及单位、精度、更新频率和累计语义。
+4. P1 值是否与热量表/生活热水表面板一致。
 
-Pre-M8 是 M8 的证据门：在真机字段和语义确认前，不决定数据库结构、Meter 数据源绑定、后台采集服务或前端布局。
+Pre-M8 是 M8 的证据门：它只交付真机事实和可重复 probe，不决定数据库结构、Meter 数据源
+绑定、后台采集服务或前端布局。
 
 ## 2. 执行边界
 
 本阶段只建立下面这条最短链路：
 
 ```text
-WarmteLink P1 → USB serial → 完整 telegram → CRC 校验 → 字段解析 → 终端输出
+WarmteLink P1 → USB serial → 原始 telegram → 完整性状态 → 字段解析 → 终端输出
 ```
 
 明确不做：
@@ -28,99 +32,302 @@ WarmteLink P1 → USB serial → 完整 telegram → CRC 校验 → 字段解析
 - 不发布 MQTT / Home Assistant Discovery。
 - 不修改现有 DSMR Reader MQTT、电费计算或 Meter 逻辑。
 - 不在本阶段决定 WarmteLink 应落在哪个正式 Device/Source 模型中。
+- 不写串口、不修改 FTDI EEPROM；probe 必须严格只读。
 
-## 3. 预期操作方式
+## 3. 2026-08-22 真机事实
 
-线到货后，在 workspace 的 virtual environment 中实现并运行一个只读 probe。命令形态暂定为：
+### 3.1 USB、权限与线材
+
+- 新 USB→P1 线是 FTDI FT232R，稳定路径形态为
+  `/dev/serial/by-id/usb-FTDI_FT232R_USB_UART_<redacted>-if00-port0`；正式配置不得依赖
+  `/dev/ttyUSB1`。
+- 运行用户加入 `dialout` 后可以直接读取 tty，无需 root，也不应把容器作为硬件 bring-up
+  的中间层。
+- 已用另一根已知正常的 DSMR 线在电表上读到连续、CRC 正确的 DSMR telegram，证明宿主机
+  串口读取方法本身可用。
+- 曾临时清除新 FTDI 线 EEPROM 的 `INVERT_RXD` 做对照；输出发生变化但仍不可解析，随后已将
+  EEPROM 逐字节恢复为原厂镜像并回读校验。正式方案不得依赖 EEPROM 修改。
+
+### 3.2 串口参数矩阵
+
+真机不是按最初假设的 `115200 8N1` 得到可读正文；当前线材与 WarmteLink 的实测最佳组合为：
+
+```text
+115200 baud, 7 data bits, no parity, 1 stop bit（7N1）
+```
+
+| 参数 | 实测结果 |
+| --- | --- |
+| `115200 7N1` | 正文稳定可读，可枚举 9 个 OBIS 字段 |
+| `115200 7N2` | 同样可读；没有理由增加停止位，正式默认仍用 `7N1` |
+| `115200 8N1/8E1/8O1` | 乱码，无有效 OBIS/CRC |
+| `115200 7E1/7O1` | 乱码，无有效 OBIS/CRC |
+| `120000–3000000`，分别用 `8N1`、`7N1` | 无完整报文；`120000` 仅残留少量可辨识文本 |
+| XON/XOFF 开/关 | 不改变 `7N1` 的字段解析结果 |
+
+[DSMR 5.0.2 P1 Companion Standard](https://www.netbeheernederland.nl/sites/default/files/2024-02/dsmr_5.0.2_p1_companion_standard.pdf)
+规定 `115200 8N1`。因此 `7N1` 是当前设备/线材组合的实测事实，不应被文档或代码包装成
+标准 DSMR framing；CLI 必须允许显式覆盖数据位、校验位和停止位。
+
+### 3.3 Framing 与 CRC 异常
+
+- 临时 bring-up 的一次 8 帧盘点中每帧均为 256 字节；正式 10 分钟采样显示长度并不固定，
+  详见 §3.6。
+- 正文结构稳定，版本、时间戳、两个 M-Bus channel、单位和值均可重复解析。
+- 实测头部为 `)TU)2NWA-MYRSKY`，偶见 `)TU{2NWA-MYRSKY`；没有标准要求的 `/` 起始符。
+- 帧尾存在 `!`，但其后的字符并非每帧都稳定为四位十六进制；即使恰好是四位，也无法从
+  缺失的 `/` 起点完成标准 DSMR CRC16 验证。
+
+因此正式 probe 必须把完整性明确表示为 `valid`、`invalid` 或 `unverifiable`，保留原始字节并
+输出原因。它可以在 `unverifiable` 状态下枚举字段用于 PoC，但绝不能把该帧报告为 CRC 已通过。
+M8 若要持久化这些读数，必须在 Planning 中单独锁定异常帧的接纳、重复确认和告警策略。
+
+### 3.4 实际字段清单
+
+连续 8 帧只出现以下 9 个字段；除 capture timestamp 外，字段集合和值均稳定：
+
+| OBIS | 实测结构/值 | 结论 |
+| --- | --- | --- |
+| `1-3:0.2.8` | `(50)` | DSMR P1 输出版本 5.0 |
+| `0-0:1.0.0` | `(YYMMDDhhmmssX)` | telegram/capture timestamp，每 10 秒变化 |
+| `0-0:96.1.1` | `<redacted>` | WarmteLink/gateway equipment identifier |
+| `0-1:24.1.0` | `(006)` | channel 1，M-Bus device type `0x06`，生活热水 |
+| `0-1:96.1.0` | `<redacted>` | channel 1 equipment identifier；样本制造商可解码为 `KAM` |
+| `0-1:24.2.1` | `(<timestamp>)(5.900*m3)` | 生活热水累计体积，输出到 `0.001 m³` 小数位 |
+| `0-2:24.1.0` | `(012)` | channel 2，M-Bus device type `0x0C`，热量表 |
+| `0-2:96.1.0` | `<redacted>` | channel 2 equipment identifier；样本制造商可解码为 `KAM` |
+| `0-2:24.2.1` | `(<timestamp>)(0.017*GJ)` | 区域供暖累计热量，输出到 `0.001 GJ` 小数位 |
+
+人工面板在同一时间显示 `5.900 m³` 和 `0.017 GJ`，与 P1 值完全一致；用户已确认这两个累计量
+足以作为后续 Home Assistant Energy 展示的数据基础。
+
+没有发现瞬时流量、当前热功率、供水温度或回水温度字段。完整 OMS/M-Bus 模型允许这些可选
+量，但 WarmteLink 当前 P1 telegram 没有导出它们；Pre-M8 和 M8 不得假设它们可用。参考
+[OMS Specification Vol. 2 Annex A](https://oms-group.org/wp-content/uploads/2024/05/OMS-Spec_Vol2_AnnexA_F121.pdf)。
+
+### 3.5 对 M8 已经成立的事实
+
+- 一个 WarmteLink serial source 同时暴露两个独立累计 measurement channel。
+- channel 1 是生活热水 `m³`；channel 2 是区域供暖 `GJ`，不能按论坛样例固定 channel。
+- 两个累计量都与物理表一致，可进入 M8 的 Meter/source 映射讨论。
+- 当前没有瞬时流量、温度或热功率；M8 只承诺累计量采集与展示。
+- 串口 framing 和 CRC 异常尚未消失，必须作为正式 ingestion 的显式质量状态处理。
+
+### 3.6 正式 CLI 10 分钟复验（脱敏）
+
+2026-08-22，仓库内 `python -m scripts.p1_probe` 在真实
+`/dev/serial/by-id/<redacted>` 上以 `115200 7N1` 运行 `--duration 600 --show-changes`，原始
+bytes 仅写入 `/tmp`。CLI 正常以 exit code 0 结束；期间没有 I/O error、未处理异常或断连。
+
+- 读取到 60 个完整 frame、15,362 raw bytes；最后 2 bytes 是下一帧的不完整残片，60 个完整帧
+  合计 15,360 bytes。
+- 设备 timestamp 从 `18:16:10` 至 `18:26:00`，严格每 10 秒递进。CLI 处理的 59 个相邻间隔中，
+  53 个为 10.0 s、1 个为 9.0 s、3 个为 0.0 s、2 个为 20.0 s，均值 9.81 s；0/20 s 配对来自
+  serial chunk 的批量交付，不代表设备 cadence 改变。
+- 完整 frame 长度分布为 237 bytes × 2、239 × 1、254 × 12、256 × 31、258 × 11、275 × 3，
+  平均 256 bytes、范围 237–275。不能再把临时样本的「固定 256 bytes」视为帧格式契约；变长与
+  非标准 footer/分块边界一致。
+- 60/60 的完整性均为 `unverifiable`，因为 60/60 缺少标准 `/` 起始符。footer 长度为 4 × 41、
+  6 × 16、23 × 3；仅 16/60 的 footer 恰为四位十六进制，但仍不能在缺失 `/` 时完成标准 CRC16
+  验证。**没有任何 frame 被报告为 CRC valid。**
+- 60/60 帧均解析到同一组 9 个 OBIS code：`1-3:0.2.8`、`0-0:1.0.0`、`0-0:96.1.1`、
+  `0-1:24.1.0`、`0-1:96.1.0`、`0-1:24.2.1`、`0-2:24.1.0`、`0-2:96.1.0`、`0-2:24.2.1`。
+  equipment identifier 已脱敏，不进入 Git。
+- channel 1 的 device type 是 `006`，累计值稳定为 `5.900 m³`；channel 2 的 device type 是
+  `012`，累计值稳定为 `0.017 GJ`。同日人工表盘复核也显示 `5.900 m³` / `0.017 GJ`，正式 CLI
+  因而复现了该基线。
+- 60 帧中未见瞬时流量、热功率、供水温度或回水温度字段。10 分钟内累计值未变化只能证明这段
+  时间的累计值稳定；它不能证明发生消费时的更新频率，也不能证明 reset 或 wrap 行为。
+
+### 3.7 最终人工走查：供暖累计量变化
+
+正式长测交付后，用户又在终端直接运行只读 probe 十几分钟并开启供暖。channel 2 的区域供暖
+累计量在本次运行中从 `0.017 GJ` 增至 `0.018 GJ`，同一时刻物理热量表也显示 `0.018 GJ`；
+因此可以确认当前 P1 输出会在实际供暖消费下更新累计量，且 `0.001 GJ` 的变化与物理表一致。
+
+这次人工走查没有改变完整性结论：telegram 仍缺少标准 `/`，数值与物理表一致不能替代 CRC
+验证。走查也没有覆盖 reset、wrap 或精确更新延迟；这些仍须由 M8 的接纳、重复确认和告警策略
+处理，而不能从一次累计量递增外推。
+
+## 4. 正式 probe 的预期操作方式
+
+实现后，在 workspace virtual environment 中运行只读 probe：
 
 ```bash
 source .venv/bin/activate
 python -m scripts.p1_probe \
   --device /dev/serial/by-id/<usb-p1-device> \
+  --baudrate 115200 \
+  --bytesize 7 \
+  --parity N \
+  --stopbits 1 \
   --duration 600 \
   --show-changes \
-  --raw-output /tmp/warmtelink-p1-telegram.txt
+  --raw-output /tmp/warmtelink-p1-telegram.bin
 ```
 
-最终参数名可在实现 probe 时调整，但应保留这些能力：
+参数名可在实现时小幅调整，但必须保留这些能力：
 
-- 使用稳定的 `/dev/serial/by-id/...` 路径，而不是依赖可能变化的 `/dev/ttyUSB0`。
-- 连续读取多帧，而不是只看一帧偶然样本。
-- 同时显示完整帧/CRC 结果、原始 OBIS 字段和解析后的值/单位。
-- 枚举 telegram 中出现的所有 M-Bus channel、device type、equipment id、capture timestamp、value 和 unit，不依赖固定字段顺序。
-- 可只显示发生变化的字段，便于观察更新频率。
-- 原始 telegram 默认只写到 `/tmp`；未经脱敏不提交到 Git。
+- 设备路径由用户显式传入，文档推荐 `/dev/serial/by-id/...`。
+- 串口默认采用本机实测 `115200 7N1`，同时允许显式覆盖 framing。
+- 连续读取多帧，输出 telegram cadence、帧长度和读取/重连错误。
+- 同时显示完整性/CRC 状态、原始 OBIS 字段和解析后的 channel、设备类型、值与单位。
+- parser 不依赖字段固定顺序，也不把 GJ/m³ 固定到 channel 1 或 2。
+- `--show-changes` 可只显示发生变化的字段，原始捕获可写入用户指定的 `/tmp` 路径。
+- 捕获文件默认按原始 bytes 保存；未经脱敏不得提交到 Git。
+- permission denied 时给出 `dialout` 指引；不得建议以 root 常驻运行。
 
-### 3.1 分两步 bring-up：Bash 冒烟验证 → Python probe
+Home Assistant Community 的
+[最小 WarmteLink Bash 读取方法](https://community.home-assistant.io/t/solved-dsmr-add-warmtelink-as-data-source/485255/2)
+仍可作为快速可见性检查，但论坛样例的 header、channel 和 device type 与本机均不同，不能作为
+parser 契约。是否引入 [`dsmr_parser`](https://github.com/ndokter/dsmr_parser) 也由 T01 的实测
+兼容性决定；若它要求标准 `/...!CRC` framing，则应保留小型专用 parser，而不是绕过其校验。
 
-Home Assistant Community 的一份 WarmteLink 实例提供了一个适合作为硬件
-bring-up 起点的[最小 Bash 读取方法](https://community.home-assistant.io/t/solved-dsmr-add-warmtelink-as-data-source/485255/2)：
-先把串口设为 115200 baud，逐行读取设备，并从带 `GJ` 的行中取出累计值。该帖展示的
-telegram 样例还给出了以下**候选事实**：
+## 5. 实现任务
 
-- 设备头为 `/NWA-WARMTELINK`，版本字段为 `1-3:0.2.8(50)`。
-- M-Bus channel 1 的 device type 样例为 `004`。
-- 累计热量样例位于 `0-1:24.2.1(<capture timestamp>)(<value>*GJ)`。
-- telegram 以 `!` 加四位 CRC 结束。
+### PRE-M8-T01 — 纯函数 telegram framing、CRC 与 OBIS parser
 
-这些是其他用户在 2022 年记录的单机样本，只用于提出假设，不能替代本机 firmware、线材和
-实际 telegram 的验证。当前 Home Assistant 的
-[DSMR 文档](https://www.home-assistant.io/integrations/dsmr/)确认其 DSMR 集成支持 DSMR v5 与
-M-Bus subdevice；该集成底层使用
-[`dsmr_parser`](https://github.com/ndokter/dsmr_parser)。实现 probe 时可把它作为候选解析基线
-进行对照，但是否引入为本项目正式依赖留到 M8 Planning 决定。
+- **Status**: `done`
+- **Depends**: `none`
+- **Context**: 先把串口 I/O 与解析分开，用脱敏 fixture 固定标准 DSMR 帧和本机异常帧行为。
 
-线到货后的执行顺序调整为：
+**Files**
 
-1. **Bash 冒烟验证**：用稳定的 `/dev/serial/by-id/...` 路径配置串口并短时读取；先保留完整
-   原始字节流，再确认是否能看到 `/NWA-WARMTELINK`、帧尾和带 `GJ` 的行。论坛脚本中的
-   `GJ` 文本提取只能用作快速可见性检查，不能算解析或验收通过。
-2. **Python probe**：在已确认物理链路工作的前提下，实现上面的 `scripts.p1_probe`，完成
-   完整 framing、CRC、全部字段枚举、结构化解析、连续多帧变化观察和人工面板对照。
+- `create  scripts/p1_probe.py`
+- `create  tests/fixtures/dsmr_p1_valid.txt`
+- `create  tests/fixtures/warmtelink_p1_7n1.txt`
+- `create  tests/test_p1_probe.py`
 
-不复制论坛脚本的 MQTT 发布步骤：Pre-M8 仍只输出到终端和 `/tmp`，MQTT / Home Assistant
-集成属于 M8 设计范围。
+**Steps**
 
-## 4. 人工对照
+1. 实现不依赖串口的增量 framing、DSMR CRC16 和通用 OBIS 行解析函数。
+2. 用数据结构表达原始 header/footer、完整性状态、timestamp、channel、device type、equipment id、
+   value 和 unit；数值使用 `Decimal`，不使用二进制浮点保存累计量。
+3. 加入一份完全脱敏的本机结构 fixture，并另造一份 CRC 正确的标准 DSMR fixture。
+4. 对字段重排、分块输入、缺失 `/`、非十六进制 footer、CRC mismatch、未知字段和两个 channel
+   写单元测试。
 
-probe 运行期间，人工从热力表和相关水表面板记录同一时间附近的显示值，并与终端结果对照：
+**Out of scope / 不要碰**
 
-| 检查项 | 需要记录 |
-| --- | --- |
-| 区域供暖 | 面板累计值、P1 值、单位、两者时间差 |
-| 生活热水 | 面板累计值、P1 是否存在对应字段、单位、两者时间差 |
-| 更新时间 | 连续 telegram 中数值变化的间隔 |
-| 累计语义 | 数值是否单调累计，是否出现每日归零或其它重置 |
+- 不打开真实 serial device，不增加依赖，不写数据库/API/MQTT。
+- 不因本机正文可读而伪造 `/` header 或把 CRC 状态升级为 valid。
 
-允许 P1 capture time 与按表时间之间存在合理延迟；不能只凭数值接近就认定字段含义，必须同时核对单位、channel/device type 和时间戳。
+**Acceptance criteria**
 
-## 5. 通过条件
+- [x] 标准 fixture 的 frame boundary 与 CRC 可验证为 `valid`。
+- [x] 脱敏本机 fixture 被标为 `unverifiable`，但能按字段而非位置解析 `m³` 和 `GJ` channel。
+- [x] 任意 chunk boundary 和字段顺序不影响结果，未知字段原样保留。
+- [x] 累计量以 `Decimal` + 原单位返回。
+- [x] `pytest tests/test_p1_probe.py`、`pytest`、`ruff check .` 全绿。
 
-Pre-M8 完成需留下以下证据：
+**Reviewer checklist**
 
-- [ ] 连续收到可识别为 WarmteLink 的完整 telegram。
-- [ ] CRC 校验通过；若失败，已区分串口/线材问题与 parser 问题。
-- [ ] parser 不依赖字段固定顺序，并列出全部实际 channel/OBIS 字段。
-- [ ] 找到 GJ 累计值并与热力表面板对照，误差可由显示精度或 capture 延迟解释。
-- [ ] 明确实际 telegram 是否包含独立的生活热水 m³ 累计量；若包含，已与水表面板对照。
-- [ ] 记录数值精度、telegram 频率、字段更新频率和累计/重置行为。
-- [ ] 形成一份脱敏结果摘要，足以支持下一轮 M8 Planning。
+- CRC 覆盖范围必须严格从 `/` 到 `!`（包含二者），不得对缺失字节做猜测性修补。
+- fixture 必须脱敏且保留足以复现 framing 异常的字节结构。
+- parser 不得硬编码 channel 1=GJ 或 channel 2=m³。
 
-如果只能确认 GJ、没有独立生活热水 m³，这也是有效结论，不视为 Pre-M8 失败。
+### PRE-M8-T02 — 只读 serial probe CLI
 
-## 6. 失败分类
+- **Status**: `done`
+- **Depends**: `PRE-M8-T01`
+- **Context**: 在纯 parser 通过后增加最小 serial I/O，使真机验证可以从仓库稳定复现。
 
-- 完全无数据：优先检查 USB 识别、串口权限、P1 request line、线材方向/供电。
-- 输出乱码或不成帧：优先检查串口参数、信号反相和线材兼容性。
-- 原始帧完整但解析失败：保存脱敏样本，调整 parser/字段映射。
-- 解析成功但面板对不上：检查 capture timestamp、累计语义、单位和 WarmteLink firmware 差异。
+**Files**
+
+- `modify  requirements.in`
+- `modify  requirements.txt`
+- `modify  dev-requirements.txt`
+- `modify  scripts/p1_probe.py`
+- `modify  tests/test_p1_probe.py`
+
+**Steps**
+
+1. 增加受约束的 `pyserial` runtime 依赖并用仓库既有 pip-compile 流程同步生成 requirements。
+2. 实现 `--device`、framing 参数、`--duration`、`--show-changes` 和 `--raw-output`。
+3. 默认使用 `115200 7N1`；串口只读，禁止 write、EEPROM 或自动修改设备配置。
+4. 输出每帧完整性状态、全部字段、值变化、cadence 和错误；SIGINT/超时后关闭串口并正常退出。
+5. 用 fake serial/chunk stream 测试 CLI，不要求 CI 存在 `/dev/ttyUSB*`。
+
+**Out of scope / 不要碰**
+
+- 不做 daemon、自动重连 worker、Docker device mapping、数据库、API、MQTT 或 HA Discovery。
+- 不内置本机 FTDI 序列号，不自动扫描或改写任意 USB 设备。
+
+**Acceptance criteria**
+
+- [x] CLI 可用 `/dev/serial/by-id/...` 读取，且所有 framing 参数都可显式覆盖。
+- [x] 默认参数准确反映本机 `115200 7N1`，帮助文本说明它是实测值而非 DSMR 标准默认。
+- [x] raw output 保留原始 bytes；终端清楚区分 `valid`、`invalid`、`unverifiable`。
+- [x] permission/busy/disconnect 错误非零退出并给出可执行诊断，绝不建议常驻 root。
+- [x] 依赖输入与两个生成 requirements 文件同步。
+- [x] `pytest tests/test_p1_probe.py`、`pytest`、`ruff check .` 全绿。
+
+**Reviewer checklist**
+
+- 确认所有 serial write path 均不存在。
+- 确认测试完全 mock 硬件、没有 CI timing flake，退出路径总会关闭文件描述符。
+- 确认 requirements 是生成结果而非仅手改 lock file。
+
+### PRE-M8-T03 — 正式真机验收与 M8 交接
+
+- **Status**: `done`
+- **Depends**: `PRE-M8-T02`
+- **Context**: 用仓库内 probe 替代本轮临时脚本，形成可重复、脱敏且能支撑 M8 Planning 的证据。
+
+**Files**
+
+- `modify  docs/design/pre-m8-warmtelink-p1-poc.md`
+- `modify  docs/design/m8-warmtelink-energy.md`
+- `modify  docs/design/README.md`
+- `modify  docs/roadmap.md`
+
+**Steps**
+
+1. 在真实 `/dev/serial/by-id/...` 上运行 probe 至少 10 分钟，并保留原始捕获在 `/tmp`。
+2. 汇总帧数、cadence、长度、完整性状态、全部字段和读数变化；不得提交原始设备标识。
+3. 再次与物理表对照 GJ 和 m³，并记录累计/重置行为中本次能证实和不能证实的部分。
+4. 更新本节事实、通过条件和 M8 入口；只有证据齐全后才把 Pre-M8 标记为完成。
+
+**Out of scope / 不要碰**
+
+- 不为完成 checklist 而修补原始字节或放宽 CRC 结果。
+- 不进入 M8 schema/API/worker/frontend 实现。
+
+**Acceptance criteria**
+
+- [x] 仓库内 probe 在真机连续运行至少 10 分钟，无未处理异常退出。
+- [x] 脱敏摘要列出两个累计 channel、单位、精度、cadence 和完整性异常。
+- [x] `0.017 GJ`、`5.900 m³` 的基线或运行时新值与物理表再次对照。
+- [x] 明确没有从当前 P1 输出读取到瞬时流量、功率或温度。
+- [x] Pre-M8 状态与 roadmap/M8 入口同步；代码闸门保持全绿。
+
+**Reviewer checklist**
+
+- 证据必须来自正式 CLI，不得只复述本轮临时脚本结果。
+- 任何 equipment id、FTDI serial 和未脱敏 raw capture 都不得进入 Git。
+- CRC/framing 风险必须原样交给 M8，不能用“数值看起来正确”替代完整性判断。
+
+## 6. 当前通过条件
+
+- [x] USB、tty 权限和稳定 `/dev/serial/by-id/...` 路径已验证。
+- [x] 串口参数矩阵已完成，实测正文可读参数为 `115200 7N1`。
+- [x] 已枚举全部实际 channel/OBIS 字段，并确认没有轮换出现的额外测量量。
+- [x] GJ 和生活热水 m³ 均与物理表面板完全一致。
+- [x] 已记录精度、约 10 秒 telegram cadence 和当前字段集合。
+- [x] framing/CRC 失败已保留为显式异常，没有误报为校验通过。
+- [x] 仓库内 parser、fixtures、probe CLI 和自动化测试完成。
+- [x] 正式 probe 完成至少 10 分钟真机复验并产出脱敏摘要。
+- [x] 最终人工走查在供暖开启后观察到 `0.017 → 0.018 GJ`，并再次与物理表核对一致。
+
+Pre-M8 已完成并向 M8 解锁 Planning；它只交付下列真机事实，不锁定正式架构或实现任务。
 
 ## 7. 向 M8 的交付物
 
-Pre-M8 只向 M8 交付事实，不交付正式架构：
+Pre-M8 完成后只向 M8 交付事实，不交付正式架构：
 
-- 已脱敏的 telegram 结构与字段清单。
-- GJ / 可选 m³ 的实际 channel、OBIS、单位、精度和更新时间。
-- 串口参数、稳定设备路径与部署权限要求。
-- parser 适配结论以及需要保留的异常样本。
-- 对“一个来源包含几个可用计量通道”的实测结论。
+- 脱敏 telegram 结构、fixture 和全部字段清单。
+- GJ 与 m³ 的实际 channel、device type、单位、精度和更新时间。
+- 串口参数、稳定设备路径形态与 `dialout` 权限要求。
+- parser 适配结论和 `unverifiable` framing/CRC 异常样本。
+- “一个 serial source 包含两个独立累计计量 channel”的实测结论。
+- 当前 P1 不提供瞬时流量、热功率或温度的明确边界。

docs/roadmap.md

@@ -2,7 +2,7 @@
 
 本文档记录 `home-automation` 在 `v1.0.3` 之后的下一阶段规划。这一阶段不是小修补，而是几次较大的结构性改动：单库化、前端重写、以及远期的移动端试水。
 
-> 每个里程碑的设计与**可执行原子任务**展开在 [`docs/design/`](./design/README.md)：M1 [`m1-db-consolidation.md`](./design/m1-db-consolidation.md)、M2 [`m2-frontend-v2.md`](./design/m2-frontend-v2.md)、M3 [`m3-token-mobile.md`](./design/m3-token-mobile.md)、M4 [`m4-login-hardening.md`](./design/m4-login-hardening.md)、M5 [`m5-iot-energy.md`](./design/m5-iot-energy.md)、M6 [`m6-tibber-dynamic-energy.md`](./design/m6-tibber-dynamic-energy.md)、M7 [`m7-meter-epochs-archival.md`](./design/m7-meter-epochs-archival.md)、Pre-M8 [`pre-m8-warmtelink-p1-poc.md`](./design/pre-m8-warmtelink-p1-poc.md)、M8 [`m8-warmtelink-energy.md`](./design/m8-warmtelink-energy.md)。Pre-M8/M8 当前仍是验证与 Planning 占位，尚无可派发的实现任务卡。
+> 每个里程碑的设计与**可执行原子任务**展开在 [`docs/design/`](./design/README.md)：M1 [`m1-db-consolidation.md`](./design/m1-db-consolidation.md)、M2 [`m2-frontend-v2.md`](./design/m2-frontend-v2.md)、M3 [`m3-token-mobile.md`](./design/m3-token-mobile.md)、M4 [`m4-login-hardening.md`](./design/m4-login-hardening.md)、M5 [`m5-iot-energy.md`](./design/m5-iot-energy.md)、M6 [`m6-tibber-dynamic-energy.md`](./design/m6-tibber-dynamic-energy.md)、M7 [`m7-meter-epochs-archival.md`](./design/m7-meter-epochs-archival.md)、Pre-M8 [`pre-m8-warmtelink-p1-poc.md`](./design/pre-m8-warmtelink-p1-poc.md)、M8 [`m8-warmtelink-energy.md`](./design/m8-warmtelink-energy.md)。Pre-M8 已完成；M8 仅解锁 Planning，架构仍开放。
 
 ## 当前基线（v1.0.3）
 
@@ -40,8 +40,8 @@
 | **M5** ✅ | IoT / 能耗采集 | 通用 Modbus 采集（YAML profile + JSON readings）+ MQTT/HA Discovery + 前端侧边栏 + Energy 视图 |
 | **M6** ✅ | 通用电价层 + DSMR 接入 + 实时电费计算 | 通用电价层（manual/tibber profile + 合同版本）+ DSMR 实时电表接入 + 每 15min 寄存器差×价计量电费（不可变快照）+ 日/月/年汇总 + 反哺 HA Energy + 前端合同/价格/费用视图 |
 | **M7** ✅ | 电表生命周期 / 换表归档 | 引入 Meter epoch，计费永不跨表算 delta，跨表/无表/异常 delta 一律降级，累计按当前表归零，追溯换表可重算，Meter CRUD API + 前端管理 UI |
-| **Pre-M8** ⏳ | WarmteLink P1 真机概念验证 | USB→P1 到货后用 workspace venv 只读采集实际 telegram，校验 CRC/解析，并与热力表和水表面板对照 |
-| **M8** 📝 | WarmteLink P1 与多数据源 Meter | 等 Pre-M8 后规划直接 P1 采集、区域供暖读数及 Meter 与数据源的可配置关系；当前仅占位 |
+| **Pre-M8** ✅ | WarmteLink P1 真机概念验证 | 正式只读 CLI 长测通过；人工开启供暖后累计量 `0.017 → 0.018 GJ` 且与物理表一致，所有 frame 的 CRC 状态仍为 `unverifiable` |
+| **M8** 📝 | WarmteLink P1 与多数据源 Meter | Pre-M8 已解锁 Planning；再讨论一个 serial source 的两个累计 channel 与 Meter 的可配置关系，尚未锁定架构或实施 |
 | **M3** | 开放与移动端（远期试水） | token 鉴权 + React Native 移动端 |
 
 排序原则：**先清地基，再在干净结构上盖楼。** M2 的新 API 和 React 必须建立在合并后的单库之上；M4 是公网安全加固，在 M5 IoT 集成之前先堵住裸密码这个洞；M5 在安全基座就绪后再做 IoT 接入。
@@ -259,11 +259,22 @@ httpx / paho-mqtt / pyyaml / apscheduler 均为 M5 已有依赖，M6 复用，
 
 ---
 
-## Pre-M8 — WarmteLink P1 真机概念验证（⏳ 等待硬件）
+## Pre-M8 — WarmteLink P1 真机概念验证（✅ 已完成）
 
 ### 目标
 
-USB→P1 线到货后，先在 workspace virtual environment 中运行只读 probe，直接采集 Vattenfall WarmteLink 的实际 telegram。验证完整帧、CRC、OBIS/M-Bus channel、单位、精度和更新频率，并把解析出的 GJ 与热力表面板、可选 m³ 与水表面板进行人工对照。
+2026-08-22 的正式仓库 CLI 以 `115200 7N1` 连续运行 10 分钟，正常退出且没有 I/O error、未处理
+异常或断连。一个 WarmteLink source 在 60/60 完整 frame 中暴露 channel 1 的生活热水累计量
+`5.900 m³` 和 channel 2 的区域供暖累计量 `0.017 GJ`；二者再次与同日物理表一致。没有流量、
+热功率或温度字段。
+
+所有完整 frame 都缺少标准 `/`，CRC 因而为 `unverifiable`；这项异常由正式 probe 原样报告，
+没有伪装成校验通过。frame 平均 256 bytes、范围 237–275，设备 timestamp 严格每 10 秒递进。
+10 分钟内累计值无变化只能证明稳定基线，不能证明消费时更新频率或 reset/wrap 行为。
+
+最终人工走查又在终端连续运行 probe 十几分钟并开启供暖；区域供暖累计量从 `0.017 GJ` 增至
+`0.018 GJ`，物理热量表同步显示 `0.018 GJ`。这确认了实际消费时的累计递增与 `0.001 GJ`
+精度，但不改变 CRC `unverifiable` 结论，也不证明精确更新延迟或 reset/wrap 行为。
 
 本阶段不落库、不接 API/前端/HA，也不决定正式 Device/Source/Meter 关系。它只向 M8 提供脱敏的真机事实，避免在未知 firmware/字段语义上提前设计。
 
@@ -271,13 +282,16 @@ USB→P1 线到货后，先在 workspace virtual environment 中运行只读 pro
 
 ---
 
-## M8 — WarmteLink P1 与多数据源 Meter（📝 Planning 占位）
+## M8 — WarmteLink P1 与多数据源 Meter（📝 Planning 已解锁）
 
 ### 候选目标
 
-在 Pre-M8 事实基础上，把 WarmteLink P1 的区域供暖 GJ、以及真机确认存在时的生活热水 m³ 接入 Energy 模块，并讨论 Meter 如何与实际数据源建立可配置关系。
+在 Pre-M8 事实基础上，把 WarmteLink P1 已确认存在的区域供暖 GJ 和生活热水 m³ 接入
+Energy 模块，并讨论 Meter 如何与实际数据源建立可配置关系。
 
-当前不锁定数据库、API、后台 worker 或 UI 结构；特别是 DSMR MQTT source、P1 serial source、Device/Data Source 的定义和多 channel 映射，都留到下一轮 Planning 讨论后再拆原子任务。
+Pre-M8 的证据门已满足，但当前仍不锁定数据库、API、后台 worker 或 UI 结构；特别是 DSMR MQTT
+source、P1 serial source、Device/Data Source 的定义和多 channel 映射，都留到下一轮 Planning 讨论
+并与用户锁定体验后再拆原子任务。
 
 > Planning 占位与待决问题：[`docs/design/m8-warmtelink-energy.md`](./design/m8-warmtelink-energy.md)
 

requirements.in

@@ -7,6 +7,7 @@ paho-mqtt>=2.0,<3.0
 pymodbus>=3.6,<4.0
 pydantic-settings>=2.6,<3.0
 pyotp>=2.9,<3.0
+pyserial>=3.5,<4.0
 python-multipart>=0.0.12,<1.0
 pyyaml>=6.0,<7.0
 sqlalchemy>=2.0,<3.0

requirements.txt

@@ -65,6 +65,8 @@ pymodbus==3.13.1
     # via -r requirements.in
 pyotp==2.10.0
     # via -r requirements.in
+pyserial==3.5
+    # via -r requirements.in
 python-dotenv==1.2.2
     # via
     #   pydantic-settings

scripts/p1_probe.py

@@ -0,0 +1,388 @@
+"""Read-only parser and command-line probe for DSMR and WarmteLink P1 telegrams."""
+
+from __future__ import annotations
+
+import argparse
+from dataclasses import dataclass
+from decimal import Decimal, InvalidOperation
+from enum import StrEnum
+import errno
+import re
+import sys
+import time
+from typing import BinaryIO, Callable, TextIO
+
+import serial
+
+
+_OBIS_LINE = re.compile(r"^(?P<code>\d+-\d+:\d+\.\d+\.\d+)(?P<values>(?:\([^)]*\))*)$")
+_NUMBER_WITH_UNIT = re.compile(r"^(?P<number>[+-]?\d+(?:\.\d+)?)(?:\*(?P<unit>.+))?$")
+_CHANNEL_OBIS = re.compile(r"^0-(?P<channel>[1-9]\d*):(24|96)\.")
+
+
+class IntegrityStatus(StrEnum):
+    """Whether a frame has a verifiable standard DSMR checksum."""
+
+    VALID = "valid"
+    INVALID = "invalid"
+    UNVERIFIABLE = "unverifiable"
+
+
+@dataclass(frozen=True)
+class ObisField:
+    """One OBIS line, including values not understood by this proof of concept."""
+
+    code: str
+    raw_values: tuple[str, ...]
+    value: Decimal | None = None
+    unit: str | None = None
+
+
+@dataclass(frozen=True)
+class P1Channel:
+    """Fields associated with one M-Bus channel, discovered from its OBIS code."""
+
+    number: int
+    device_type: str | None
+    equipment_id: str | None
+    readings: tuple[ObisField, ...]
+
+
+@dataclass(frozen=True)
+class P1Telegram:
+    """A parsed telegram while retaining its raw framing and all OBIS fields."""
+
+    raw: bytes
+    header: bytes
+    footer: bytes
+    integrity: IntegrityStatus
+    integrity_reason: str
+    timestamp: str | None
+    fields: tuple[ObisField, ...]
+    channels: tuple[P1Channel, ...]
+
+
+def dsmr_crc16(data: bytes) -> int:
+    """Return the DSMR CRC-16 over *data* (normally from ``/`` through ``!``)."""
+
+    crc = 0
+    for byte in data:
+        crc ^= byte
+        for _ in range(8):
+            crc = (crc >> 1) ^ 0xA001 if crc & 1 else crc >> 1
+    return crc & 0xFFFF
+
+
+class TelegramFramer:
+    """Incrementally extract newline-terminated telegrams from byte chunks.
+
+    A standard telegram starts with ``/``.  WarmteLink's observed telegrams do
+    not, so a non-standard frame is retained from the current buffer start until
+    its ``!`` footer line instead of fabricating a standard header.
+    """
+
+    def __init__(self) -> None:
+        self._buffer = bytearray()
+
+    def feed(self, chunk: bytes) -> list[bytes]:
+        """Append *chunk* and return every complete frame now available."""
+
+        self._buffer.extend(chunk)
+        frames: list[bytes] = []
+
+        while (bang := self._buffer.find(b"!")) >= 0:
+            newline = self._buffer.find(b"\n", bang)
+            if newline < 0:
+                break
+
+            standard_start = self._buffer.find(b"/")
+            start = standard_start if 0 <= standard_start < bang else 0
+            frames.append(bytes(self._buffer[start : newline + 1]))
+            del self._buffer[: newline + 1]
+
+        return frames
+
+
+def parse_telegram(frame: bytes) -> P1Telegram:
+    """Parse a complete frame without guessing missing DSMR framing bytes."""
+
+    bang = frame.find(b"!")
+    if bang < 0:
+        raise ValueError("telegram has no footer marker '!'")
+
+    body = frame[:bang]
+    footer = frame[bang + 1 :].rstrip(b"\r\n")
+    header = body.splitlines()[0] if body else b""
+    integrity, reason = _integrity(frame, bang, footer)
+    fields = _parse_obis_fields(body)
+    timestamp = _field_value(fields, "0-0:1.0.0")
+    channels = _parse_channels(fields)
+
+    return P1Telegram(
+        raw=frame,
+        header=header,
+        footer=footer,
+        integrity=integrity,
+        integrity_reason=reason,
+        timestamp=timestamp,
+        fields=tuple(fields),
+        channels=channels,
+    )
+
+
+def _integrity(frame: bytes, bang: int, footer: bytes) -> tuple[IntegrityStatus, str]:
+    if not frame.startswith(b"/"):
+        return IntegrityStatus.UNVERIFIABLE, "missing standard DSMR '/' header"
+    if len(footer) != 4 or not all(chr(byte) in "0123456789abcdefABCDEF" for byte in footer):
+        return IntegrityStatus.UNVERIFIABLE, "footer is not a four-digit hexadecimal CRC"
+
+    expected = int(footer, 16)
+    actual = dsmr_crc16(frame[: bang + 1])
+    if actual == expected:
+        return IntegrityStatus.VALID, "CRC16 verified from '/' through '!'"
+    return IntegrityStatus.INVALID, f"CRC16 mismatch: expected {expected:04X}, calculated {actual:04X}"
+
+
+def _parse_obis_fields(body: bytes) -> list[ObisField]:
+    fields: list[ObisField] = []
+    for line in body.decode("ascii", errors="replace").splitlines()[1:]:
+        match = _OBIS_LINE.fullmatch(line)
+        if not match:
+            continue
+        raw_values = tuple(re.findall(r"\(([^)]*)\)", match.group("values")))
+        value, unit = _numeric_value(raw_values)
+        fields.append(ObisField(match.group("code"), raw_values, value, unit))
+    return fields
+
+
+def _numeric_value(raw_values: tuple[str, ...]) -> tuple[Decimal | None, str | None]:
+    if not raw_values:
+        return None, None
+    match = _NUMBER_WITH_UNIT.fullmatch(raw_values[-1])
+    if not match:
+        return None, None
+    try:
+        return Decimal(match.group("number")), match.group("unit")
+    except InvalidOperation:
+        return None, None
+
+
+def _field_value(fields: list[ObisField], code: str) -> str | None:
+    field = next((item for item in fields if item.code == code), None)
+    return field.raw_values[-1] if field and field.raw_values else None
+
+
+def _parse_channels(fields: list[ObisField]) -> tuple[P1Channel, ...]:
+    by_channel: dict[int, list[ObisField]] = {}
+    for field in fields:
+        match = _CHANNEL_OBIS.match(field.code)
+        if match:
+            by_channel.setdefault(int(match.group("channel")), []).append(field)
+
+    return tuple(
+        P1Channel(
+            number=number,
+            device_type=_field_value(channel_fields, f"0-{number}:24.1.0"),
+            equipment_id=_field_value(channel_fields, f"0-{number}:96.1.0"),
+            readings=tuple(
+                field
+                for field in channel_fields
+                if field.code == f"0-{number}:24.2.1" and field.value is not None
+            ),
+        )
+        for number, channel_fields in sorted(by_channel.items())
+    )
+
+
+def build_parser() -> argparse.ArgumentParser:
+    """Build the CLI parser for an explicitly selected, read-only serial device."""
+
+    parser = argparse.ArgumentParser(
+        description="Read-only WarmteLink P1 serial probe; it never writes to the device.",
+        epilog="Defaults are the locally measured WarmteLink 115200 7N1 framing, not DSMR defaults.",
+    )
+    parser.add_argument("--device", required=True, help="Explicit serial path, preferably /dev/serial/by-id/..."
+    )
+    parser.add_argument(
+        "--baudrate",
+        type=int,
+        default=115200,
+        help="Baud rate (default: 115200, the locally measured WarmteLink value).",
+    )
+    parser.add_argument(
+        "--bytesize",
+        type=int,
+        choices=(5, 6, 7, 8),
+        default=7,
+        help="Data bits (default: 7, locally measured; not the DSMR standard default).",
+    )
+    parser.add_argument(
+        "--parity",
+        choices=("N", "E", "O", "M", "S"),
+        type=lambda value: value.upper(),
+        default="N",
+        help="Parity (default: N, locally measured; not the DSMR standard default).",
+    )
+    parser.add_argument(
+        "--stopbits",
+        type=float,
+        choices=(1, 1.5, 2),
+        default=1,
+        help="Stop bits (default: 1, locally measured; not the DSMR standard default).",
+    )
+    parser.add_argument(
+        "--duration",
+        type=_positive_duration,
+        default=600.0,
+        help="Maximum capture time in seconds (default: 600).",
+    )
+    parser.add_argument(
+        "--show-changes",
+        action="store_true",
+        help="After the first frame, print only OBIS fields whose values changed.",
+    )
+    parser.add_argument(
+        "--raw-output",
+        type=argparse.FileType("wb"),
+        help="Optional path for the exact raw bytes read from the serial device.",
+    )
+    return parser
+
+
+def _positive_duration(value: str) -> float:
+    try:
+        duration = float(value)
+    except ValueError as exc:
+        raise argparse.ArgumentTypeError("duration must be a positive number") from exc
+    if duration <= 0:
+        raise argparse.ArgumentTypeError("duration must be greater than zero")
+    return duration
+
+
+def _serial_error_message(exc: BaseException) -> str:
+    """Return a practical, non-root diagnostic for a serial open/read failure."""
+
+    error_number = getattr(exc, "errno", None)
+    message = str(exc)
+    if error_number == errno.EACCES or "permission denied" in message.lower():
+        return f"serial permission denied: {message}. Add your user to the dialout group; do not run it as root."
+    if error_number == errno.EBUSY or "resource busy" in message.lower():
+        return f"serial device is busy: {message}. Close the program currently using this device and retry."
+    if error_number in {errno.ENODEV, errno.ENOENT, errno.EIO}:
+        return f"serial device disconnected or unavailable: {message}. Check the cable and --device path."
+    return f"serial I/O failed: {message}. Check the cable, device path, and serial framing settings."
+
+
+def _format_field(field: ObisField) -> str:
+    raw_values = ", ".join(field.raw_values) or "<no values>"
+    value = f" parsed={field.value} {field.unit or ''}" if field.value is not None else ""
+    return f"  {field.code}: {raw_values}{value}".rstrip()
+
+
+def _print_telegram(
+    telegram: P1Telegram,
+    frame_number: int,
+    cadence: float | None,
+    previous_fields: dict[str, tuple[str, ...]],
+    show_changes: bool,
+    output: TextIO,
+) -> dict[str, tuple[str, ...]]:
+    cadence_text = "first frame" if cadence is None else f"cadence={cadence:.1f}s"
+    print(
+        f"frame {frame_number}: {telegram.integrity.value}; bytes={len(telegram.raw)}; {cadence_text}",
+        file=output,
+    )
+    print(f"  integrity: {telegram.integrity_reason}", file=output)
+    current_fields = {field.code: field.raw_values for field in telegram.fields}
+    fields = telegram.fields
+    if show_changes and previous_fields:
+        fields = tuple(field for field in fields if previous_fields.get(field.code) != field.raw_values)
+        print(f"  changed fields: {len(fields)}", file=output)
+    for field in fields:
+        print(_format_field(field), file=output)
+    for channel in telegram.channels:
+        print(
+            f"  channel {channel.number}: device_type={channel.device_type or '<unknown>'}; "
+            f"readings={len(channel.readings)}",
+            file=output,
+        )
+    return current_fields
+
+
+SerialFactory = Callable[..., serial.Serial]
+
+
+def run_probe(
+    args: argparse.Namespace,
+    *,
+    serial_factory: SerialFactory = serial.Serial,
+    clock: Callable[[], float] = time.monotonic,
+    output: TextIO = sys.stdout,
+    error_output: TextIO = sys.stderr,
+) -> int:
+    """Capture and report frames until duration elapses or the user interrupts.
+
+    The only operation on ``serial_port`` is ``read``.  It is always closed,
+    including after an interrupt, timeout, or a read error.
+    """
+
+    raw_output: BinaryIO | None = args.raw_output
+    serial_port: serial.Serial | None = None
+    try:
+        serial_port = serial_factory(
+            port=args.device,
+            baudrate=args.baudrate,
+            bytesize=args.bytesize,
+            parity=args.parity,
+            stopbits=args.stopbits,
+            timeout=1,
+        )
+        framer = TelegramFramer()
+        deadline = clock() + args.duration
+        frame_number = 0
+        last_frame_at: float | None = None
+        previous_fields: dict[str, tuple[str, ...]] = {}
+        while clock() < deadline:
+            chunk = serial_port.read(4096)
+            if not chunk:
+                continue
+            if raw_output is not None:
+                raw_output.write(chunk)
+                raw_output.flush()
+            for frame in framer.feed(chunk):
+                now = clock()
+                telegram = parse_telegram(frame)
+                frame_number += 1
+                cadence = None if last_frame_at is None else now - last_frame_at
+                previous_fields = _print_telegram(
+                    telegram,
+                    frame_number,
+                    cadence,
+                    previous_fields,
+                    args.show_changes,
+                    output,
+                )
+                last_frame_at = now
+        return 0
+    except KeyboardInterrupt:
+        print("capture interrupted; serial device closed", file=output)
+        return 0
+    except (serial.SerialException, OSError) as exc:
+        print(_serial_error_message(exc), file=error_output)
+        return 1
+    finally:
+        if serial_port is not None:
+            serial_port.close()
+        if raw_output is not None:
+            raw_output.close()
+
+
+def main(argv: list[str] | None = None) -> int:
+    """Run the command-line probe."""
+
+    args = build_parser().parse_args(argv)
+    return run_probe(args)
+
+
+if __name__ == "__main__":  # pragma: no cover - exercised through main()
+    raise SystemExit(main())

tests/fixtures/dsmr_p1_valid.txt

@@ -0,0 +1,12 @@
+/ISk5\2MT382-1000
+
+1-3:0.2.8(50)
+0-0:1.0.0(240822120000S)
+0-0:96.1.1(TEST-GATEWAY)
+0-1:24.1.0(006)
+0-1:96.1.0(TEST-DHW)
+0-1:24.2.1(240822120000S)(5.900*m3)
+0-2:24.1.0(012)
+0-2:96.1.0(TEST-HEAT)
+0-2:24.2.1(240822120000S)(0.017*GJ)
+!D18A

tests/fixtures/warmtelink_p1_7n1.txt

@@ -0,0 +1,11 @@
+)TU)2NWA-MYRSKY
+0-2:24.2.1(240822120000S)(0.017*GJ)
+0-1:96.1.0(KAM-REDACTED)
+0-0:1.0.0(240822120000S)
+0-2:24.1.0(012)
+0-1:24.2.1(240822120000S)(5.900*m3)
+0-0:96.1.1(WARMTE-REDACTED)
+0-1:24.1.0(006)
+0-2:96.1.0(KAM-REDACTED)
+1-3:0.2.8(50)
+!x7z?

tests/test_p1_probe.py

@@ -0,0 +1,234 @@
+from decimal import Decimal
+import io
+from pathlib import Path
+
+import serial
+
+from scripts.p1_probe import (
+    IntegrityStatus,
+    TelegramFramer,
+    build_parser,
+    parse_telegram,
+    run_probe,
+)
+
+
+FIXTURES = Path(__file__).parent / "fixtures"
+
+
+def _fixture(name: str) -> bytes:
+    return (FIXTURES / name).read_bytes()
+
+
+def test_standard_dsmr_fixture_has_a_valid_frame_and_crc():
+    frame = _fixture("dsmr_p1_valid.txt")
+    framer = TelegramFramer()
+
+    assert framer.feed(frame[:23]) == []
+    assert framer.feed(frame[23:]) == [frame]
+
+    telegram = parse_telegram(frame)
+    assert telegram.integrity is IntegrityStatus.VALID
+    assert telegram.header == b"/ISk5\\2MT382-1000"
+    assert telegram.timestamp == "240822120000S"
+
+
+def test_warmtelink_fixture_is_unverifiable_but_parses_channels_by_obis_code():
+    telegram = parse_telegram(_fixture("warmtelink_p1_7n1.txt"))
+
+    assert telegram.integrity is IntegrityStatus.UNVERIFIABLE
+    assert "missing standard" in telegram.integrity_reason
+    assert [channel.number for channel in telegram.channels] == [1, 2]
+    values = {
+        channel.number: [(field.value, field.unit) for field in channel.readings]
+        for channel in telegram.channels
+    }
+    assert values[1] == [(Decimal("5.900"), "m3")]
+    assert values[2] == [(Decimal("0.017"), "GJ")]
+
+
+def test_arbitrary_chunk_boundaries_and_field_order_do_not_change_parsing():
+    frame = _fixture("warmtelink_p1_7n1.txt")
+    framer = TelegramFramer()
+    frames = []
+    for byte in frame:
+        frames.extend(framer.feed(bytes([byte])))
+
+    assert frames == [frame]
+    assert parse_telegram(frames[0]).channels == parse_telegram(frame).channels
+
+
+def test_unknown_fields_are_preserved_verbatim():
+    frame = b")HEADER\n9-9:9.9.9(opaque)(still-opaque)\n!nope\n"
+
+    telegram = parse_telegram(frame)
+
+    assert len(telegram.fields) == 1
+    assert telegram.fields[0].code == "9-9:9.9.9"
+    assert telegram.fields[0].raw_values == ("opaque", "still-opaque")
+    assert telegram.fields[0].value is None
+
+
+def test_missing_header_and_non_hex_footer_are_unverifiable():
+    telegram = parse_telegram(b")HEADER\n0-0:1.0.0(240822120000S)\n!zzzz\n")
+
+    assert telegram.integrity is IntegrityStatus.UNVERIFIABLE
+    assert telegram.footer == b"zzzz"
+
+
+def test_crc_mismatch_is_invalid_when_standard_framing_is_present():
+    frame = _fixture("dsmr_p1_valid.txt")
+    invalid = frame[:-5] + b"0000\n"
+
+    telegram = parse_telegram(invalid)
+
+    assert telegram.integrity is IntegrityStatus.INVALID
+
+
+class FakeSerial:
+    def __init__(self, chunks: list[bytes]) -> None:
+        self.chunks = iter(chunks)
+        self.closed = False
+
+    def read(self, _size: int) -> bytes:
+        return next(self.chunks, b"")
+
+    def close(self) -> None:
+        self.closed = True
+
+
+def _clock(values: list[float]):
+    ticks = iter(values)
+    return lambda: next(ticks, values[-1])
+
+
+def test_cli_uses_measured_7n1_defaults_and_allows_overrides():
+    parser = build_parser()
+
+    defaults = parser.parse_args(["--device", "/dev/serial/by-id/example"])
+    overridden = parser.parse_args(
+        [
+            "--device",
+            "/dev/test",
+            "--baudrate",
+            "9600",
+            "--bytesize",
+            "8",
+            "--parity",
+            "E",
+            "--stopbits",
+            "2",
+            "--duration",
+            "1",
+        ]
+    )
+
+    assert (defaults.baudrate, defaults.bytesize, defaults.parity, defaults.stopbits) == (115200, 7, "N", 1)
+    assert (overridden.baudrate, overridden.bytesize, overridden.parity, overridden.stopbits) == (
+        9600,
+        8,
+        "E",
+        2,
+    )
+    assert "locally measured" in parser.format_help()
+
+
+def test_probe_reads_fake_chunks_writes_exact_raw_bytes_and_closes_device(tmp_path):
+    frame = _fixture("warmtelink_p1_7n1.txt")
+    fake = FakeSerial([frame[:17], frame[17:]])
+    raw_path = tmp_path / "capture.bin"
+    args = build_parser().parse_args(
+        ["--device", "/dev/serial/by-id/fake", "--duration", "3", "--raw-output", str(raw_path)]
+    )
+    created: dict[str, object] = {}
+
+    def serial_factory(**kwargs):
+        created.update(kwargs)
+        return fake
+
+    output = io.StringIO()
+    result = run_probe(
+        args,
+        serial_factory=serial_factory,
+        clock=_clock([0, 0.1, 0.2, 0.3, 4]),
+        output=output,
+    )
+
+    assert result == 0
+    assert fake.closed
+    assert raw_path.read_bytes() == frame
+    assert created == {
+        "port": "/dev/serial/by-id/fake",
+        "baudrate": 115200,
+        "bytesize": 7,
+        "parity": "N",
+        "stopbits": 1,
+        "timeout": 1,
+    }
+    assert "unverifiable" in output.getvalue()
+    assert "0-1:24.2.1" in output.getvalue()
+
+
+def test_show_changes_filters_unchanged_fields_and_reports_cadence():
+    frame = _fixture("warmtelink_p1_7n1.txt")
+    changed = frame.replace(b"(5.900*m3)", b"(5.901*m3)")
+    fake = FakeSerial([frame + changed])
+    args = build_parser().parse_args(["--device", "/dev/fake", "--duration", "2", "--show-changes"])
+    output = io.StringIO()
+
+    result = run_probe(
+        args,
+        serial_factory=lambda **_kwargs: fake,
+        clock=_clock([0, 0.1, 0.2, 1.2, 3]),
+        output=output,
+    )
+
+    assert result == 0
+    assert "cadence=1.0s" in output.getvalue()
+    assert "changed fields: 1" in output.getvalue()
+    assert fake.closed
+
+
+def test_probe_diagnoses_permission_errors_without_suggesting_root():
+    args = build_parser().parse_args(["--device", "/dev/fake", "--duration", "1"])
+    errors = io.StringIO()
+
+    def serial_factory(**_kwargs):
+        raise serial.SerialException("[Errno 13] Permission denied: '/dev/fake'")
+
+    assert run_probe(args, serial_factory=serial_factory, error_output=errors) == 1
+    assert "dialout" in errors.getvalue()
+    assert "run it as root" in errors.getvalue()
+
+
+def test_probe_diagnoses_busy_or_disconnected_device_and_closes_after_read_error():
+    args = build_parser().parse_args(["--device", "/dev/fake", "--duration", "1"])
+    errors = io.StringIO()
+
+    def busy_factory(**_kwargs):
+        raise serial.SerialException("[Errno 16] Device or resource busy: '/dev/fake'")
+
+    assert run_probe(args, serial_factory=busy_factory, error_output=errors) == 1
+    assert "Close the program" in errors.getvalue()
+
+    class DisconnectingSerial(FakeSerial):
+        def read(self, _size: int) -> bytes:
+            raise serial.SerialException("[Errno 5] device disconnected")
+
+    fake = DisconnectingSerial([])
+    errors = io.StringIO()
+    assert run_probe(args, serial_factory=lambda **_kwargs: fake, error_output=errors) == 1
+    assert fake.closed
+    assert "Check the cable" in errors.getvalue()
+
+
+def test_probe_closes_device_when_interrupted():
+    class InterruptingSerial(FakeSerial):
+        def read(self, _size: int) -> bytes:
+            raise KeyboardInterrupt
+
+    fake = InterruptingSerial([])
+    args = build_parser().parse_args(["--device", "/dev/fake", "--duration", "1"])
+
+    assert run_probe(args, serial_factory=lambda **_kwargs: fake) == 0
+    assert fake.closed