From ffc693e9956529da7efa1c209af272bcc38090fc Mon Sep 17 00:00:00 2001 From: Tianyu Liu Date: Sun, 23 Aug 2026 03:28:03 +0200 Subject: [PATCH] M8-T07: extract reusable P1 parser --- app/integrations/p1.py | 192 +++++++++++++++++++++++++++ docs/design/m8-warmtelink-energy.md | 2 +- scripts/p1_probe.py | 196 ++-------------------------- tests/test_p1_parser.py | 92 +++++++++++++ tests/test_p1_probe.py | 60 ++++++++- 5 files changed, 354 insertions(+), 188 deletions(-) create mode 100644 app/integrations/p1.py create mode 100644 tests/test_p1_parser.py diff --git a/app/integrations/p1.py b/app/integrations/p1.py new file mode 100644 index 0000000..9601a6a --- /dev/null +++ b/app/integrations/p1.py @@ -0,0 +1,192 @@ +"""Pure, privacy-preserving parser for DSMR and WarmteLink P1 telegrams.""" + +from __future__ import annotations + +from dataclasses import dataclass, field as dataclass_field +from decimal import Decimal, InvalidOperation +from enum import StrEnum +import hashlib +import re + + +_OBIS_LINE = re.compile(r"^(?P\d+-\d+:\d+\.\d+\.\d+)(?P(?:\([^)]*\))*)$") +_NUMBER_WITH_UNIT = re.compile(r"^(?P[+-]?\d+(?:\.\d+)?)(?:\*(?P.+))?$") +_CHANNEL_OBIS = re.compile(r"^0-(?P[1-9]\d*):(24|96)\.") +_EQUIPMENT_ID_CODES = re.compile(r"^0-(?:0|[1-9]\d*):96\.1\.[01]$") + + +class IntegrityStatus(StrEnum): + """Whether a frame has a verifiable standard DSMR checksum.""" + + VALID = "valid" + INVALID = "invalid" + UNVERIFIABLE = "unverifiable" + + +class P1ParseError(ValueError): + """A parse error whose message never includes telegram contents.""" + + +@dataclass(frozen=True) +class ObisField: + """One sanitized OBIS line, including values not understood by the parser.""" + + code: str + raw_values: tuple[str, ...] + value: Decimal | None = None + unit: str | None = None + comparison_token: str | None = dataclass_field(default=None, repr=False) + + +@dataclass(frozen=True) +class P1Channel: + """Fields associated with one M-Bus channel, without its raw identifier.""" + + number: int + device_type: str | None + equipment_fingerprint: str | None = dataclass_field(repr=False) + readings: tuple[ObisField, ...] + + +@dataclass(frozen=True) +class P1Telegram: + """A parsed telegram with only sanitized, persistence-safe data.""" + + frame_length: int + integrity: IntegrityStatus + integrity_reason: str + timestamp: str | None + equipment_fingerprint: str | None = dataclass_field(repr=False) + fields: tuple[ObisField, ...] + channels: tuple[P1Channel, ...] + + +class TelegramFramer: + """Incrementally extract newline-terminated variable-length telegrams.""" + + 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 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 + + +def parse_telegram(frame: bytes) -> P1Telegram: + """Parse one complete frame without retaining raw telegram bytes or IDs.""" + + bang = frame.find(b"!") + if bang < 0: + raise P1ParseError("telegram has no footer marker") + body = frame[:bang] + footer = frame[bang + 1 :].rstrip(b"\r\n") + integrity, reason = _integrity(frame, bang, footer) + fields, identifiers = _parse_obis_fields(body) + timestamp = _field_value(fields, "0-0:1.0.0") + return P1Telegram( + frame_length=len(frame), + integrity=integrity, + integrity_reason=reason, + timestamp=timestamp, + equipment_fingerprint=identifiers.get("0-0:96.1.1"), + fields=tuple(fields), + channels=_parse_channels(fields, identifiers), + ) + + +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) -> tuple[list[ObisField], dict[str, str]]: + fields: list[ObisField] = [] + identifiers: dict[str, str] = {} + for line in body.decode("ascii", errors="replace").splitlines()[1:]: + match = _OBIS_LINE.fullmatch(line) + if not match: + continue + code = match.group("code") + raw_values = tuple(re.findall(r"\(([^)]*)\)", match.group("values"))) + if _EQUIPMENT_ID_CODES.fullmatch(code): + comparison_token = _fingerprint(raw_values[-1]) if raw_values else None + if comparison_token is not None: + identifiers[code] = comparison_token + fields.append(ObisField(code, ("",), comparison_token=comparison_token)) + continue + value, unit = _numeric_value(raw_values) + fields.append(ObisField(code, raw_values, value, unit)) + return fields, identifiers + + +def _fingerprint(identifier: str) -> str: + """Hash an identifier locally; callers never receive its original value.""" + + return hashlib.sha256(identifier.encode("ascii", errors="replace")).hexdigest() + + +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], identifiers: dict[str, str]) -> 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_fingerprint=identifiers.get(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()) + ) diff --git a/docs/design/m8-warmtelink-energy.md b/docs/design/m8-warmtelink-energy.md index 7d504fe..8b6910c 100644 --- a/docs/design/m8-warmtelink-energy.md +++ b/docs/design/m8-warmtelink-energy.md @@ -541,7 +541,7 @@ T01～T06 先把现有 DSMR 安全迁到统一 source/binding；T07～T11 再接 ### M8-T07 — 提取可复用 P1 parser（零行为变化） -- **Status**: `todo` +- **Status**: `done` - **Depends**: M8-T02 - **Context**: Pre-M8 parser 已有真机 fixture 证据；先无损提取，避免 worker 与 probe 维护两份协议逻辑。 diff --git a/scripts/p1_probe.py b/scripts/p1_probe.py index 9763371..a562d69 100644 --- a/scripts/p1_probe.py +++ b/scripts/p1_probe.py @@ -3,195 +3,15 @@ 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 +from app.integrations.p1 import IntegrityStatus, ObisField, P1Telegram, TelegramFramer, parse_telegram - -_OBIS_LINE = re.compile(r"^(?P\d+-\d+:\d+\.\d+\.\d+)(?P(?:\([^)]*\))*)$") -_NUMBER_WITH_UNIT = re.compile(r"^(?P[+-]?\d+(?:\.\d+)?)(?:\*(?P.+))?$") -_CHANNEL_OBIS = re.compile(r"^0-(?P[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()) - ) +__all__ = ["IntegrityStatus", "TelegramFramer", "build_parser", "parse_telegram", "run_probe"] def build_parser() -> argparse.ArgumentParser: @@ -279,6 +99,12 @@ def _format_field(field: ObisField) -> str: return f" {field.code}: {raw_values}{value}".rstrip() +def _comparison_values(field: ObisField) -> tuple[str, ...]: + """Return a local change-detection token without exposing identifiers.""" + + return (field.comparison_token,) if field.comparison_token is not None else field.raw_values + + def _print_telegram( telegram: P1Telegram, frame_number: int, @@ -289,14 +115,14 @@ def _print_telegram( ) -> 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}", + f"frame {frame_number}: {telegram.integrity.value}; bytes={telegram.frame_length}; {cadence_text}", file=output, ) print(f" integrity: {telegram.integrity_reason}", file=output) - current_fields = {field.code: field.raw_values for field in telegram.fields} + current_fields = {field.code: _comparison_values(field) 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) + fields = tuple(field for field in fields if previous_fields.get(field.code) != _comparison_values(field)) print(f" changed fields: {len(fields)}", file=output) for field in fields: print(_format_field(field), file=output) diff --git a/tests/test_p1_parser.py b/tests/test_p1_parser.py new file mode 100644 index 0000000..e555aad --- /dev/null +++ b/tests/test_p1_parser.py @@ -0,0 +1,92 @@ +from __future__ import annotations + +from decimal import Decimal +import hashlib +from pathlib import Path + +import pytest + +from app.integrations.p1 import IntegrityStatus, P1ParseError, TelegramFramer, parse_telegram + + +FIXTURES = Path(__file__).parent / "fixtures" + + +def _fixture(name: str) -> bytes: + return (FIXTURES / name).read_bytes() + + +def test_parser_preserves_fixture_quality_and_uses_only_fingerprints_for_identifiers(): + frame = _fixture("warmtelink_p1_7n1.txt") + telegram = parse_telegram(frame) + + assert telegram.integrity is IntegrityStatus.UNVERIFIABLE + assert telegram.frame_length == len(frame) + assert [channel.number for channel in telegram.channels] == [1, 2] + assert telegram.channels[0].readings[0].value == Decimal("5.900") + assert telegram.channels[1].readings[0].unit == "GJ" + assert all(field.raw_values == ("",) for field in telegram.fields if "96.1" in field.code) + assert repr(telegram).find("") >= 0 + assert "KAM" not in repr(telegram) + + +def test_parser_hashes_identifiers_without_returning_them(): + telegram_identifier = "secret-telegram-equipment-42" + channel_identifier = "secret-channel-equipment-42" + telegram = parse_telegram( + f")HEADER\n0-0:96.1.1({telegram_identifier})\n" + f"0-1:96.1.0({channel_identifier})\n!nope\n".encode() + ) + + telegram_fingerprint = hashlib.sha256(telegram_identifier.encode()).hexdigest() + channel_fingerprint = hashlib.sha256(channel_identifier.encode()).hexdigest() + assert telegram.equipment_fingerprint == telegram_fingerprint + assert telegram.channels[0].equipment_fingerprint == channel_fingerprint + rendered = repr(telegram) + assert telegram_identifier not in rendered + assert channel_identifier not in rendered + assert telegram_fingerprint not in rendered + assert channel_fingerprint not in rendered + + +def test_framer_handles_truncation_concatenation_and_variable_lengths_without_channel_mixup(): + first = _fixture("warmtelink_p1_7n1.txt") + second = first.replace(b"(5.900*m3)", b"(12345.678*m3)") + framer = TelegramFramer() + + assert framer.feed(first[:-4]) == [] + frames = framer.feed(first[-4:] + second) + + assert [parse_telegram(frame).channels[0].readings[0].value for frame in frames] == [ + Decimal("5.900"), + Decimal("12345.678"), + ] + + +def test_parser_tolerates_bad_text_without_exposing_it_and_rejects_missing_footer(): + telegram = parse_telegram(b")HEADER\n0-1:24.2.1(1.000*m3)\n\xff\n!not-hex\n") + assert telegram.integrity is IntegrityStatus.UNVERIFIABLE + assert telegram.channels[0].readings[0].value == Decimal("1.000") + + with pytest.raises(P1ParseError, match="footer marker") as exc_info: + parse_telegram(b"secret raw telegram") + assert "secret" not in str(exc_info.value) + + +def test_parser_error_does_not_expose_equipment_identifiers(): + first_identifier = "private-equipment-id-one" + second_identifier = "private-equipment-id-two" + + with pytest.raises(P1ParseError) as exc_info: + parse_telegram( + f")HEADER\n0-0:96.1.1({first_identifier})\n" + f"0-1:96.1.0({second_identifier})\n".encode() + ) + + assert first_identifier not in str(exc_info.value) + assert second_identifier not in str(exc_info.value) + + +def test_parser_keeps_standard_invalid_crc_invalid(): + frame = _fixture("dsmr_p1_valid.txt") + assert parse_telegram(frame[:-5] + b"0000\n").integrity is IntegrityStatus.INVALID diff --git a/tests/test_p1_probe.py b/tests/test_p1_probe.py index df06b5f..3990354 100644 --- a/tests/test_p1_probe.py +++ b/tests/test_p1_probe.py @@ -1,4 +1,5 @@ from decimal import Decimal +import hashlib import io from pathlib import Path @@ -29,7 +30,7 @@ def test_standard_dsmr_fixture_has_a_valid_frame_and_crc(): telegram = parse_telegram(frame) assert telegram.integrity is IntegrityStatus.VALID - assert telegram.header == b"/ISk5\\2MT382-1000" + assert telegram.frame_length == len(frame) assert telegram.timestamp == "240822120000S" @@ -73,7 +74,7 @@ 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" + assert telegram.frame_length == len(b")HEADER\n0-0:1.0.0(240822120000S)\n!zzzz\n") def test_crc_mismatch_is_invalid_when_standard_framing_is_present(): @@ -189,6 +190,61 @@ def test_show_changes_filters_unchanged_fields_and_reports_cadence(): assert fake.closed +def test_probe_stdout_redacts_equipment_identifiers(): + identifier = "private-equipment-id" + frame = ( + f")HEADER\n0-0:96.1.1({identifier})\n0-1:96.1.0({identifier})\n" + "0-1:24.2.1(1.000*m3)\n!nope\n" + ).encode() + fake = FakeSerial([frame]) + args = build_parser().parse_args(["--device", "/dev/fake", "--duration", "1"]) + output = io.StringIO() + + assert run_probe(args, serial_factory=lambda **_kwargs: fake, clock=_clock([0, 0.1, 2]), output=output) == 0 + assert identifier not in output.getvalue() + assert "" in output.getvalue() + + +def test_show_changes_detects_redacted_identifier_changes_without_leaking_them(): + first_telegram_identifier = "private-telegram-id-one" + first_channel_identifier = "private-channel-id-one" + second_telegram_identifier = "private-telegram-id-two" + second_channel_identifier = "private-channel-id-two" + + def frame(telegram_identifier: str, channel_identifier: str) -> bytes: + return ( + f")HEADER\n0-0:96.1.1({telegram_identifier})\n" + f"0-1:96.1.0({channel_identifier})\n" + "0-1:24.2.1(1.000*m3)\n!nope\n" + ).encode() + + first_frame = frame(first_telegram_identifier, first_channel_identifier) + second_frame = frame(second_telegram_identifier, second_channel_identifier) + fake = FakeSerial([first_frame + second_frame]) + args = build_parser().parse_args(["--device", "/dev/fake", "--duration", "2", "--show-changes"]) + output = io.StringIO() + + assert run_probe( + args, + serial_factory=lambda **_kwargs: fake, + clock=_clock([0, 0.1, 0.2, 1.2, 3]), + output=output, + ) == 0 + rendered = output.getvalue() + assert "changed fields: 2" in rendered + assert "" in rendered + identifiers = ( + first_telegram_identifier, + first_channel_identifier, + second_telegram_identifier, + second_channel_identifier, + ) + assert all(identifier not in rendered for identifier in identifiers) + assert all(hashlib.sha256(identifier.encode()).hexdigest() not in rendered for identifier in identifiers) + assert all(identifier not in repr(parse_telegram(first_frame)) for identifier in identifiers[:2]) + assert all(identifier not in repr(parse_telegram(second_frame)) for identifier in identifiers[2:]) + + def test_probe_diagnoses_permission_errors_without_suggesting_root(): args = build_parser().parse_args(["--device", "/dev/fake", "--duration", "1"]) errors = io.StringIO()