From 392095d54992fbb471336400e826dfcc5668492e Mon Sep 17 00:00:00 2001 From: Grigo Date: Tue, 10 Nov 2026 10:58:05 +0000 Subject: [PATCH] collector: RSSI of connected sensors via kernel mgmt API BlueZ fills Device1.RSSI only while scanning; sensors connect to us, so the field stayed empty. MGMT_OP_GET_CONN_INFO (like btmgmt conn-info) returns the link RSSI; polled per connected sensor and after each write. Needs CAP_NET_ADMIN (AmbientCapabilities in the collector unit). Co-Authored-By: Claude Opus 5.5 --- collector/bluez_adv.py | 20 +++-- collector/config.py | 2 + collector/main.py | 51 +++++++++++- collector/mgmt.py | 104 +++++++++++++++++++++++++ webconfig/buttontask-collector.service | 3 + 5 files changed, 172 insertions(+), 8 deletions(-) create mode 100644 collector/mgmt.py diff --git a/collector/bluez_adv.py b/collector/bluez_adv.py index fb2afd5..773f4f1 100644 --- a/collector/bluez_adv.py +++ b/collector/bluez_adv.py @@ -1,5 +1,5 @@ #!/usr/bin/env python3 -"""BlueZ helpers for connected devices (RSSI / disconnect). Advertising is via bless. +"""BlueZ helpers for connected devices (list / disconnect). Advertising is via bless. All calls here are blocking (python3-dbus). main.py runs them in a worker thread with a timeout, so they must raise on failure instead of returning @@ -8,7 +8,8 @@ empty results: an empty list would look like "everyone disconnected". from __future__ import annotations import logging -from typing import Dict, Optional +import re +from typing import Dict, Optional, Tuple log = logging.getLogger("collector.bluez_adv") @@ -54,16 +55,21 @@ def bluez_owner(bus) -> Optional[str]: raise -def connected_devices(bus) -> Dict[str, Optional[int]]: - """{ADDR: rssi or None} for all connected devices. Raises on D-Bus errors.""" - out: Dict[str, Optional[int]] = {} +def connected_devices(bus) -> Dict[str, Tuple[str, int]]: + """{ADDR: (address type, controller index)} for all connected devices. + + Device1.RSSI is not used: BlueZ sets it only while scanning, and sensors + connect to us. RSSI comes from mgmt.conn_rssi(), which needs these two. + Raises on D-Bus errors.""" + out: Dict[str, Tuple[str, int]] = {} for _path, ifaces in _managed_objects(bus).items(): dev = ifaces.get(DEVICE_IFACE) if not dev or not bool(dev.get("Connected", False)): continue addr = str(dev.get("Address", "")).upper() - rssi = dev.get("RSSI") - out[addr] = int(rssi) if rssi is not None else None + addr_type = str(dev.get("AddressType", "random")) + m = re.search(r"/hci(\d+)$", str(dev.get("Adapter", ""))) + out[addr] = (addr_type, int(m.group(1)) if m else 0) return out diff --git a/collector/config.py b/collector/config.py index a3f6c17..df60b5c 100644 --- a/collector/config.py +++ b/collector/config.py @@ -7,6 +7,8 @@ N = 4 DEFAULT_REPORT_INTERVAL_MIN = 30 STALE_PERIODS = 3 # k: silence for k*period → stale; another k → lost TICK_INTERVAL_S = 0.5 # connect poll + flush cadence +RSSI_PERIOD_S = 2.0 # link RSSI refresh while a sensor is connected +RSSI_TIMEOUT_S = 3.0 # one mgmt get-conn-info call ADV_DEBOUNCE_S = 1.0 UI_LOG_LIMIT = 200 diff --git a/collector/main.py b/collector/main.py index 5793c5e..fdde174 100644 --- a/collector/main.py +++ b/collector/main.py @@ -48,6 +48,8 @@ from config import ( # noqa: E402 BT_SERVICE_HELPER, LOCAL_NAME, N, + RSSI_PERIOD_S, + RSSI_TIMEOUT_S, TICK_INTERVAL_S, collector_log_path, registry_path, @@ -55,6 +57,7 @@ from config import ( # noqa: E402 ) from gatt import CollectorGatt, build_bless_server # noqa: E402 from history import History # noqa: E402 +from mgmt import MgmtPermissionError, conn_rssi # noqa: E402 from protocol import ( # noqa: E402 CFG_JOIN_OPEN, CFG_POOL_FULL, @@ -163,6 +166,11 @@ class CollectorApp: self._bluez_owner: Optional[str] = None self._bluez_fail_since: Optional[float] = None self._last_owner_check = 0.0 + # RSSI of connected sensors via the kernel mgmt API (see mgmt.py). + self._peer: Dict[str, tuple] = {} # addr -> (addr_type, hci index) + self._rssi_ts: Dict[str, float] = {} # addr -> last read (monotonic) + self._rssi_off = False # no CAP_NET_ADMIN: stop trying + self._loop: Optional[asyncio.AbstractEventLoop] = None self.gatt.on_write = self._on_write self.gatt.on_read_slot = self._on_read_slot @@ -296,6 +304,7 @@ class CollectorApp: dist=report.dist, ) self._mark_dirty(registry=True, status=True, adv=True) + self._rssi_after_write(addr) log.debug( "report slot=%s addr=%s batt=%s dist=%s", sensor.slot if sensor else "?", @@ -398,8 +407,9 @@ class CollectorApp: return self._bluez_ok() connected = set(devices) + self._peer = dict(devices) for addr in connected - self._known_connected: - ok = self._on_connect(addr, devices.get(addr)) + ok = self._on_connect(addr) if not ok: try: await self._bluez(disconnect_device, addr) @@ -408,11 +418,49 @@ class CollectorApp: for addr in self._known_connected - connected: self._on_disconnect(addr) self._known_connected = connected + for addr in connected: + await self._refresh_rssi(addr) if connected and self.gatt.current_addr() is None: # Prefer the sole connected peer as current writer context. if len(connected) == 1: self.gatt.set_current_addr(next(iter(connected))) + async def _refresh_rssi(self, addr: str, force: bool = False) -> None: + """Read the link RSSI of a connected sensor (at most every 2 s).""" + if self._rssi_off or self._loop is None: + return + now = time.monotonic() + if not force and now - self._rssi_ts.get(addr, 0.0) < RSSI_PERIOD_S: + return + self._rssi_ts[addr] = now + addr_type, index = self._peer.get(addr, ("random", 0)) + try: + rssi = await asyncio.wait_for( + self._loop.run_in_executor(None, conn_rssi, addr, addr_type, index), + RSSI_TIMEOUT_S, + ) + except MgmtPermissionError: + self._rssi_off = True + log.warning("RSSI unavailable: collector needs CAP_NET_ADMIN " + "(AmbientCapabilities in buttontask-collector.service)") + return + except Exception as e: + log.debug("rssi %s: %s", addr, _err(e)) + return + if rssi is None: + return + sensor = self.registry.get(addr) + if sensor is not None and sensor.rssi != rssi: + sensor.rssi = rssi + self._mark_dirty(status=True) + + def _rssi_after_write(self, addr: str) -> None: + # Right after a report the link is still up: read RSSI now, not on the + # next poll (a sensor may disconnect before it). + if self._loop is not None: + self._loop.call_soon_threadsafe( + lambda: asyncio.ensure_future(self._refresh_rssi(addr, force=True))) + def _handle_cmd(self) -> None: cmd = read_cmd() if not cmd: @@ -478,6 +526,7 @@ class CollectorApp: self.bus = system_bus() loop = asyncio.get_running_loop() + self._loop = loop # Owner before registration: if bluetoothd restarts later, the owner # changes and we know our GATT app / advertisement are gone. self._bluez_owner = await self._bluez(bluez_owner) diff --git a/collector/mgmt.py b/collector/mgmt.py new file mode 100644 index 0000000..5212264 --- /dev/null +++ b/collector/mgmt.py @@ -0,0 +1,104 @@ +#!/usr/bin/env python3 +"""RSSI of a connected LE peer via the kernel Bluetooth Management API. + +BlueZ fills Device1.RSSI only while *scanning*; our sensors connect to us +(we are the GATT server), so that property stays empty. The controller still +measures the link: MGMT_OP_GET_CONN_INFO returns it (same as +`btmgmt conn-info`). + +Needs CAP_NET_ADMIN (mgmt sockets without it are "untrusted" and this command +is refused): buttontask-collector.service sets AmbientCapabilities. +Blocking; main.py calls it in an executor with a timeout. +""" +from __future__ import annotations + +import ctypes +import os +import select +import socket +import struct +import time +from typing import Optional + +AF_BLUETOOTH = 31 +BTPROTO_HCI = 1 +HCI_DEV_NONE = 0xFFFF +HCI_CHANNEL_CONTROL = 3 + +MGMT_OP_GET_CONN_INFO = 0x0031 +MGMT_EV_CMD_COMPLETE = 0x0001 +MGMT_EV_CMD_STATUS = 0x0002 +MGMT_STATUS_NOT_CONNECTED = 0x02 +MGMT_STATUS_PERMISSION_DENIED = 0x14 + +ADDR_TYPES = {"bredr": 0, "public": 1, "random": 2} +RSSI_INVALID = 127 + +_libc = ctypes.CDLL(None, use_errno=True) + + +class MgmtPermissionError(PermissionError): + """The collector lacks CAP_NET_ADMIN.""" + + +def _open() -> int: + # Python's bind() for BTPROTO_HCI cannot select a channel, hence libc. + # Stays a raw fd: socket.socket(fileno=...) calls getsockname(), which the + # kernel refuses (EBADFD) on the control channel. + fd = _libc.socket(AF_BLUETOOTH, socket.SOCK_RAW | socket.SOCK_CLOEXEC, BTPROTO_HCI) + if fd < 0: + err = ctypes.get_errno() + raise OSError(err, os.strerror(err)) + sa = struct.pack(" bytes: + parts = addr.split(":") + if len(parts) != 6: + raise ValueError(f"bad address {addr!r}") + return bytes(int(p, 16) for p in reversed(parts)) + + +def conn_rssi(addr: str, addr_type: str = "random", index: int = 0, + timeout: float = 1.5) -> Optional[int]: + """RSSI (dBm) of the connected peer, or None if not connected / unknown. + + Raises MgmtPermissionError without CAP_NET_ADMIN, OSError on socket errors. + """ + params = _bdaddr(addr) + bytes([ADDR_TYPES.get(addr_type, 2)]) + cmd = struct.pack("