home-automation/scripts/p1_probe.py

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