85 lines
2.3 KiB
Python
85 lines
2.3 KiB
Python
#!/usr/bin/env python3
|
|
"""BlueZ helpers for connected devices (RSSI / disconnect). Advertising is via bless."""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import 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"
|
|
PROPS = "org.freedesktop.DBus.Properties"
|
|
OBJECT_MANAGER = "org.freedesktop.DBus.ObjectManager"
|
|
|
|
|
|
def system_bus():
|
|
if dbus is None:
|
|
raise RuntimeError("python3-dbus required")
|
|
return dbus.SystemBus()
|
|
|
|
|
|
def read_device_rssi(bus, device_path: str) -> Optional[int]:
|
|
if dbus is None:
|
|
return None
|
|
try:
|
|
obj = bus.get_object(BLUEZ, device_path)
|
|
props = dbus.Interface(obj, PROPS)
|
|
return int(props.Get("org.bluez.Device1", "RSSI"))
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def find_device_path(bus, addr: str) -> Optional[str]:
|
|
if dbus is None:
|
|
return None
|
|
addr_u = (addr or "").upper()
|
|
try:
|
|
obj = bus.get_object(BLUEZ, "/")
|
|
mgr = dbus.Interface(obj, OBJECT_MANAGER)
|
|
for path, ifaces in mgr.GetManagedObjects().items():
|
|
dev = ifaces.get("org.bluez.Device1")
|
|
if not dev:
|
|
continue
|
|
if str(dev.get("Address", "")).upper() == addr_u:
|
|
return str(path)
|
|
except Exception as e:
|
|
log.debug("find_device: %s", e)
|
|
return None
|
|
|
|
|
|
def disconnect_device(bus, addr: str) -> None:
|
|
path = find_device_path(bus, addr)
|
|
if not path:
|
|
return
|
|
try:
|
|
obj = bus.get_object(BLUEZ, path)
|
|
dbus.Interface(obj, "org.bluez.Device1").Disconnect()
|
|
except Exception as e:
|
|
log.warning("disconnect %s: %s", addr, e)
|
|
|
|
|
|
def list_connected_addresses(bus) -> list:
|
|
if dbus is None:
|
|
return []
|
|
out = []
|
|
try:
|
|
obj = bus.get_object(BLUEZ, "/")
|
|
mgr = dbus.Interface(obj, OBJECT_MANAGER)
|
|
for _path, ifaces in mgr.GetManagedObjects().items():
|
|
dev = ifaces.get("org.bluez.Device1")
|
|
if not dev:
|
|
continue
|
|
if bool(dev.get("Connected", False)):
|
|
out.append(str(dev.get("Address", "")).upper())
|
|
except Exception as e:
|
|
log.debug("list_connected: %s", e)
|
|
return out
|