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, Dict
+from typing import Optional, TYPE_CHECKING, List, Any, cast, Callable, Dict
 import html
 import logging
 import cairo
@@ -16,12 +16,10 @@
 from blueman.Functions import launch
 from blueman.Sdp import ServiceUUID, OBEX_OBJPUSH_SVCLASS_ID
 from blueman.gui.GtkAnimation import TreeRowFade, CellFade, AnimBase
-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 Gio
 from gi.repository import Gdk
 from gi.repository import GdkPixbuf
@@ -48,9 +46,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")},
@@ -61,7 +56,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},
@@ -75,8 +69,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.manager.connect_signal("battery-created", self.on_battery_created)
  self.manager.connect_signal("battery-removed", self.on_battery_removed)
  self._batteries: Dict[str, Battery] = {}
@@ -381,48 +373,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.liststore
- 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}")
 
@@ -451,9 +401,7 @@ def row_update_event(self, tree_iter: Gtk.TreeIter, key: str, value: Any) -> Non
  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)
  elif key == "Name":
  self.set(tree_iter, no_name=False)
@@ -462,52 +410,34 @@ def row_update_event(self, tree_iter: Gtk.TreeIter, key: str, value: Any) -> Non
  elif key == "Blocked":
  self.set(tree_iter, blocked=value)
 
- 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))
 
-  obj_path = device.get_object_path()
- if obj_path in self._batteries:
-  bars["battery"] = self._batteries[obj_path]["Percentage"]
+ 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
 
- # cinfo init may fail for bluetooth devices version 4 and up
- # FIXME Workaround is horrible and we should show something better
- if cinfo.failed:
- bars.update({"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:
+ 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, battery=0, rssi=0, lq=0, tpl=0)
- self._prepare_fader(row["cell_fader"], lambda: self.set(tree_iter, battery_pb=None, rssi_pb=None, lq_pb=None,
+ self.set(tree_iter, battery=0, rssi=0, tpl=0)
+ self._prepare_fader(row["cell_fader"], lambda: self.set(tree_iter, battery_pb=None, 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:
@@ -559,7 +489,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
@@ -572,7 +501,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:
@@ -600,12 +528,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