From 22faeb45bb65e913537118d2cc1262fc9698796f Mon Sep 17 00:00:00 2001 From: Tianyu Liu Date: Sat, 22 Aug 2026 17:47:30 +0200 Subject: [PATCH] PRE-M8-T01: add WarmteLink P1 telegram parser --- scripts/p1_probe.py | 191 +++++++++++++++++++++++++++ tests/fixtures/dsmr_p1_valid.txt | 12 ++ tests/fixtures/warmtelink_p1_7n1.txt | 11 ++ tests/test_p1_probe.py | 76 +++++++++++ 4 files changed, 290 insertions(+) create mode 100644 scripts/p1_probe.py create mode 100644 tests/fixtures/dsmr_p1_valid.txt create mode 100644 tests/fixtures/warmtelink_p1_7n1.txt create mode 100644 tests/test_p1_probe.py diff --git a/scripts/p1_probe.py b/scripts/p1_probe.py new file mode 100644 index 0000000..6036851 --- /dev/null +++ b/scripts/p1_probe.py @@ -0,0 +1,191 @@ +"""Pure parsing helpers for DSMR and WarmteLink P1 telegrams. + +Serial I/O deliberately belongs to the follow-up probe task. This module only +turns byte chunks into frames and parses the resulting telegrams. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from decimal import Decimal, InvalidOperation +from enum import StrEnum +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)\.") + + +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()) + ) diff --git a/tests/fixtures/dsmr_p1_valid.txt b/tests/fixtures/dsmr_p1_valid.txt new file mode 100644 index 0000000..9e9824b --- /dev/null +++ b/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 diff --git a/tests/fixtures/warmtelink_p1_7n1.txt b/tests/fixtures/warmtelink_p1_7n1.txt new file mode 100644 index 0000000..320a762 --- /dev/null +++ b/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? diff --git a/tests/test_p1_probe.py b/tests/test_p1_probe.py new file mode 100644 index 0000000..d7dffcc --- /dev/null +++ b/tests/test_p1_probe.py @@ -0,0 +1,76 @@ +from decimal import Decimal +from pathlib import Path + +from scripts.p1_probe import IntegrityStatus, TelegramFramer, parse_telegram + + +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