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>
105 lines
3.7 KiB
Python
105 lines
3.7 KiB
Python
#!/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("<HHH", AF_BLUETOOTH, HCI_DEV_NONE, HCI_CHANNEL_CONTROL)
|
|
if _libc.bind(fd, ctypes.c_char_p(sa), len(sa)) < 0:
|
|
err = ctypes.get_errno()
|
|
os.close(fd)
|
|
raise OSError(err, os.strerror(err))
|
|
return fd
|
|
|
|
|
|
def _bdaddr(addr: str) -> 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("<HHH", MGMT_OP_GET_CONN_INFO, index, len(params)) + params
|
|
deadline = time.monotonic() + timeout
|
|
fd = _open()
|
|
try:
|
|
os.write(fd, cmd)
|
|
while True:
|
|
left = deadline - time.monotonic()
|
|
if left <= 0:
|
|
raise TimeoutError("mgmt get-conn-info timed out")
|
|
ready, _, _ = select.select([fd], [], [], left)
|
|
if not ready:
|
|
continue
|
|
pkt = os.read(fd, 512)
|
|
if len(pkt) < 9:
|
|
continue
|
|
ev, ev_index, _plen = struct.unpack_from("<HHH", pkt, 0)
|
|
opcode, status = struct.unpack_from("<HB", pkt, 6)
|
|
# The control channel also carries unrelated events: skip them.
|
|
if ev not in (MGMT_EV_CMD_COMPLETE, MGMT_EV_CMD_STATUS):
|
|
continue
|
|
if opcode != MGMT_OP_GET_CONN_INFO or ev_index != index:
|
|
continue
|
|
if status == MGMT_STATUS_PERMISSION_DENIED:
|
|
raise MgmtPermissionError("mgmt: permission denied (need CAP_NET_ADMIN)")
|
|
if status != 0 or ev != MGMT_EV_CMD_COMPLETE or len(pkt) < 9 + 10:
|
|
return None # not connected / unsupported
|
|
# return params: bdaddr(6) type(1) rssi(s8) tx(s8) max_tx(s8)
|
|
(rssi,) = struct.unpack_from("<b", pkt, 9 + 7)
|
|
return None if rssi == RSSI_INVALID else int(rssi)
|
|
finally:
|
|
os.close(fd)
|