From 2992bbb0ef02f6e137cbedd1065dff394cfbabc0 Mon Sep 17 00:00:00 2001 From: Tianyu Liu Date: Sat, 22 Aug 2026 18:06:15 +0200 Subject: [PATCH] PRE-M8-T02: add read-only WarmteLink serial probe --- dev-requirements.txt | 2 + requirements.in | 1 + requirements.txt | 2 + scripts/p1_probe.py | 207 ++++++++++++++++++++++++++++++++++++++++- tests/test_p1_probe.py | 160 ++++++++++++++++++++++++++++++- 5 files changed, 366 insertions(+), 6 deletions(-) diff --git a/dev-requirements.txt b/dev-requirements.txt index 245fef7..6c74074 100644 --- a/dev-requirements.txt +++ b/dev-requirements.txt @@ -88,6 +88,8 @@ pyproject-hooks==1.2.0 # via # build # pip-tools +pyserial==3.5 + # via -r requirements.in pytest==8.4.2 # via -r dev-requirements.in python-dotenv==1.2.2 diff --git a/requirements.in b/requirements.in index e3c72fd..f466676 100644 --- a/requirements.in +++ b/requirements.in @@ -7,6 +7,7 @@ paho-mqtt>=2.0,<3.0 pymodbus>=3.6,<4.0 pydantic-settings>=2.6,<3.0 pyotp>=2.9,<3.0 +pyserial>=3.5,<4.0 python-multipart>=0.0.12,<1.0 pyyaml>=6.0,<7.0 sqlalchemy>=2.0,<3.0 diff --git a/requirements.txt b/requirements.txt index 749bcd8..f236d2e 100644 --- a/requirements.txt +++ b/requirements.txt @@ -65,6 +65,8 @@ pymodbus==3.13.1 # via -r requirements.in pyotp==2.10.0 # via -r requirements.in +pyserial==3.5 + # via -r requirements.in python-dotenv==1.2.2 # via # pydantic-settings diff --git a/scripts/p1_probe.py b/scripts/p1_probe.py index 6036851..9763371 100644 --- a/scripts/p1_probe.py +++ b/scripts/p1_probe.py @@ -1,15 +1,18 @@ -"""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. -""" +"""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(?:\([^)]*\))*)$") @@ -189,3 +192,197 @@ def _parse_channels(fields: list[ObisField]) -> tuple[P1Channel, ...]: ) 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()) diff --git a/tests/test_p1_probe.py b/tests/test_p1_probe.py index d7dffcc..df06b5f 100644 --- a/tests/test_p1_probe.py +++ b/tests/test_p1_probe.py @@ -1,7 +1,16 @@ from decimal import Decimal +import io from pathlib import Path -from scripts.p1_probe import IntegrityStatus, TelegramFramer, parse_telegram +import serial + +from scripts.p1_probe import ( + IntegrityStatus, + TelegramFramer, + build_parser, + parse_telegram, + run_probe, +) FIXTURES = Path(__file__).parent / "fixtures" @@ -74,3 +83,152 @@ def test_crc_mismatch_is_invalid_when_standard_framing_is_present(): telegram = parse_telegram(invalid) assert telegram.integrity is IntegrityStatus.INVALID + + +class FakeSerial: + def __init__(self, chunks: list[bytes]) -> None: + self.chunks = iter(chunks) + self.closed = False + + def read(self, _size: int) -> bytes: + return next(self.chunks, b"") + + def close(self) -> None: + self.closed = True + + +def _clock(values: list[float]): + ticks = iter(values) + return lambda: next(ticks, values[-1]) + + +def test_cli_uses_measured_7n1_defaults_and_allows_overrides(): + parser = build_parser() + + defaults = parser.parse_args(["--device", "/dev/serial/by-id/example"]) + overridden = parser.parse_args( + [ + "--device", + "/dev/test", + "--baudrate", + "9600", + "--bytesize", + "8", + "--parity", + "E", + "--stopbits", + "2", + "--duration", + "1", + ] + ) + + assert (defaults.baudrate, defaults.bytesize, defaults.parity, defaults.stopbits) == (115200, 7, "N", 1) + assert (overridden.baudrate, overridden.bytesize, overridden.parity, overridden.stopbits) == ( + 9600, + 8, + "E", + 2, + ) + assert "locally measured" in parser.format_help() + + +def test_probe_reads_fake_chunks_writes_exact_raw_bytes_and_closes_device(tmp_path): + frame = _fixture("warmtelink_p1_7n1.txt") + fake = FakeSerial([frame[:17], frame[17:]]) + raw_path = tmp_path / "capture.bin" + args = build_parser().parse_args( + ["--device", "/dev/serial/by-id/fake", "--duration", "3", "--raw-output", str(raw_path)] + ) + created: dict[str, object] = {} + + def serial_factory(**kwargs): + created.update(kwargs) + return fake + + output = io.StringIO() + result = run_probe( + args, + serial_factory=serial_factory, + clock=_clock([0, 0.1, 0.2, 0.3, 4]), + output=output, + ) + + assert result == 0 + assert fake.closed + assert raw_path.read_bytes() == frame + assert created == { + "port": "/dev/serial/by-id/fake", + "baudrate": 115200, + "bytesize": 7, + "parity": "N", + "stopbits": 1, + "timeout": 1, + } + assert "unverifiable" in output.getvalue() + assert "0-1:24.2.1" in output.getvalue() + + +def test_show_changes_filters_unchanged_fields_and_reports_cadence(): + frame = _fixture("warmtelink_p1_7n1.txt") + changed = frame.replace(b"(5.900*m3)", b"(5.901*m3)") + fake = FakeSerial([frame + changed]) + args = build_parser().parse_args(["--device", "/dev/fake", "--duration", "2", "--show-changes"]) + output = io.StringIO() + + result = run_probe( + args, + serial_factory=lambda **_kwargs: fake, + clock=_clock([0, 0.1, 0.2, 1.2, 3]), + output=output, + ) + + assert result == 0 + assert "cadence=1.0s" in output.getvalue() + assert "changed fields: 1" in output.getvalue() + assert fake.closed + + +def test_probe_diagnoses_permission_errors_without_suggesting_root(): + args = build_parser().parse_args(["--device", "/dev/fake", "--duration", "1"]) + errors = io.StringIO() + + def serial_factory(**_kwargs): + raise serial.SerialException("[Errno 13] Permission denied: '/dev/fake'") + + assert run_probe(args, serial_factory=serial_factory, error_output=errors) == 1 + assert "dialout" in errors.getvalue() + assert "run it as root" in errors.getvalue() + + +def test_probe_diagnoses_busy_or_disconnected_device_and_closes_after_read_error(): + args = build_parser().parse_args(["--device", "/dev/fake", "--duration", "1"]) + errors = io.StringIO() + + def busy_factory(**_kwargs): + raise serial.SerialException("[Errno 16] Device or resource busy: '/dev/fake'") + + assert run_probe(args, serial_factory=busy_factory, error_output=errors) == 1 + assert "Close the program" in errors.getvalue() + + class DisconnectingSerial(FakeSerial): + def read(self, _size: int) -> bytes: + raise serial.SerialException("[Errno 5] device disconnected") + + fake = DisconnectingSerial([]) + errors = io.StringIO() + assert run_probe(args, serial_factory=lambda **_kwargs: fake, error_output=errors) == 1 + assert fake.closed + assert "Check the cable" in errors.getvalue() + + +def test_probe_closes_device_when_interrupted(): + class InterruptingSerial(FakeSerial): + def read(self, _size: int) -> bytes: + raise KeyboardInterrupt + + fake = InterruptingSerial([]) + args = build_parser().parse_args(["--device", "/dev/fake", "--duration", "1"]) + + assert run_probe(args, serial_factory=lambda **_kwargs: fake) == 0 + assert fake.closed