"""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\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()) ) 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 "" 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 ''}; " 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())