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 <noreply@anthropic.com>
This commit is contained in:
2026-11-10 10:58:05 +00:00
co-authored by Claude Opus 5.5
parent 56d635f051
commit 392095d549
5 changed files with 172 additions and 8 deletions
+13 -7
View File
@@ -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