122 lines
4.1 KiB
Python
122 lines
4.1 KiB
Python
"""Drive wmbusmeters as a subprocess and stream parsed telegram events."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from typing import Callable, Dict, Optional
|
|
|
|
from .parser import parse_line
|
|
|
|
EventCallback = Callable[[dict], None]
|
|
|
|
|
|
class WMBusSource:
|
|
"""Runs one wmbusmeters process and feeds parsed events to a callback.
|
|
|
|
The command is::
|
|
|
|
wmbusmeters --format=json --verbose [--listento=MODES] DEVICE \
|
|
[m_<id> auto <id> <key> ...] \
|
|
scan auto '*' NOKEY
|
|
|
|
The trailing wildcard meter makes wmbusmeters emit a verbose DLL line for
|
|
*every* telegram (universal discovery); keyed meters additionally emit
|
|
decrypted JSON. Changing keys requires relaunching the process (:meth:`restart`).
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
device: str = "rtlwmbus",
|
|
modes: str = "c1,t1",
|
|
wmbusmeters: str = "wmbusmeters",
|
|
stdin_data: Optional[bytes] = None,
|
|
) -> None:
|
|
self.device = device
|
|
self.modes = modes
|
|
self.wmbusmeters = wmbusmeters
|
|
self.stdin_data = stdin_data
|
|
self._keys: Dict[str, str] = {}
|
|
self._proc: Optional[asyncio.subprocess.Process] = None
|
|
self._task: Optional[asyncio.Task] = None
|
|
self._on_event: Optional[EventCallback] = None
|
|
self.running = False
|
|
|
|
def set_keys(self, keys: Dict[str, str]) -> None:
|
|
self._keys = dict(keys)
|
|
|
|
def build_cmd(self) -> list[str]:
|
|
cmd = [self.wmbusmeters, "--format=json", "--verbose"]
|
|
if self.device.startswith("rtl"):
|
|
cmd.append(f"--listento={self.modes}")
|
|
cmd.append(self.device)
|
|
for device_id, key in self._keys.items():
|
|
cmd += [f"m_{device_id}", "auto", device_id, key]
|
|
cmd += ["scan", "auto", "*", "NOKEY"]
|
|
return cmd
|
|
|
|
async def start(self, on_event: EventCallback) -> None:
|
|
self._on_event = on_event
|
|
await self._spawn()
|
|
|
|
async def _spawn(self) -> None:
|
|
cmd = self.build_cmd()
|
|
use_stdin = self.stdin_data is not None
|
|
self._proc = await asyncio.create_subprocess_exec(
|
|
*cmd,
|
|
stdin=asyncio.subprocess.PIPE if use_stdin else asyncio.subprocess.DEVNULL,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.STDOUT,
|
|
)
|
|
self.running = True
|
|
if use_stdin and self._proc.stdin is not None:
|
|
self._proc.stdin.write(self.stdin_data) # type: ignore[arg-type]
|
|
await self._proc.stdin.drain()
|
|
self._proc.stdin.close()
|
|
self._task = asyncio.create_task(self._read_loop())
|
|
|
|
async def _read_loop(self) -> None:
|
|
assert self._proc is not None and self._proc.stdout is not None
|
|
try:
|
|
while True:
|
|
raw = await self._proc.stdout.readline()
|
|
if not raw:
|
|
break
|
|
line = raw.decode("utf-8", "replace").rstrip("\n")
|
|
if not self._on_event:
|
|
continue
|
|
ev = parse_line(line)
|
|
if ev is not None:
|
|
self._on_event(ev)
|
|
except asyncio.CancelledError: # pragma: no cover - shutdown path
|
|
pass
|
|
|
|
async def _kill(self) -> None:
|
|
if self._task is not None:
|
|
self._task.cancel()
|
|
try:
|
|
await self._task
|
|
except (asyncio.CancelledError, Exception):
|
|
pass
|
|
self._task = None
|
|
if self._proc is not None and self._proc.returncode is None:
|
|
try:
|
|
self._proc.terminate()
|
|
await asyncio.wait_for(self._proc.wait(), timeout=3)
|
|
except (asyncio.TimeoutError, ProcessLookupError):
|
|
try:
|
|
self._proc.kill()
|
|
except ProcessLookupError:
|
|
pass
|
|
self._proc = None
|
|
self.running = False
|
|
|
|
async def restart(self) -> None:
|
|
await self._kill()
|
|
# a one-shot stdin replay has been consumed; don't re-feed on restart
|
|
if self.stdin_data is not None:
|
|
self.stdin_data = None
|
|
await self._spawn()
|
|
|
|
async def stop(self) -> None:
|
|
await self._kill()
|