GrigoandClaude Opus 5.5
115879ab67
- every BlueZ call runs off the asyncio thread with a timeout; if bluetoothd stays unresponsive for 30 s the collector restarts bluetooth.service and exits for a clean re-registration - Type=notify + WatchdogSec for buttontask-collector, never give up restarting - bluetooth.service drop-in: Restart=always (BlueZ 5.55 crashes on assert/SEGV), installed by device-install.sh and OTA (bt-update) - bt-service: bluez-restart action for the collector Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
85 lines
2.6 KiB
Python
85 lines
2.6 KiB
Python
#!/usr/bin/env python3
|
|
"""BlueZ helpers for connected devices (RSSI / 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
|
|
from typing import Dict, Optional
|
|
|
|
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, Optional[int]]:
|
|
"""{ADDR: rssi or None} for all connected devices. Raises on D-Bus errors."""
|
|
out: Dict[str, Optional[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
|
|
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)
|