"""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()) )