From 32080ab17f9eed5028a9cfc2fdd2518a67c5cf2e Mon Sep 17 00:00:00 2001 From: S4NKALP Date: Wed, 17 Jun 2026 04:51:51 +0545 Subject: [PATCH] feat(src/services): add bluetooth device proxy and wifi connection logic MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implemented robust BluetoothDevice proxy handling and enhanced Wi‑Fi reconnection with async callbacks and error handling. --- src/services/bluetooth.py | 700 ++++++++++++++++++++++++++++++++++++++ src/services/network.py | 66 +++- 2 files changed, 756 insertions(+), 10 deletions(-) create mode 100644 src/services/bluetooth.py diff --git a/src/services/bluetooth.py b/src/services/bluetooth.py new file mode 100644 index 0000000..be57ebb --- /dev/null +++ b/src/services/bluetooth.py @@ -0,0 +1,700 @@ +from collections.abc import Callable +from typing import Any, Concatenate, ParamSpec + +from fabric.core.service import Property, Service, Signal +from fabric.utils import Gio, GLib, logger + +P = ParamSpec("P") + +BLUEZ_SERVICE = "org.bluez" +BLUEZ_ADAPTER_IFACE = "org.bluez.Adapter1" +BLUEZ_DEVICE_IFACE = "org.bluez.Device1" +BLUEZ_BATTERY_IFACE = "org.bluez.Battery1" +DBUS_OM_IFACE = "org.freedesktop.DBus.ObjectManager" +DBUS_PROPS_IFACE = "org.freedesktop.DBus.Properties" + + +def _unpack_variant(v): + if v is None: + return None + if isinstance(v, GLib.Variant): + return _unpack_variant(v.unpack()) + if isinstance(v, dict): + return {k: _unpack_variant(val) for k, val in v.items()} + if isinstance(v, (list, tuple)): + return [_unpack_variant(i) for i in v] + return v + + +def _make_proxy(bus: Gio.DBusConnection, path: str, iface: str) -> Gio.DBusProxy: + return Gio.DBusProxy.new_sync( + bus, + Gio.DBusProxyFlags.NONE, + None, + BLUEZ_SERVICE, + path, + iface, + None, + ) + + +class BluetoothDevice(Service): + @Signal + def changed(self) -> None: ... + + @Property(bool, "read-write", "is-connected", default_value=False) + def connected(self) -> bool: + return bool(self._get_prop("Connected")) + + @connected.setter + def connected(self, value: bool): + self.connecting = value + + @Property(bool, "read-write", "is-connecting", default_value=False) + def connecting(self) -> bool: + return self._connecting + + @connecting.setter + def connecting(self, value: bool): + self._connecting = True + self.notify("connecting") + + def _cb(proxy, res, _): + try: + proxy.call_finish(res) + logger.info( + f"[Bluetooth] {'Connected' if value else 'Disconnected'}: {self.address}" + ) + except Exception as e: + logger.warning( + f"[Bluetooth] connect_device failed for {self.address}: {e}" + ) + finally: + self._connecting = False + self.notify("connecting") + self.notify("connected") + self.emit("changed") + + method = "Connect" if value else "Disconnect" + self._proxy.call(method, None, Gio.DBusCallFlags.NONE, 30000, None, _cb, None) + + @Property(bool, "readable", "is-closed", default_value=False) + def closed(self) -> bool: + return self._closed + + @Property(bool, "read-write", "is-paired", default_value=False) + def paired(self) -> bool: + return bool(self._get_prop("Paired")) + + @Property(bool, "readable", "is-trusted", default_value=False) + def trusted(self) -> bool: + return bool(self._get_prop("Trusted")) + + @Property(str, "readable") + def address(self) -> str: + return str(self._get_prop("Address") or "") + + @Property(str, "readable") + def name(self) -> str: + return str(self._get_prop("Name") or self._get_prop("Alias") or "") + + @Property(str, "readable") + def alias(self) -> str: + return str(self._get_prop("Alias") or "") + + @Property(str, "readable") + def icon_name(self) -> str: + return str(self._get_prop("Icon") or "bluetooth") + + @Property(str, "readable") + def type(self) -> str: + return _icon_to_type(str(self._get_prop("Icon") or "")) + + @Property(int, "readable") + def battery_level(self) -> int: + return int(self._get_battery_prop("Percentage") or 0) + + @Property(float, "readable") + def battery_percentage(self) -> float: + return float(self._get_battery_prop("Percentage") or 0.0) + + def __init__( + self, bus: Gio.DBusConnection, object_path: str, props: dict, **kwargs + ): + super().__init__(**kwargs) + self._bus = bus + self._object_path = object_path + self._props: dict = props + self._connecting = False + self._closed = False + self._prop_sub_id: int = 0 + self._battery_proxy: Gio.DBusProxy | None = None + + self._proxy = _make_proxy(bus, object_path, BLUEZ_DEVICE_IFACE) + + try: + self._battery_proxy = _make_proxy(bus, object_path, BLUEZ_BATTERY_IFACE) + except Exception: + pass + + self._prop_sub_id = bus.signal_subscribe( + BLUEZ_SERVICE, + DBUS_PROPS_IFACE, + "PropertiesChanged", + object_path, + None, + Gio.DBusSignalFlags.NONE, + self._on_properties_changed, + None, + ) + + def _get_prop(self, name: str): + try: + v = self._proxy.get_cached_property(name) + if v is not None: + return v.unpack() + except Exception: + pass + return _unpack_variant(self._props.get(name)) + + def _get_battery_prop(self, name: str): + if not self._battery_proxy: + return None + try: + v = self._battery_proxy.get_cached_property(name) + return v.unpack() if v else None + except Exception: + return None + + def _on_properties_changed( + self, _conn, _sender, _path, _iface, _signal, params, _data + ): + changed = _unpack_variant(params)[1] if params else {} + self._props.update(changed) + prop_map = { + "Connected": "connected", + "Paired": "paired", + "Trusted": "trusted", + "Name": "name", + "Alias": "alias", + "Icon": "icon-name", + "Percentage": "battery-percentage", + } + for bluez_key, fabric_prop in prop_map.items(): + if bluez_key in changed: + self.notify(fabric_prop) + self.emit("changed") + + def close(self): + if self._prop_sub_id: + self._bus.signal_unsubscribe(self._prop_sub_id) + self._prop_sub_id = 0 + self._closed = True + self.notify("closed") + + def connect_device( + self, + connect: bool = True, + callback: Callable[Concatenate[bool, P], Any] | None = None, + *args: P.args, + **kwargs: P.kwargs, + ): + self.connecting = connect + if callback: + + def _once(*_): + callback(self.connected, *args, **kwargs) + self.disconnect(_id) + + _id = self.connect("changed", _once) + + def notifier(self, name: str, args=None): + self.notify(name) + self.emit("changed") + + +class BluetoothAdapter(Service): + @Signal + def changed(self) -> None: ... + + @Signal + def device_added(self, address: str) -> None: ... + + @Signal + def device_removed(self, address: str) -> None: ... + + @Property(list, "readable") + def devices(self) -> list: + + return [d for d in self._devices.values() if d.paired or d.trusted] + + @Property(list, "readable") + def connected_devices(self) -> list: + return [d for d in self._devices.values() if d.connected] + + @Property(str, "readable") + def object_path(self) -> str: + return self._object_path + + @Property(str, "readable") + def address(self) -> str: + return str(self._get_prop("Address") or "") + + @Property(str, "readable") + def name(self) -> str: + return str( + self._get_prop("Name") or self._get_prop("Alias") or self._object_path + ) + + @Property(str, "readable") + def state(self) -> str: + if not self._get_prop("Powered"): + return "off" + if self._get_prop("Discovering"): + return "discovering" + return "on" + + @Property(bool, "read-write", default_value=False) + def powered(self) -> bool: + return bool(self._get_prop("Powered")) + + @powered.setter + def powered(self, value: bool): + self._set_prop("Powered", GLib.Variant("b", value)) + + @Property(bool, "read-write", default_value=False) + def enabled(self) -> bool: + return self.powered + + @enabled.setter + def enabled(self, value: bool): + self.powered = value + + @Property(bool, "read-write", default_value=False) + def scanning(self) -> bool: + return bool(self._get_prop("Discovering")) + + @scanning.setter + def scanning(self, value: bool): + if value == self.scanning: + return + method = "StartDiscovery" if value else "StopDiscovery" + self._proxy.call( + method, + None, + Gio.DBusCallFlags.NONE, + 10000, + None, + lambda p, r, _: self._finish_call(p, r, method), + None, + ) + + def __init__( + self, bus: Gio.DBusConnection, object_path: str, props: dict, **kwargs + ): + super().__init__(**kwargs) + self._bus = bus + self._object_path = object_path + self._props: dict = props + + self._devices: dict[str, BluetoothDevice] = {} + self._prop_sub_id: int = 0 + self._scan_timeout_id: int = 0 + + self._proxy = _make_proxy(bus, object_path, BLUEZ_ADAPTER_IFACE) + + self._prop_sub_id = bus.signal_subscribe( + BLUEZ_SERVICE, + DBUS_PROPS_IFACE, + "PropertiesChanged", + object_path, + None, + Gio.DBusSignalFlags.NONE, + self._on_properties_changed, + None, + ) + + def _get_prop(self, name: str): + try: + v = self._proxy.get_cached_property(name) + if v is not None: + return v.unpack() + except Exception: + pass + return _unpack_variant(self._props.get(name)) + + def _set_prop(self, name: str, value: GLib.Variant): + try: + self._proxy.call_sync( + "org.freedesktop.DBus.Properties.Set", + GLib.Variant("(ssv)", (BLUEZ_ADAPTER_IFACE, name, value)), + Gio.DBusCallFlags.NONE, + 5000, + None, + ) + except Exception as e: + logger.warning(f"[Bluetooth] Set {name} on {self._object_path} failed: {e}") + + def _finish_call(self, proxy, res, method: str): + try: + proxy.call_finish(res) + except Exception as e: + logger.warning(f"[Bluetooth] {method} on {self._object_path} failed: {e}") + + def _on_properties_changed( + self, _conn, _sender, _path, _iface, _signal, params, _data + ): + changed = _unpack_variant(params)[1] if params else {} + self._props.update(changed) + for key in ("Powered", "Discovering"): + if key in changed: + self.notify("powered" if key == "Powered" else "scanning") + self.notify("enabled") + self.notify("state") + self.emit("changed") + + def add_device(self, object_path: str, props: dict): + addr: str = props.get("Address", _path_to_addr(object_path)) + if addr in self._devices: + return + + device = BluetoothDevice( + self._bus, + object_path, + props, + on_changed=lambda *_: self.emit("changed"), + ) + self._devices[addr] = device + + logger.info(f"[Bluetooth:{self.name}] Adding device: {addr}") + self.emit("device-added", addr) + self.notify("devices") + self.notify("connected-devices") + self.emit("changed") + + def remove_device(self, object_path: str): + addr = _path_to_addr(object_path) + device = self._devices.pop(addr, None) + if not device: + return + + was_visible = device.paired or device.trusted + logger.info(f"[Bluetooth:{self.name}] Removing device: {addr}") + + if was_visible: + self.emit("device-removed", addr) + if device.connected: + self.notify("connected-devices") + self.notify("devices") + self.emit("changed") + + device.close() + + def get_device(self, address: str) -> BluetoothDevice | None: + return self._devices.get(address) + + def scan(self, duration_ms: int = 10_000): + """Start discovery and automatically stop after duration_ms.""" + if self._scan_timeout_id: + GLib.source_remove(self._scan_timeout_id) + self._scan_timeout_id = 0 + + self.scanning = True + + def _stop(): + self.scanning = False + self._scan_timeout_id = 0 + return False + + self._scan_timeout_id = GLib.timeout_add(duration_ms, _stop) + + def stop_scan(self): + if self._scan_timeout_id: + GLib.source_remove(self._scan_timeout_id) + self._scan_timeout_id = 0 + self.scanning = False + + def toggle_power(self): + self.powered = not self.powered + + def toggle_scan(self): + if self.scanning: + self.stop_scan() + else: + self.scan() + + def close(self): + if self._scan_timeout_id: + GLib.source_remove(self._scan_timeout_id) + self._scan_timeout_id = 0 + if self._prop_sub_id: + self._bus.signal_unsubscribe(self._prop_sub_id) + self._prop_sub_id = 0 + for device in list(self._devices.values()): + device.close() + self._devices.clear() + + def notifier(self, name: str, *args): + self.notify(name) + self.emit("changed") + + +class BluetoothClient(Service): + @Signal + def changed(self) -> None: ... + + @Signal + def closed(self) -> None: ... + + @Signal + def device_added(self, address: str) -> None: ... + + @Signal + def device_removed(self, address: str) -> None: ... + + @Signal + def adapter_added(self, path: str) -> None: ... + + @Signal + def adapter_removed(self, path: str) -> None: ... + + @Property(list, "readable") + def adapters(self) -> list: + return list(self._adapters.values()) + + @Property(list, "readable") + def devices(self) -> list: + return [d for a in self._adapters.values() for d in a.devices] + + @Property(list, "readable") + def connected_devices(self) -> list: + return [d for d in self.devices if d.connected] + + @Property(str, "readable") + def state(self) -> str: + for a in self._adapters.values(): + if a.powered: + return a.state + return "absent" if not self._adapters else "off" + + @Property(bool, "read-write", default_value=False) + def enabled(self) -> bool: + return any(a.powered for a in self._adapters.values()) + + @enabled.setter + def enabled(self, value: bool): + for a in self._adapters.values(): + a.powered = value + + @Property(bool, "read-write", default_value=False) + def scanning(self) -> bool: + return any(a.scanning for a in self._adapters.values()) + + @scanning.setter + def scanning(self, value: bool): + for a in self._adapters.values(): + if a.powered: + a.scanning = value + + @Property(bool, "read-write", "is-powered", default_value=False) + def powered(self) -> bool: + return self.enabled + + @powered.setter + def powered(self, value: bool): + self.enabled = value + + def __init__(self, **kwargs): + super().__init__(**kwargs) + self._adapters: dict[str, BluetoothAdapter] = {} + self._bus: Gio.DBusConnection | None = None + self._om_proxy: Gio.DBusProxy | None = None + self._iface_added_sub: int = 0 + self._iface_removed_sub: int = 0 + + self._bus = Gio.bus_get_sync(Gio.BusType.SYSTEM, None) + self._om_proxy = Gio.DBusProxy.new_sync( + self._bus, + Gio.DBusProxyFlags.NONE, + None, + BLUEZ_SERVICE, + "/", + DBUS_OM_IFACE, + None, + ) + + self._iface_added_sub = self._bus.signal_subscribe( + BLUEZ_SERVICE, + DBUS_OM_IFACE, + "InterfacesAdded", + None, + None, + Gio.DBusSignalFlags.NONE, + self._on_interfaces_added, + None, + ) + self._iface_removed_sub = self._bus.signal_subscribe( + BLUEZ_SERVICE, + DBUS_OM_IFACE, + "InterfacesRemoved", + None, + None, + Gio.DBusSignalFlags.NONE, + self._on_interfaces_removed, + None, + ) + + self._populate_from_object_manager() + + def _populate_from_object_manager(self): + try: + result = self._om_proxy.call_sync( + "GetManagedObjects", None, Gio.DBusCallFlags.NONE, 5000, None + ) + objects = _unpack_variant(result)[0] + except Exception as e: + logger.warning(f"[Bluetooth] GetManagedObjects failed: {e}") + return + + for path, ifaces in objects.items(): + if BLUEZ_ADAPTER_IFACE in ifaces: + self._add_adapter(path, ifaces[BLUEZ_ADAPTER_IFACE]) + + for path, ifaces in objects.items(): + if BLUEZ_DEVICE_IFACE in ifaces: + self._add_device(path, ifaces[BLUEZ_DEVICE_IFACE]) + + def _add_adapter(self, path: str, props: dict): + if path in self._adapters: + return + logger.info(f"[Bluetooth] Adding adapter: {path}") + adapter = BluetoothAdapter( + self._bus, + path, + props, + on_changed=lambda *_: self.emit("changed"), + ) + adapter.connect( + "device-added", + lambda a, addr: ( + self.emit("device-added", addr), + self.notify("devices"), + self.notify("connected-devices"), + self.emit("changed"), + ), + ) + adapter.connect( + "device-removed", + lambda a, addr: ( + self.emit("device-removed", addr), + self.notify("devices"), + self.notify("connected-devices"), + self.emit("changed"), + ), + ) + self._adapters[path] = adapter + self.emit("adapter-added", path) + self.notify("adapters") + self.emit("changed") + + def _add_device(self, path: str, props: dict): + adapter_path = _device_path_to_adapter_path(path) + adapter = self._adapters.get(adapter_path) + if not adapter: + logger.warning(f"[Bluetooth] No adapter found for device at {path}") + return + + adapter.add_device(path, props) + + def _on_interfaces_added( + self, _conn, _sender, _path, _iface, _signal, params, _data + ): + unpacked = _unpack_variant(params) + obj_path: str = unpacked[0] + ifaces: dict = unpacked[1] + + if BLUEZ_ADAPTER_IFACE in ifaces: + self._add_adapter(obj_path, ifaces[BLUEZ_ADAPTER_IFACE]) + if BLUEZ_DEVICE_IFACE in ifaces: + self._add_device(obj_path, ifaces[BLUEZ_DEVICE_IFACE]) + + def _on_interfaces_removed( + self, _conn, _sender, _path, _iface, _signal, params, _data + ): + unpacked = _unpack_variant(params) + obj_path: str = unpacked[0] + ifaces: list = unpacked[1] + + if BLUEZ_ADAPTER_IFACE in ifaces: + adapter = self._adapters.pop(obj_path, None) + if adapter: + logger.info(f"[Bluetooth] Removing adapter: {obj_path}") + adapter.close() + self.emit("adapter-removed", obj_path) + self.notify("adapters") + self.emit("changed") + + if BLUEZ_DEVICE_IFACE in ifaces: + adapter_path = _device_path_to_adapter_path(obj_path) + adapter = self._adapters.get(adapter_path) + if adapter: + adapter.remove_device(obj_path) + + def scan(self): + for a in self._adapters.values(): + if a.powered: + a.scan() + + def toggle_power(self): + self.enabled = not self.enabled + + def toggle_scan(self): + for a in self._adapters.values(): + if a.powered: + a.toggle_scan() + + def get_device(self, address: str) -> BluetoothDevice | None: + for adapter in self._adapters.values(): + if d := adapter.get_device(address): + return d + return None + + def get_adapter(self, path: str) -> BluetoothAdapter | None: + return self._adapters.get(path) + + def connect_device( + self, + device: BluetoothDevice, + connect: bool = True, + callback: Callable[Concatenate[bool, P], Any] | None = None, + *args: P.args, + **kwargs: P.kwargs, + ): + return device.connect_device(connect, callback, *args, **kwargs) + + def notifier(self, name: str, *args): + self.notify(name) + self.emit("changed") + + +def _device_path_to_adapter_path(device_path: str) -> str: + parts = device_path.rsplit("/", 1) + return parts[0] if len(parts) == 2 else device_path + + +def _path_to_addr(object_path: str) -> str: + return object_path.split("/")[-1][4:].replace("_", ":") + + +def _icon_to_type(icon: str) -> str: + return { + "audio-headset": "Headset", + "audio-headphones": "Headphones", + "audio-speakers": "Speaker", + "audio-card": "Speaker", + "input-keyboard": "Keyboard", + "input-mouse": "Mouse", + "input-gaming": "Joypad", + "phone": "Phone", + "printer": "Printer", + }.get(icon, icon) diff --git a/src/services/network.py b/src/services/network.py index 7a010c3..76c79f8 100644 --- a/src/services/network.py +++ b/src/services/network.py @@ -73,13 +73,22 @@ class Wifi(Service): self._client.wireless_set_enabled(not self._client.wireless_get_enabled()) def scan(self): - self._device.request_scan_async( - None, - lambda device, result: [ - device.request_scan_finish(result), - self.emit("changed"), - ], - ) + if hasattr(self, "_scanning") and self._scanning: + return + self._scanning = True + + def _on_scan_finish(device, result): + self._scanning = False + try: + device.request_scan_finish(result) + except Exception as e: + logger.debug(f"Scan finished with error: {e}") + self.emit("changed") + + try: + self._device.request_scan_async(None, _on_scan_finish) + except Exception: + self._scanning = False def notifier(self, name: str, *args): self.notify(name) @@ -339,9 +348,10 @@ class NetworkClient(Service): return None def is_network_saved(self, ssid: str) -> bool: - if not self._client: + client = self._client or NM.Client.new(None) + if not client: return False - for conn in self._client.get_connections(): + for conn in client.get_connections(): s_wifi = conn.get_setting_wireless() if s_wifi: saved_ssid = s_wifi.get_ssid() @@ -350,7 +360,43 @@ class NetworkClient(Service): return True return False - def connect_wifi_bssid(self, bssid: str): + def connect_wifi_bssid(self, bssid: str, callback=None): + if not self._client or not self.wifi_device: + if callback: + callback(False, "No wifi device") + return + + device = self.wifi_device._device + handler_id: list[int] = [] + + def _on_state_changed(dev, new_state, old_state, reason): + if not callback: + return + state = NM.DeviceState(new_state) + if state == NM.DeviceState.ACTIVATED: + _cleanup() + callback(True, "Connected") + elif state in (NM.DeviceState.FAILED, NM.DeviceState.DISCONNECTED): + _cleanup() + reason_str = NM.DeviceStateReason(reason).value_nick + if "secret" in reason_str or "auth" in reason_str: + callback(False, "Wrong password") + else: + callback(False, f"Failed ({reason_str})") + elif state == NM.DeviceState.NEED_AUTH: + _cleanup() + callback(False, "Wrong password") + + def _cleanup(): + if handler_id: + try: + device.disconnect_by_func(_on_state_changed) + except Exception: + pass + + if callback: + handler_id.append(device.connect("state-changed", _on_state_changed)) + exec_shell_command_async( f"nmcli device wifi connect {bssid}", lambda *args: logger.debug(f"connect result: {args}"),