GrigoandClaude Opus 5.5
392095d549
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>
91 lines
2.9 KiB
Python
91 lines
2.9 KiB
Python
#!/usr/bin/env python3
|
|
"""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
|
|
empty results: an empty list would look like "everyone disconnected".
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import re
|
|
from typing import Dict, Optional, Tuple
|
|
|
|
log = logging.getLogger("collector.bluez_adv")
|
|
|
|
dbus = None
|
|
try:
|
|
import dbus as _dbus
|
|
|
|
dbus = _dbus
|
|
except Exception: # pragma: no cover
|
|
dbus = None
|
|
|
|
BLUEZ = "org.bluez"
|
|
OBJECT_MANAGER = "org.freedesktop.DBus.ObjectManager"
|
|
DEVICE_IFACE = "org.bluez.Device1"
|
|
|
|
# Per-call D-Bus timeout (python3-dbus default is 25 s).
|
|
CALL_TIMEOUT_S = 5.0
|
|
|
|
|
|
def system_bus():
|
|
if dbus is None:
|
|
raise RuntimeError("python3-dbus required")
|
|
return dbus.SystemBus()
|
|
|
|
|
|
def _managed_objects(bus) -> dict:
|
|
# introspect=False: no extra Introspect round-trip (it has its own 25 s timeout).
|
|
obj = bus.get_object(BLUEZ, "/", introspect=False)
|
|
mgr = dbus.Interface(obj, OBJECT_MANAGER)
|
|
return mgr.GetManagedObjects(timeout=CALL_TIMEOUT_S)
|
|
|
|
|
|
def bluez_owner(bus) -> Optional[str]:
|
|
"""Unique bus name of the running bluetoothd, or None if it is not running.
|
|
|
|
Asks dbus-daemon only, so it does not depend on bluetoothd being responsive.
|
|
"""
|
|
try:
|
|
return str(bus.get_name_owner(BLUEZ))
|
|
except dbus.exceptions.DBusException as e:
|
|
if e.get_dbus_name() == "org.freedesktop.DBus.Error.NameHasNoOwner":
|
|
return None
|
|
raise
|
|
|
|
|
|
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()
|
|
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
|
|
|
|
|
|
def find_device_path(bus, addr: str) -> Optional[str]:
|
|
addr_u = (addr or "").upper()
|
|
for path, ifaces in _managed_objects(bus).items():
|
|
dev = ifaces.get(DEVICE_IFACE)
|
|
if dev and str(dev.get("Address", "")).upper() == addr_u:
|
|
return str(path)
|
|
return None
|
|
|
|
|
|
def disconnect_device(bus, addr: str) -> None:
|
|
path = find_device_path(bus, addr)
|
|
if not path:
|
|
return
|
|
obj = bus.get_object(BLUEZ, path, introspect=False)
|
|
dbus.Interface(obj, DEVICE_IFACE).Disconnect(timeout=CALL_TIMEOUT_S)
|