Source code for libqtile.widget.bluetooth

# Copyright (c) 2023 elParaguayo
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
import asyncio
import contextlib
from enum import Enum

from dbus_next.aio import MessageBus
from dbus_next.constants import BusType
from dbus_next.errors import DBusError, InterfaceNotFoundError

from libqtile.command.base import expose_command
from libqtile.log_utils import logger
from libqtile.utils import create_task
from libqtile.widget import base

BLUEZ_SERVICE = "org.bluez"
BLUEZ_DEVICE = "org.bluez.Device1"
BLUEZ_ADAPTER = "org.bluez.Adapter1"
BLUEZ_BATTERY = "org.bluez.Battery1"
OBJECT_MANAGER_INTERFACE = "org.freedesktop.DBus.ObjectManager"
PROPERTIES_INTERFACE = "org.freedesktop.DBus.Properties"

def _catch_dbus_error(msg):
    """Decorator to catch DBusErrors and log a message."""

    def _wrapper(func):
        async def f(self):
                await func(self)
            except DBusError:

        return f

    return _wrapper

class DeviceState(Enum):
    PAIRED = 1
    UNPAIRED = 2

class _BluetoothBase:
    """Base class with some common requirements for devices and adapters."""

    def __init__(self, path, interface, properties_interface, widget):
        self.path = path
        self.interface = interface
        self.widget = widget = properties_interface
        self._name = ""

    def __repr__(self):
        """Neater repr to help debugging."""
        return f"<{self.__class__.__name__}: {} ({self.path})>"

    def __del__(self):
        """Remove signal listener when garbage collected."""
        with contextlib.suppress(RuntimeError):

    def properties_changed(self, _interface_name, _changed_properties, _invalidated_properties):
        """Handler for properties_changed signal."""

    async def update_props(self):
        raise NotImplementedError

class BluetoothDevice(_BluetoothBase):
    Helper class to represent an org.bluez.Device1 object.

    Exposes basic properties/methods and also listens to signals to
    update properties as needed.

    def __init__(self, path, interface, properties_interface, widget):
        _BluetoothBase.__init__(self, path, interface, properties_interface, widget)
        self._connected = False
        self._paired = False
        self._status = DeviceState.UNPAIRED
        self._adapter = ""
        self.has_name = False
        self.battery_device = None
        self._battery = 0

    @_catch_dbus_error("Unable to connect to device: %s.")
    async def connect(self):
        await self.interface.call_connect()

    @_catch_dbus_error("Unable to disconnect device: %s.")
    async def disconnect(self):
        await self.interface.call_disconnect()

    @_catch_dbus_error("Unable to pair device: %s.")
    async def pair_and_connect(self):
        await self.interface.call_pair()
        await self.connect()

    async def action(self):
        """Helper method to call appropriate method based on device status."""
        if self.connected:
            await self.disconnect()
        elif self.paired and not self.connected:
            await self.connect()
        elif not self.paired:
            await self.pair_and_connect()

    def name(self):
        return self._name

    def connected(self):
        return self._connected

    def paired(self):
        return self._paired

    def status(self):
        return self._status

    def battery(self):
        if self.battery_device:
            return self._battery
        return ""

    def adapter(self):
        a = self.widget.adapters.get(self._adapter)
        if a:
            return "Unknown"

    def add_battery(self):
        """Triggers adding battery interface."""

        def refresh(_):
            if self.battery_device:

        task = create_task(self.get_battery())

    def remove_battery(self):
        self.battery_device = None
        self._battery = 0

    async def get_battery(self):
        proxy = await self.widget.get_proxy(self.path)
        with contextlib.suppress(InterfaceNotFoundError):
            self.battery_device = proxy.get_interface(BLUEZ_BATTERY)

    async def update_props(self, setup=False):
        """Refresh all the properties for the device."""
        # Some devices don't report a name so we fall back to the device address
            self._name = await self.interface.get_name()
            self.has_name = True
        except (AttributeError, DBusError):
            self._name = await self.interface.get_address()
            self.has_name = False

        self._connected, self._paired, self._adapter = await asyncio.gather(

        # If we're setting up, let's see if a battery device is available
        # This may happen if the device is already connected when the widget starts
        if setup:
            await self.get_battery()

        if self.battery_device:
            self._battery = await self.battery_device.get_percentage()

        if self._connected:
            self._status = DeviceState.CONNECTED
        elif self._paired and not self._connected:
            self._status = DeviceState.PAIRED
            self._status = DeviceState.UNPAIRED

        if not setup:

    async def check(self):
        """Checks if device belongs to requested adapter."""
        await self.update_props(setup=True)

        if not self.widget.adapter_paths:
            return True, self

        for path in self.widget.adapter_paths:
            if path == self._adapter:
                return True, self

        return False, self

class BluetoothAdapter(_BluetoothBase):
    Helper class for Bluetooth adapters.

    Exposes basic properties/methods and also listens to signals to
    update properties as needed.

    def __init__(self, path, interface, properties_interface, widget):
        _BluetoothBase.__init__(self, path, interface, properties_interface, widget)
        self._discovering = False
        self._powered = False

    @_catch_dbus_error("Unable to start discovery on adapter: %s.")
    async def start_discovery(self):
        await self.interface.call_start_discovery()

    @_catch_dbus_error("Unable to stop discovery on adapter: %s.")
    async def stop_discovery(self):
        await self.interface.call_stop_discovery()

    @_catch_dbus_error("Unable to set power state for adapter: %s.")
    async def power(self):
        await self.interface.set_powered(not self._powered)

    def discovering(self):
        return self._discovering

    def powered(self):
        return self._powered

    def name(self):
        return self._name

    async def discover(self):
        if self.discovering:
            await self.stop_discovery()
            await self.start_discovery()

    async def update_props(self, setup=False):
        self._discovering = await self.interface.get_discovering()
        self._powered = await self.interface.get_powered()
        self._name = await self.interface.get_name()

        if not setup:

[docs]class Bluetooth(base._TextBox, base.MarginMixin): """ Bluetooth widget that provides following functionality: - View multiple adapters/devices (adapters can be filtered) - Set power and discovery status for adapters - Connect/disconnect/pair devices The widget works by providing a menu in the bar. Different items are accessed by scrolling up and down on the widget. Clicking on an adapter will open a submenu allowing you to set power and discovery status. Clicking on a device will perform an action based on the status of that device: - Connected devices will be disconnected - Disconnected devices will be connected - Unpaired devices (which appear if discovery is on) will be paired and connected Symbols are used to show the status of adapters and devices. Battery level for bluetooth devices can also be shown if available. This functionality is not available by default on all distros. If it doesn't work, you can try adding ``Experimental = true`` to ``/etc/bluetooth/main.conf``. """ defaults = [ ("hide_unnamed_devices", False, "Devices with no name will be hidden from scan results"), ("symbol_connected", "*", "Symbol to indicate device is connected"), ("symbol_paired", "-", "Symbol to indicate device is paired but unconnected"), ("symbol_unknown", "?", "Symbol to indicate device is unpaired"), ("symbol_powered", ("*", "-"), "Symbols when adapter is powered and unpowered."), ( "symbol_discovery", ("D", ""), "Symbols when adapter is discovering and not discovering", ), ( "device_format", "Device: {name}{battery_level} [{symbol}]", "Text to display when showing bluetooth device. " "The ``{adapter`` field is also available if you're using multiple adapters.", ), ( "device_battery_format", " ({battery}%)", "Text to be shown if device reports battery level", ), ( "adapter_format", "Adapter: {name} [{powered}{discovery}]", "Text to display when showing adapter device.", ), ( "adapter_paths", [], "List of DBus object path for bluetooth adapter (e.g. '/org/bluez/hci0'). " "Empty list will show all adapters.", ), ( "default_text", "BT {connected_devices}", "Text to show when not scrolling through menu. " "Available fields: 'connected_devices' list of connected devices, " "'num_connected_devices' number of connected devices, " "'adapters' list of bluetooth adapters, 'num_adapters' number of bluetooth adapters.", ), ( "default_show_battery", False, "Include battery level of 'connected_devices' in 'default_text'. Uses 'device_battery_format'.", ), ("separator", ", ", "Separator for lists in 'default_text'."), ( "default_timeout", None, "Time before reverting to default_text. If 'None', text will stay on selected item.", ), ( "device", None, "Device path, can be found with d-feet or similar dbus explorer. " "When set, the widget will default to showing this device status.", ), ] def __init__(self, **config): base._TextBox.__init__(self, **config) self.add_defaults(Bluetooth.defaults) self.add_defaults(base.MarginMixin.defaults) self.connected = False self.bus = None self.devices = {} self.adapters = {} self._lines = [] self._line_index = 0 self._adapter_index = 0 self._setting_up = True self.show_adapter = False self.device_found = False self.add_callbacks( {"Button1":, "Button4": self.scroll_up, "Button5": self.scroll_down} ) self.timer = None self.object_manager = None def _configure(self, qtile, bar): base._TextBox._configure(self, qtile, bar) self.symbols = (self.symbol_connected, self.symbol_paired, self.symbol_unknown) async def _config_async(self): await self._connect() async def _connect(self): """Connect to bus and set up key listeners.""" self.bus = await MessageBus(bus_type=BusType.SYSTEM).connect() # Get the object manager proxy = await self.get_proxy("/") self.object_manager = proxy.get_interface(OBJECT_MANAGER_INTERFACE) # Subscribe to signals for new and removed interfaces self.object_manager.on_interfaces_added(self._interface_added) self.object_manager.on_interfaces_removed(self._interface_removed) await self._get_managed_objects() self.refresh() async def get_proxy(self, path): """Provides proxy object after introspecting the given path.""" device_introspection = await self.bus.introspect(BLUEZ_SERVICE, path) proxy = self.bus.get_proxy_object(BLUEZ_SERVICE, path, device_introspection) return proxy async def _get_managed_objects(self): """ Retrieve list of managed objects. These are devices that have previously been paired but may or may not currently be connected. Additionally, if the device is scanning, available objects will also appear here, albeit temporarily. """ self._setting_up = True objects = await self.object_manager.call_get_managed_objects() for path, interfaces in objects.items(): self._interface_added(path, interfaces) self._setting_up = False def _interface_added(self, path, interfaces): """Handles the object based on the interface type.""" task = None # Create device or adapter for device_type in (BLUEZ_DEVICE, BLUEZ_ADAPTER): if device_type in interfaces: task = create_task(self._add_object(path, device_type)) break if task is not None and not self._setting_up: task.add_done_callback(lambda *args: self.refresh()) # Battery interface is added after the device is connected # so no task will have been created at this point. if not task and BLUEZ_BATTERY in interfaces: if device := self.devices.get(path): # Get device to load battery interface device.add_battery() def _interface_removed(self, path, interfaces): # Object has been removed so remove from our list of available devices updated = False if BLUEZ_DEVICE in interfaces: with contextlib.suppress(KeyError): del self.devices[path] updated = True elif BLUEZ_ADAPTER in interfaces: with contextlib.suppress(KeyError): del self.adapters[path] updated = True elif BLUEZ_BATTERY in interfaces: device = self.devices.get(path) if device: device.remove_battery() if updated and not self._setting_up: self.refresh() async def _add_object(self, path, device_type): proxy = await self.get_proxy(path) # Check if object is a valid bluetooth device and ignore it if not try: interface = proxy.get_interface(device_type) except InterfaceNotFoundError: return # Get the properties interface so we can listed to signals properties = proxy.get_interface(PROPERTIES_INTERFACE) # Create an object to represent this device and add to our list if device_type == BLUEZ_DEVICE: # The device will be added to self.devices as part of the __init__ # process after checking whether we need to filter out the device # if it's connected to an unwatched adapter # This is preferable to substring matching on 'path' which could # result in false positives (in very rare situations) device = BluetoothDevice(path, interface, properties, self) task = create_task(device.check()) task.add_done_callback(self._add_device) elif device_type == BLUEZ_ADAPTER: if not self.adapter_paths or path in self.adapter_paths: adapter = BluetoothAdapter(path, interface, properties, self) self.adapters[path] = adapter def _add_device(self, task): success, device = task.result() if success: self.devices[device.path] = device self.refresh() def refresh(self): if self._setting_up: return # Store lines in a variable as we'll need to access them elsewhere # Each entry is a tuple of (display text, callable) self._lines = [] # If we've clicked on an adapter then we're just showing the adapter submenu if self.show_adapter: self._lines.extend(self._get_adapter_menu(self._shown_adapter)) # Otherwise we're in default behavior else: # Line 1 is the text to be formatted according to "default_text" connected = [d for d in self.devices.values() if d.connected] adapters = [ for a in self.adapters.values()] if self.default_show_battery: connected_devices = [ "{name}{battery}".format(, battery=self.device_battery_format.format(battery=d.battery) if d.battery else "", ) for d in connected ] else: connected_devices = [ for d in connected] self._lines.append( ( self.default_text.format( connected_devices=self.separator.join(connected_devices), num_connected_devices=len(connected_devices), adapters=self.separator.join(adapters), num_adapters=len(adapters), ), lambda: None, ) ) # Next is the adapters... def show(adapter): """Function to trigger the adapter submenu.""" self.show_adapter = True self._shown_adapter = adapter # Store the current menu position self._adapter_index = self._line_index # Change menu position to the first item in the submenu self._line_index = 0 self.refresh() for adapter in self.adapters.values(): self._lines.append((adapter, lambda a=adapter: show(a))) # Finally, loop over all the devices for device in self.devices.values(): self._lines.append((device, lambda d=device: create_task(d.action()))) if self._lines: # If user has set default device, check if it should be shown # This will only force the display to that widget the first time the device is found # i.e. once user has scrolled to a different device, it will no longer return to the # set device. if not self.device_found and self.device is not None: for i, (obj, _) in enumerate(self._lines): if isinstance(obj, BluetoothDevice): if self.device in obj.path: self._line_index = i self.device_found = True break self.show_line() else: self.update("") def _get_adapter_menu(self, adapter): """Builds a submenu for the selected adapter.""" state = "off" if adapter.powered else "on" discovery = "off" if adapter.discovering else "on" def exit(): self.show_adapter = False # Restore menu position self._line_index = self._adapter_index self.refresh() return [ (f"Turn power {state}", lambda a=adapter: create_task(a.power())), (f"Turn discovery {discovery}", lambda a=adapter: create_task(, ("Exit", lambda: exit()), ] def show_line(self): """Formats the text of the current menu item.""" if not self._lines: return obj = None # If devices disappear we may have an invalid line index while obj is None: try: obj, action = self._lines[self._line_index] except IndexError: self._line_index -= 1 self.update(self.format_object(obj)) def format_object(self, obj): """Takes the given object and returns a formatted string representing the object.""" if isinstance(obj, BluetoothDevice): # status.value is 0 for connected, 1 for paired (and unconnected), 2 for unpaired symbol = self.symbols[obj.status.value] if obj.battery: battery_level = self.device_battery_format.format(battery=obj.battery) else: battery_level = "" return self.device_format.format( symbol=symbol,, adapter=obj.adapter, battery_level=battery_level ) elif isinstance(obj, BluetoothAdapter): powered = 0 if obj.powered else 1 discovery = 0 if obj.discovering else 1 return self.adapter_format.format( powered=self.symbol_powered[powered], discovery=self.symbol_discovery[discovery],, ) elif isinstance(obj, str): return obj # Shouldn't happen but let's be safe! return ""
[docs] @expose_command def scroll_up(self): """Scroll up to next item.""" self._scroll(1)
[docs] @expose_command def scroll_down(self): """Scroll down to next item.""" self._scroll(-1)
[docs] @expose_command def click(self): """Perform default action on visible item.""" with contextlib.suppress(IndexError): _, action = self._lines[self._line_index] action()
def _scroll(self, step): if self.timer is not None: self.timer.cancel() if self._lines: self._line_index = (self._line_index + step) % len(self._lines) self.show_line() if self.default_timeout is not None: self.timer = self.timeout_add(self.default_timeout, self.hide) def hide(self): """Revert widget contents to default.""" self._line_index = 0 self.show_adapter = False self.refresh() def finalize(self): # if we failed to connect, there is nothing to finalize. if self.bus is None: return # Remove dbus signal handlers before finalising. # Clearing dicts will call the __del__ method on the stored objects # which has been defined to remove signal handlers self.devices.clear() self.adapters.clear() # Remove object manager's handlers if self.object_manager is not None: self.object_manager.off_interfaces_added(self._interface_added) self.object_manager.off_interfaces_removed(self._interface_removed) # Disconnect the bus connection self.bus.disconnect() self.bus = None base._TextBox.finalize(self)