Get connection info from D-Bus API

There are device properties for RSSI and Tx power now and link quality is pointless.

blueman/gui/manager/ManagerDeviceList.py

@@ -1,5 +1,5 @@
 from gettext import gettext as _
-from typing import Optional, TYPE_CHECKING, List, Any, cast, Callable, Set
+from typing import Optional, TYPE_CHECKING, List, Any, cast, Callable
 import html
 import logging
 import cairo
@@ -16,12 +16,10 @@
 from blueman.Sdp import ServiceUUID, OBEX_OBJPUSH_SVCLASS_ID, BATTERY_SERVICE_SVCLASS_ID
 from blueman.gui.GtkAnimation import TreeRowFade, CellFade, AnimBase
 from blueman.main.Config import Config
-from _blueman import ConnInfoReadError, conn_info
 
 import gi
 gi.require_version("Gtk", "3.0")
 from gi.repository import Gtk
-from gi.repository import GLib
 from gi.repository import Gdk
 from gi.repository import GdkPixbuf
 from gi.repository import Pango
@@ -47,9 +45,6 @@ def __init__(self, adapter: Optional[str] = None, inst: Optional["Blueman"] = No
             {"id": "rssi_pb", "type": GdkPixbuf.Pixbuf, "renderer": Gtk.CellRendererPixbuf(),
              "render_attrs": {}, "view_props": {"spacing": 0},
              "celldata_func": (self._set_cell_data, "rssi")},
-            {"id": "lq_pb", "type": GdkPixbuf.Pixbuf, "renderer": Gtk.CellRendererPixbuf(),
-             "render_attrs": {}, "view_props": {"spacing": 0},
-             "celldata_func": (self._set_cell_data, "lq")},
             {"id": "tpl_pb", "type": GdkPixbuf.Pixbuf, "renderer": Gtk.CellRendererPixbuf(),
              "render_attrs": {}, "view_props": {"spacing": 0},
              "celldata_func": (self._set_cell_data, "tpl")},
@@ -60,7 +55,6 @@ def __init__(self, adapter: Optional[str] = None, inst: Optional["Blueman"] = No
             {"id": "objpush", "type": bool},  # used to set Send File button
             {"id": "battery", "type": float},
             {"id": "rssi", "type": float},
-            {"id": "lq", "type": float},
             {"id": "tpl", "type": float},
             {"id": "icon_info", "type": Gtk.IconInfo},
             {"id": "cell_fader", "type": CellFade},
@@ -73,8 +67,6 @@ def __init__(self, adapter: Optional[str] = None, inst: Optional["Blueman"] = No
         self.props.has_tooltip = True
         self.Blueman = inst
 
-        self._monitored_devices: Set[str] = set()
-
         self.Config = Config("org.blueman.general")
         self.Config.connect('changed', self._on_settings_changed)
         # Set the correct sorting
@@ -326,48 +318,6 @@ def row_setup_event(self, tree_iter: Gtk.TreeIter, device: Device) -> None:
         except Exception as e:
             logging.exception(e)
 
-        if device["Connected"]:
-            self._monitor_power_levels(tree_iter, device)
-
-    def _monitor_power_levels(self, tree_iter: Gtk.TreeIter, device: Device) -> None:
-        if device["Address"] in self._monitored_devices:
-            return
-
-        assert self.Adapter is not None
-        cinfo = conn_info(device["Address"], os.path.basename(self.Adapter.get_object_path()))
-        try:
-            cinfo.init()
-        except ConnInfoReadError:
-            logging.warning("Failed to get power levels, probably a LE device.")
-
-        model = self.get_model()
-        assert isinstance(model, Gtk.TreeModel)
-        r = Gtk.TreeRowReference.new(model, model.get_path(tree_iter))
-        self._update_power_levels(tree_iter, device, cinfo)
-        GLib.timeout_add(1000, self._check_power_levels, r, cinfo, device["Address"])
-        self._monitored_devices.add(device["Address"])
-
-    def _check_power_levels(self, row_ref: Gtk.TreeRowReference, cinfo: conn_info, address: str) -> bool:
-        if not row_ref.valid():
-            logging.warning("stopping monitor (row does not exist)")
-            cinfo.deinit()
-            self._monitored_devices.remove(address)
-            return False
-
-        tree_iter = self.get_iter(row_ref.get_path())
-        assert tree_iter is not None
-
-        device = self.get(tree_iter, "device")["device"]
-
-        if device["Connected"]:
-            self._update_power_levels(tree_iter, device, cinfo)
-            return True
-        else:
-            cinfo.deinit()
-            self._disable_power_levels(tree_iter)
-            self._monitored_devices.remove(address)
-            return False
-
     def row_update_event(self, tree_iter: Gtk.TreeIter, key: str, value: Any) -> None:
         logging.info(f"{key} {value}")
 
@@ -392,62 +342,50 @@ def row_update_event(self, tree_iter: Gtk.TreeIter, key: str, value: Any) -> Non
             device = self.get(tree_iter, "device")["device"]
             has_objpush = self._has_objpush(device)
             self.set(tree_iter, objpush=has_objpush)
+            self._check_battery(tree_iter, device)
 
         elif key == "Connected":
             self.set(tree_iter, connected=value)
 
-            if value:
-                self._monitor_power_levels(tree_iter, self.get(tree_iter, "device")["device"])
-            else:
+            if not value:
                 self._disable_power_levels(tree_iter)
 
-    def _update_power_levels(self, tree_iter: Gtk.TreeIter, device: Device, cinfo: conn_info) -> None:
-        row = self.get(tree_iter, "cell_fader", "battery", "rssi", "lq", "tpl")
+        elif key == "RSSI":
+            self._update_bar(tree_iter, "rssi", 50 if value is None else max(50 + float(value) / 127 * 50, 10))
 
-        bars = {}
+        elif key == "TxPower":
+            self._update_bar(tree_iter, "tpl", 0 if value is None else max(float(value) / 127 * 100, 10))
 
+        elif key == "ServicesResolved":
+            self._check_battery(tree_iter, self.get(tree_iter, "device")["device"])
+
+    def _check_battery(self, tree_iter: Gtk.TreeIter, device: Device) -> None:
         if device["ServicesResolved"] and any(ServiceUUID(uuid).short_uuid == BATTERY_SERVICE_SVCLASS_ID
                                               for uuid in device["UUIDs"]):
-            bars["battery"] = Battery(obj_path=device.get_object_path())["Percentage"]
+            self._update_bar(tree_iter, "battery", Battery(obj_path=device.get_object_path())["Percentage"])
 
-        # cinfo init may fail for bluetooth devices version 4 and up
-        # FIXME Workaround is horrible and we should show something better
-        if cinfo.failed:
-            if not bars:
-                bars = {"rssi": 100.0, "tpl": 100.0, "lq": 100.0}
-        else:
-            try:
-                bars["rssi"] = max(50 + float(cinfo.get_rssi()) / 127 * 50, 10)
-            except ConnInfoReadError:
-                bars["rssi"] = 50
-            try:
-                bars["lq"] = max(float(cinfo.get_lq()) / 255 * 100, 10)
-            except ConnInfoReadError:
-                bars["lq"] = 10
-            try:
-                bars["tpl"] = max(50 + float(cinfo.get_tpl()) / 127 * 50, 10)
-            except ConnInfoReadError:
-                bars["tpl"] = 50
-
-        if row["battery"] == row["rssi"] == row["tpl"] == row["lq"] == 0:
+    def _update_bar(self, tree_iter: Gtk.TreeIter, name: str, perc: float) -> None:
+        row = self.get(tree_iter, "cell_fader", "battery", "rssi", "tpl")
+        row[name] = perc
+
+        if row["battery"] == row["rssi"] == row["tpl"] == 0:
             self._prepare_fader(row["cell_fader"]).animate(start=0.0, end=1.0, duration=400)
 
         w = 14 * self.get_scale_factor()
         h = 48 * self.get_scale_factor()
 
-        for (name, perc) in bars.items():
-            if round(row[name], -1) != round(perc, -1):
-                icon_name = f"blueman-{name}-{int(round(perc, -1))}.png"
-                icon = GdkPixbuf.Pixbuf.new_from_file_at_scale(os.path.join(PIXMAP_PATH, icon_name), w, h, True)
-                self.set(tree_iter, **{name: perc, f"{name}_pb": icon})
+        if round(row[name], -1) != round(perc, -1):
+            icon_name = f"blueman-{name}-{int(round(perc, -1))}.png"
+            icon = GdkPixbuf.Pixbuf.new_from_file_at_scale(os.path.join(PIXMAP_PATH, icon_name), w, h, True)
+            self.set(tree_iter, **{name: perc, f"{name}_pb": icon})
 
     def _disable_power_levels(self, tree_iter: Gtk.TreeIter) -> None:
-        row = self.get(tree_iter, "cell_fader", "battery", "rssi", "lq", "tpl")
-        if row["battery"] == row["rssi"] == row["tpl"] == row["lq"] == 0:
+        row = self.get(tree_iter, "cell_fader", "battery", "rssi", "tpl")
+        if row["battery"] == row["rssi"] == row["tpl"] == 0:
             return
 
-        self.set(tree_iter, rssi=0, lq=0, tpl=0)
-        self._prepare_fader(row["cell_fader"], lambda: self.set(tree_iter, rssi_pb=None, lq_pb=None, tpl_pb=None))\
+        self.set(tree_iter, rssi=0, tpl=0)
+        self._prepare_fader(row["cell_fader"], lambda: self.set(tree_iter, rssi_pb=None, tpl_pb=None))\
             .animate(start=1.0, end=0.0, duration=400)
 
     def _prepare_fader(self, fader: AnimBase, callback: Optional[Callable[[], None]] = None) -> AnimBase:
@@ -493,7 +431,6 @@ def tooltip_query(self, _tw: Gtk.Widget, x: int, y: int, _kb: bool, tooltip: Gtk
 
         elif path[1] == self.columns["battery_pb"] \
                 or path[1] == self.columns["tpl_pb"] \
-                or path[1] == self.columns["lq_pb"] \
                 or path[1] == self.columns["rssi_pb"]:
             tree_iter = self.get_iter(path[0])
             assert tree_iter is not None
@@ -506,7 +443,6 @@ def tooltip_query(self, _tw: Gtk.Widget, x: int, y: int, _kb: bool, tooltip: Gtk
 
             battery = self.get(tree_iter, "battery")["battery"]
             rssi = self.get(tree_iter, "rssi")["rssi"]
-            lq = self.get(tree_iter, "lq")["lq"]
             tpl = self.get(tree_iter, "tpl")["tpl"]
 
             if battery != 0:
@@ -534,12 +470,6 @@ def tooltip_query(self, _tw: Gtk.Widget, x: int, y: int, _kb: bool, tooltip: Gtk
                     lines.append(_("Received Signal Strength: %(rssi)u%% <i>(%(rssi_state)s)</i>") %
                                  {"rssi": rssi, "rssi_state": rssi_state})
 
-            if lq != 0:
-                if path[1] == self.columns["lq_pb"]:
-                    lines.append(_("<b>Link Quality: %(lq)u%%</b>") % {"lq": lq})
-                else:
-                    lines.append(_("Link Quality: %(lq)u%%") % {"lq": lq})
-
             if tpl != 0:
                 if tpl < 30:
                     tpl_state = _("Low")

data/icons/pixmaps/Makefile.am

@@ -11,16 +11,6 @@ pixmaps_DATA =			\
 	blueman-battery-80.png	\
 	blueman-battery-90.png	\
 	blueman-battery-100.png	\
-	blueman-lq-10.png	\
-	blueman-lq-20.png	\
-	blueman-lq-30.png	\
-	blueman-lq-40.png	\
-	blueman-lq-50.png	\
-	blueman-lq-60.png	\
-	blueman-lq-70.png	\
-	blueman-lq-80.png	\
-	blueman-lq-90.png	\
-	blueman-lq-100.png	\
 	blueman-rssi-10.png	\
 	blueman-rssi-20.png	\
 	blueman-rssi-30.png	\

meson.build

@@ -100,16 +100,6 @@ install_data(
     'data/icons/pixmaps/blueman-battery-80.png',
     'data/icons/pixmaps/blueman-battery-90.png',
     'data/icons/pixmaps/blueman-battery-100.png',
-    'data/icons/pixmaps/blueman-lq-10.png',
-    'data/icons/pixmaps/blueman-lq-20.png',
-    'data/icons/pixmaps/blueman-lq-30.png',
-    'data/icons/pixmaps/blueman-lq-40.png',
-    'data/icons/pixmaps/blueman-lq-50.png',
-    'data/icons/pixmaps/blueman-lq-60.png',
-    'data/icons/pixmaps/blueman-lq-70.png',
-    'data/icons/pixmaps/blueman-lq-80.png',
-    'data/icons/pixmaps/blueman-lq-90.png',
-    'data/icons/pixmaps/blueman-lq-100.png',
     'data/icons/pixmaps/blueman-rssi-10.png',
     'data/icons/pixmaps/blueman-rssi-20.png',
     'data/icons/pixmaps/blueman-rssi-30.png',

module/_blueman.pyx

@@ -72,16 +72,6 @@ cdef extern from "bluetooth/rfcomm.h":
 
 
 cdef extern from "libblueman.h":
-    cdef struct conn_info_handles:
-        unsigned int handle
-        int dd
-
-
-    cdef int connection_init(int dev_id, char *addr, conn_info_handles *ci)
-    cdef int connection_get_rssi(conn_info_handles *ci, signed char *ret_rssi)
-    cdef int connection_get_lq(conn_info_handles *ci, unsigned char *ret_lq)
-    cdef int connection_get_tpl(conn_info_handles *ci, signed char *ret_tpl, unsigned char type)
-    cdef int connection_close(conn_info_handles *ci)
     cdef int c_get_rfcomm_channel "get_rfcomm_channel" (unsigned short service_class, char* btd_addr)
     cdef int get_rfcomm_list(rfcomm_dev_list_req **ret)
     cdef int c_create_rfcomm_device "create_rfcomm_device" (char *local_address, char *remote_address, int channel)
@@ -95,12 +85,6 @@ class RFCOMMError(Exception):
 
 ERR = {
     -1:"Can't allocate memory",
-    -2:"HCI device open failed",
-    -3:"Not connected",
-    -4:"Get connection info failed",
-    -5:"Read RSSI failed",
-    -6:"Read transmit power level request failed",
-    -7:"Read Link quality failed",
     -8:"Getting rfcomm list failed",
     -9:"ERR_SOCKET_FAILED",
     -12: "Can't bind RFCOMM socket",
@@ -207,63 +191,6 @@ def destroy_bridge(py_name="pan1"):
     if err < 0:
         raise BridgeException(-err)
 
-
-class ConnInfoReadError(Exception):
-    pass
-
-cdef class conn_info:
-    cdef conn_info_handles ci
-    cdef int hci
-    cdef char* addr
-    cdef public bint failed
-
-    def __init__(self, py_addr, py_hci_name="hci0"):
-        self.failed = False
-        py_bytes_addr = py_addr.encode("UTF-8")
-        cdef char* addr = py_bytes_addr
-
-        py_bytes_hci_name = py_hci_name.encode("UTF-8")
-        cdef char* hci_name = py_bytes_hci_name
-
-
-        self.hci = int(hci_name[3:])
-        self.addr = addr
-
-    def init(self):
-        res = connection_init(self.hci, self.addr, & self.ci)
-        if res < 0:
-            self.failed = True
-            raise ConnInfoReadError(ERR[res])
-
-    def deinit(self):
-        if self.failed:
-            return
-        connection_close(&self.ci)
-
-    def get_rssi(self):
-        cdef signed char rssi
-        res = connection_get_rssi(&self.ci, &rssi)
-        if res < 0:
-            raise ConnInfoReadError(ERR[res])
-
-        return rssi
-
-    def get_lq(self):
-        cdef unsigned char lq
-        res = connection_get_lq(&self.ci, &lq)
-        if res < 0:
-            raise ConnInfoReadError(ERR[res])
-
-        return lq
-
-    def get_tpl(self, tp=0):
-        cdef signed char tpl
-        res = connection_get_tpl(&self.ci, &tpl, tp)
-        if res < 0:
-            raise ConnInfoReadError(ERR[res])
-
-        return tpl
-
 def device_info(py_hci_name="hci0"):
     py_bytes_hci_name = py_hci_name.encode("UTF-8")
     cdef char* hci_name = py_bytes_hci_name