Something went wrong. Try again.
A Material 3 inspired desktop shell for Niri and Hyprland created with Ignis.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408from __future__ import annotations
import logging
from gi.repository import Gio, GLib
from modules.shared_modules.gdbus import ( OBJECT_MANAGER_INTERFACE, PROPERTIES_INTERFACE, call, remote_error_name, spawn,)from modules.shared_modules.notifier import ChangeNotifier
from .agent import AGENT_PATH, CAPABILITY, PairingAgentfrom .model import ( Adapter, Availability, BluetoothError, BluetoothSnapshot, Device, ErrorCode, LinkState, Transport,)
log = logging.getLogger("exo.bluetooth")
BUS_NAME = "org.bluez"ROOT_PATH = "/"BLUEZ_ROOT = "/org/bluez"
ADAPTER = "org.bluez.Adapter1"DEVICE = "org.bluez.Device1"BATTERY = "org.bluez.Battery1"AGENT_MANAGER = "org.bluez.AgentManager1"WATCHED = (ADAPTER, DEVICE, BATTERY)
PAIR_TIMEOUT_MS = 120_000CONNECT_TIMEOUT_MS = 60_000
ERROR_BY_REMOTE = { "NotReady": ErrorCode.ADAPTER_OFF, "InProgress": ErrorCode.BUSY, "DoesNotExist": ErrorCode.NOT_FOUND, "AuthenticationFailed": ErrorCode.PAIRING_FAILED, "AuthenticationRejected": ErrorCode.PAIRING_REJECTED, "AuthenticationCanceled": ErrorCode.PAIRING_REJECTED, "AuthenticationTimeout": ErrorCode.PAIRING_TIMEOUT, "ConnectionAttemptFailed": ErrorCode.CONNECTION_FAILED, "NotAvailable": ErrorCode.CONNECTION_FAILED, "InvalidArguments": ErrorCode.INVALID_INPUT,}
def failure(error: GLib.Error) -> BluetoothError: short = remote_error_name(error) # BlueZ reports a blocked radio only as free text on the generic Failed error. if short == "Failed" and "rfkill" in error.message.lower(): return BluetoothError(ErrorCode.BLOCKED, error.message) return BluetoothError(ERROR_BY_REMOTE.get(short, ErrorCode.FAILED), error.message)
def device_from(path: str, interfaces: dict, state: LinkState) -> Device: props = interfaces[DEVICE] address = props.get("Address", "") alias = props.get("Alias") or props.get("Name") or address connected = bool(props.get("Connected")) if state is LinkState.IDLE and connected: state = LinkState.CONNECTED return Device( device_id=path, name=alias, address=address, named=alias != address.replace(":", "-"), icon=props.get("Icon", ""), paired=bool(props.get("Paired")), connected=connected, trusted=bool(props.get("Trusted")), blocked=bool(props.get("Blocked")), state=state, rssi=props.get("RSSI"), battery=interfaces.get(BATTERY, {}).get("Percentage"), uuids=tuple(props.get("UUIDs", ())), low_energy_only=props.get("AddressType") == "random", )
def adapter_from(path: str, props: dict) -> Adapter: return Adapter( adapter_id=path, name=props.get("Alias") or props.get("Name") or path.rsplit("/", 1)[-1], address=props.get("Address", ""), powered=bool(props.get("Powered")), discoverable=bool(props.get("Discoverable")), discoverable_timeout=int(props.get("DiscoverableTimeout", 180)), pairable=bool(props.get("Pairable")), pairable_timeout=int(props.get("PairableTimeout", 0)), discovering=bool(props.get("Discovering")), )
class BluetoothController(ChangeNotifier): def __init__(self, bus: Gio.DBusConnection): super().__init__() self._bus = bus self._objects: dict[str, dict[str, dict]] = {} self._busy: dict[str, LinkState] = {} self._selected: str | None = None self._transport = Transport.AUTO self._service_running = False self._agent = PairingAgent(bus, self._device_name, self._on_agent_changed)
self._subscribe() self._name_watch = Gio.bus_watch_name_on_connection( bus, BUS_NAME, Gio.BusNameWatcherFlags.NONE, self._on_name_appeared, self._on_name_vanished, )
def _subscribe(self): def subscribe(interface, member, handler): self._bus.signal_subscribe( BUS_NAME, interface, member, None, None, Gio.DBusSignalFlags.NONE, handler, None )
subscribe(PROPERTIES_INTERFACE, "PropertiesChanged", self._on_properties_changed) subscribe(OBJECT_MANAGER_INTERFACE, "InterfacesAdded", self._on_interfaces_added) subscribe(OBJECT_MANAGER_INTERFACE, "InterfacesRemoved", self._on_interfaces_removed)
def _on_name_appeared(self, _bus, _name, _owner): self._service_running = True spawn(self._load(), "bluez initial load")
def _on_name_vanished(self, _bus, _name): self._service_running = False self._objects = {} self._busy = {} self._agent.clear() self.notify_changed()
async def _load(self): try: reply = await call( self._bus, BUS_NAME, ROOT_PATH, OBJECT_MANAGER_INTERFACE, "GetManagedObjects" ) except GLib.Error as error: log.warning("bluez GetManagedObjects failed: %s", error.message) return self._objects = {path: dict(ifaces) for path, ifaces in reply.unpack()[0].items()} self.notify_changed() try: await call( self._bus, BUS_NAME, BLUEZ_ROOT, AGENT_MANAGER, "RegisterAgent", GLib.Variant("(os)", (AGENT_PATH, CAPABILITY)), ) except GLib.Error as error: if remote_error_name(error) != "AlreadyExists": log.warning("bluez RegisterAgent failed: %s", error.message)
def _on_properties_changed(self, _bus, _sender, path, _iface, _signal, params, *_): interface, changed, invalidated = params.unpack() if interface not in WATCHED: return props = self._objects.setdefault(path, {}).setdefault(interface, {}) props.update(changed) for name in invalidated: props.pop(name, None) self._drop_finished_display(path, interface, changed) self.notify_changed()
def _on_interfaces_added(self, _bus, _sender, _path, _iface, _signal, params, *_): path, interfaces = params.unpack() self._objects.setdefault(path, {}).update(interfaces) self.notify_changed()
def _on_interfaces_removed(self, _bus, _sender, _path, _iface, _signal, params, *_): path, interfaces = params.unpack() known = self._objects.get(path, {}) for interface in interfaces: known.pop(interface, None) if not known: self._objects.pop(path, None) self._busy.pop(path, None) self.notify_changed()
def _drop_finished_display(self, path: str, interface: str, changed: dict): request = self._agent.request if request is None or request.device_id != path or interface != DEVICE: return if request.kind.is_display_only and (changed.get("Paired") or "Connected" in changed): self._agent.clear()
def _on_agent_changed(self): self.notify_changed()
def _device_name(self, path: str) -> str: interfaces = self._objects.get(path, {}) if DEVICE not in interfaces: return path.rsplit("/", 1)[-1].replace("dev_", "").replace("_", ":") return device_from(path, interfaces, LinkState.IDLE).name
def _adapter_paths(self) -> list[str]: return sorted(p for p, ifaces in self._objects.items() if ADAPTER in ifaces)
def _active_adapter_path(self) -> str | None: paths = self._adapter_paths() if self._selected in paths: return self._selected powered = [p for p in paths if self._objects[p][ADAPTER].get("Powered")] return (powered or paths or [None])[0]
@property def snapshot(self) -> BluetoothSnapshot: if not self._service_running: return BluetoothSnapshot() paths = self._adapter_paths() adapters = tuple(adapter_from(p, self._objects[p][ADAPTER]) for p in paths) active = self._active_adapter_path() if active is None: return BluetoothSnapshot(Availability.NO_ADAPTER, transport=self._transport)
devices = tuple( device_from(path, interfaces, self._busy.get(path, LinkState.IDLE)) for path, interfaces in self._objects.items() if DEVICE in interfaces and interfaces[DEVICE].get("Adapter") == active ) return BluetoothSnapshot( availability=Availability.READY, adapters=adapters, adapter=next(a for a in adapters if a.adapter_id == active), devices=devices, transport=self._transport, request=self._agent.request, )
def _require_adapter(self) -> str: path = self._active_adapter_path() if path is None: raise BluetoothError(ErrorCode.UNAVAILABLE, "no adapter") return path
def _require_device(self, device_id: str) -> None: if DEVICE not in self._objects.get(device_id, {}): raise BluetoothError(ErrorCode.NOT_FOUND, device_id)
async def _invoke(self, path, interface, method, params=None, timeout_ms=30_000): try: return await call(self._bus, BUS_NAME, path, interface, method, params, timeout_ms) except GLib.Error as error: raise failure(error) from error
async def _set(self, path: str, interface: str, name: str, signature: str, value): await self._invoke( path, PROPERTIES_INTERFACE, "Set", GLib.Variant("(ssv)", (interface, name, GLib.Variant(signature, value))), )
async def _with_busy(self, device_id: str, state: LinkState, action): self._busy[device_id] = state self.notify_changed() try: return await action finally: self._busy.pop(device_id, None) self.notify_changed()
def select_adapter(self, adapter_id: str) -> None: if adapter_id in self._adapter_paths() and adapter_id != self._selected: self._selected = adapter_id self.notify_changed()
async def set_powered(self, enabled: bool) -> None: await self._set(self._require_adapter(), ADAPTER, "Powered", "b", enabled)
async def set_alias(self, name: str) -> None: await self._set(self._require_adapter(), ADAPTER, "Alias", "s", name.strip())
async def set_discoverable(self, enabled: bool) -> None: await self._set(self._require_adapter(), ADAPTER, "Discoverable", "b", enabled)
async def set_discoverable_timeout(self, seconds: int) -> None: await self._set(self._require_adapter(), ADAPTER, "DiscoverableTimeout", "u", seconds)
async def set_pairable(self, enabled: bool) -> None: await self._set(self._require_adapter(), ADAPTER, "Pairable", "b", enabled)
async def set_transport(self, transport: Transport) -> None: self._transport = transport self.notify_changed() adapter = self._active_adapter_path() if adapter is not None and self._objects[adapter][ADAPTER].get("Discovering"): await self._apply_discovery_filter(adapter)
async def _apply_discovery_filter(self, adapter: str) -> None: options = { "Transport": GLib.Variant("s", self._transport.value), "DuplicateData": GLib.Variant("b", False), } await self._invoke( adapter, ADAPTER, "SetDiscoveryFilter", GLib.Variant("(a{sv})", (options,)) )
async def start_discovery(self) -> None: adapter = self._require_adapter() await self._apply_discovery_filter(adapter) try: await self._invoke(adapter, ADAPTER, "StartDiscovery") except BluetoothError as error: if error.code is not ErrorCode.BUSY: raise
async def stop_discovery(self) -> None: adapter = self._active_adapter_path() if adapter is None or not self._objects[adapter][ADAPTER].get("Discovering"): return try: await self._invoke(adapter, ADAPTER, "StopDiscovery") except BluetoothError as error: if error.code not in (ErrorCode.NOT_FOUND, ErrorCode.ADAPTER_OFF): raise
async def connect_to(self, device_id: str) -> None: self._require_device(device_id) try: await self._with_busy( device_id, LinkState.CONNECTING, self._invoke(device_id, DEVICE, "Connect", timeout_ms=CONNECT_TIMEOUT_MS), ) except BluetoothError as error: if error.code is not ErrorCode.BUSY: raise
async def disconnect_from(self, device_id: str) -> None: self._require_device(device_id) await self._with_busy( device_id, LinkState.DISCONNECTING, self._invoke(device_id, DEVICE, "Disconnect") )
async def pair(self, device_id: str) -> None: self._require_device(device_id) await self._with_busy( device_id, LinkState.PAIRING, self._invoke(device_id, DEVICE, "Pair", timeout_ms=PAIR_TIMEOUT_MS), ) self._agent.clear() await self._set(device_id, DEVICE, "Trusted", "b", True) await self.connect_to(device_id)
async def cancel_pairing(self, device_id: str) -> None: await self._invoke(device_id, DEVICE, "CancelPairing")
async def forget(self, device_id: str) -> None: self._require_device(device_id) adapter = self._objects[device_id][DEVICE].get("Adapter", self._require_adapter()) await self._invoke( adapter, ADAPTER, "RemoveDevice", GLib.Variant("(o)", (device_id,)) )
async def set_trusted(self, device_id: str, trusted: bool) -> None: self._require_device(device_id) await self._set(device_id, DEVICE, "Trusted", "b", trusted)
async def set_blocked(self, device_id: str, blocked: bool) -> None: self._require_device(device_id) await self._set(device_id, DEVICE, "Blocked", "b", blocked)
async def rename(self, device_id: str, name: str) -> None: self._require_device(device_id) await self._set(device_id, DEVICE, "Alias", "s", name.strip())
def respond(self, request_id: int, accepted: bool, text: str = "") -> None: request = self._agent.request if request is not None and request.request_id == request_id and request.kind.is_display_only: self._agent.clear() spawn(self.cancel_pairing(request.device_id), "bluez cancel pairing") return self._agent.respond(request_id, accepted, text)
_controller: BluetoothController | None = None_resolved = False
def get_bluetooth_controller() -> BluetoothController | None: global _controller, _resolved if _resolved: return _controller _resolved = True try: bus = Gio.bus_get_sync(Gio.BusType.SYSTEM, None) except GLib.Error as error: log.warning("system bus unavailable: %s", error.message) return None _controller = BluetoothController(bus) return _controller