diff --git a/src/services/network.py b/src/services/network.py index 6f87c36..16767cb 100644 --- a/src/services/network.py +++ b/src/services/network.py @@ -1,302 +1,773 @@ -import shlex -from types import SimpleNamespace +import subprocess +from dataclasses import dataclass, field from typing import List, Literal -import gi from fabric.core.service import Property, Service, Signal -from fabric.utils import Gio, GLib, bulk_connect, exec_shell_command_async, logger +from fabric.utils import Gio, GLib, logger -try: - gi.require_version("NM", "1.0") - from gi.repository import NM -except (ValueError, GLib.Error, ImportError) as e: - logger.error(f"Failed to start network manager: {e}") - NM = None +from services.bluetooth import _rfkill_soft_blocked + +NM_SERVICE = "org.freedesktop.NetworkManager" +NM_PATH = "/org/freedesktop/NetworkManager" +NM_SETTINGS_PATH = "/org/freedesktop/NetworkManager/Settings" +NM_DEVICE_PATH = "/org/freedesktop/NetworkManager/Device" +NM_WIRELESS_PATH = "/org/freedesktop/NetworkManager/Device/Wireless" +NM_AP_PATH = "/org/freedesktop/NetworkManager/AccessPoint" +NM_ACTIVE_CONN_PATH = "/org/freedesktop/NetworkManager/Connection/Active" +NM_CONN_SETTINGS_PATH = "/org/freedesktop/NetworkManager/Settings/Connection" + +NM_IFACE = "org.freedesktop.NetworkManager" +NM_DEVICE_IFACE = "org.freedesktop.NetworkManager.Device" +NM_WIRELESS_IFACE = "org.freedesktop.NetworkManager.Device.Wireless" +NM_AP_IFACE = "org.freedesktop.NetworkManager.AccessPoint" +NM_SETTINGS_IFACE = "org.freedesktop.NetworkManager.Settings" +NM_ACTIVE_CONN_IFACE = "org.freedesktop.NetworkManager.Connection.Active" +NM_CONN_SETTINGS_IFACE = "org.freedesktop.NetworkManager.Settings.Connection" + +# DeviceType constants +DEVICE_TYPE_ETHERNET = 1 +DEVICE_TYPE_WIFI = 2 + +# DeviceState constants +DEVICE_STATE_UNMANAGED = 10 +DEVICE_STATE_UNAVAILABLE = 20 +DEVICE_STATE_DISCONNECTED = 30 +DEVICE_STATE_PREPARE = 40 +DEVICE_STATE_CONFIG = 50 +DEVICE_STATE_NEED_AUTH = 60 +DEVICE_STATE_IP_CONFIG = 70 +DEVICE_STATE_IP_CHECK = 80 +DEVICE_STATE_SECONDARIES = 90 +DEVICE_STATE_ACTIVATED = 100 +DEVICE_STATE_DEACTIVATING = 110 +DEVICE_STATE_FAILED = 120 + +# ActiveConnectionState constants +AC_STATE_ACTIVATED = 2 +AC_STATE_ACTIVATING = 1 +AC_STATE_DEACTIVATING = 3 +AC_STATE_DEACTIVATED = 0 + +# ConnectivityState constants +CONNECTIVITY_NONE = 1 +CONNECTIVITY_PORTAL = 2 +CONNECTIVITY_LIMITED = 3 +CONNECTIVITY_FULL = 4 + +_DEVICE_STATE_NAMES = { + DEVICE_STATE_UNMANAGED: "unmanaged", + DEVICE_STATE_UNAVAILABLE: "unavailable", + DEVICE_STATE_DISCONNECTED: "disconnected", + DEVICE_STATE_PREPARE: "prepare", + DEVICE_STATE_CONFIG: "config", + DEVICE_STATE_NEED_AUTH: "need_auth", + DEVICE_STATE_IP_CONFIG: "ip_config", + DEVICE_STATE_IP_CHECK: "ip_check", + DEVICE_STATE_SECONDARIES: "secondaries", + DEVICE_STATE_ACTIVATED: "activated", + DEVICE_STATE_DEACTIVATING: "deactivating", + DEVICE_STATE_FAILED: "failed", +} + +_AC_STATE_NAMES = { + AC_STATE_ACTIVATED: "activated", + AC_STATE_ACTIVATING: "activating", + AC_STATE_DEACTIVATING: "deactivating", + AC_STATE_DEACTIVATED: "deactivated", +} + +_CONNECTIVITY_NAMES = { + CONNECTIVITY_NONE: "none", + CONNECTIVITY_PORTAL: "portal", + CONNECTIVITY_LIMITED: "loss", + CONNECTIVITY_FULL: "full", +} + + +def _make_proxy(bus: Gio.DBusConnection, path: str, iface: str) -> Gio.DBusProxy: + return Gio.DBusProxy.new_sync( + bus, + Gio.DBusProxyFlags.NONE, + None, + NM_SERVICE, + path, + iface, + None, + ) + + +def _get(proxy: Gio.DBusProxy, prop: str): + v = proxy.get_cached_property(prop) + if v is None: + return None + return v.unpack() if isinstance(v, GLib.Variant) else v + + +def _set(proxy: Gio.DBusProxy, prop: str, value): + proxy.set_cached_property(prop, value) + + +def _call(proxy: Gio.DBusProxy, method: str, args=None, flags=0): + try: + result = proxy.call_sync(method, args, flags, -1, None) + return result.unpack() if result else None + except GLib.Error as e: + logger.debug(f"[Network] D-Bus call {method} failed: {e}") + return None + + +def _ssid_bytes_to_str(data) -> str: + if data is None: + return "Unknown" + if isinstance(data, (bytes, bytearray)): + return data.decode("utf-8", errors="replace") or "Unknown" + if isinstance(data, (list, tuple)): + return bytes(data).decode("utf-8", errors="replace") or "Unknown" + if isinstance(data, str): + return data or "Unknown" + return "Unknown" + + +def _signal_icon(strength: int) -> str: + return { + 80: "network-wireless-signal-excellent-symbolic", + 60: "network-wireless-signal-good-symbolic", + 40: "network-wireless-signal-ok-symbolic", + 20: "network-wireless-signal-weak-symbolic", + 0: "network-wireless-signal-none-symbolic", + }.get( + min(80, 20 * round(strength / 20)), + "network-wireless-no-route-symbolic", + ) + + +@dataclass +class AccessPointData: + ssid: str + bssid: str | None + strength: int + max_bitrate: int = 0 + frequency: int = 0 + state: str = "unknown" + requires_password: bool = False + is_active: bool = False + working: bool = False + icon: str = "" + path: str = "/" + device_path: str = "/" + + @staticmethod + def is_better( + max_bitrate1: int, + frequency1: int, + strength1: int, + max_bitrate2: int, + frequency2: int, + strength2: int, + ) -> bool: + if max_bitrate1 > max_bitrate2: + return True + if max_bitrate1 == max_bitrate2: + if frequency1 > frequency2: + return True + if frequency1 == frequency2: + return strength1 > strength2 + return False + + +@dataclass +class ActiveConnectionInfo: + name: str + connection_type: Literal["wired", "wifi", "vpn"] = "wired" + strength: int = -1 + object_path: str = "" + + @property + def is_vpn(self) -> bool: + return self.connection_type == "vpn" + + +@dataclass +class Vpn: + name: str + path: str = "" + + +@dataclass +class KnownConnection: + connection_type: Literal["access_point", "vpn"] = "access_point" + access_point: AccessPointData | None = None + vpn: Vpn | None = None + + +@dataclass +class NetworkData: + wifi_present: bool = False + wireless_access_points: list = field(default_factory=list) + active_connections: list = field(default_factory=list) + known_connections: list = field(default_factory=list) + wifi_enabled: bool = False + airplane_mode: bool = False + connectivity: str = "unknown" + scanning_nearby_wifi: bool = False + + +class NmProxy: + """Thin wrapper around Gio.DBusProxy for org.freedesktop.NetworkManager. + + Mirrors the Rust NetworkDbus struct from network/dbus.rs. + """ + + def __init__(self, bus: Gio.DBusConnection): + self._bus = bus + self._nm = _make_proxy(bus, NM_PATH, NM_IFACE) + self._settings = _make_proxy(bus, NM_SETTINGS_PATH, NM_SETTINGS_IFACE) + + # -- NetworkManager properties -- + + @property + def wireless_enabled(self) -> bool: + return bool(_get(self._nm, "WirelessEnabled")) + + @wireless_enabled.setter + def wireless_enabled(self, value: bool): + self._nm.set_cached_property("WirelessEnabled", GLib.Variant("b", value)) + _call( + self._nm, + "org.freedesktop.DBus.Properties.Set", + GLib.Variant( + "(ssv)", + (NM_IFACE, "WirelessEnabled", GLib.Variant("b", value)), + ), + ) + + @property + def connectivity(self) -> str: + return _CONNECTIVITY_NAMES.get(_get(self._nm, "Connectivity") or 0, "unknown") + + @property + def devices(self) -> list[str]: + return list(_get(self._nm, "Devices") or []) + + @property + def active_connection_paths(self) -> list[str]: + return list(_get(self._nm, "ActiveConnections") or []) + + # -- Methods (mirrors Rust proxy methods) -- + + def activate_connection( + self, conn_path: str, device_path: str, specific_path: str = "/" + ) -> str | None: + return _call( + self._nm, + "ActivateConnection", + GLib.Variant("(sss)", (conn_path, device_path, specific_path)), + ) + + def add_and_activate_connection( + self, settings: dict, device_path: str, ap_path: str = "/" + ) -> tuple | None: + return _call( + self._nm, + "AddAndActivateConnection", + GLib.Variant("(sa{sv}ss)", ("", settings, device_path, ap_path)), + ) + + def deactivate_connection(self, active_conn_path: str) -> bool: + return ( + _call( + self._nm, + "DeactivateConnection", + GLib.Variant("(o)", (active_conn_path,)), + ) + is not None + ) + + # Device enumeration + + def wifi_devices(self) -> list[str]: + return [d for d in self.devices if self._device_type(d) == DEVICE_TYPE_WIFI] + + def ethernet_devices(self) -> list[str]: + return [d for d in self.devices if self._device_type(d) == DEVICE_TYPE_ETHERNET] + + def wifi_device_present(self) -> bool: + return len(self.wifi_devices()) > 0 + + def _device_type(self, path: str) -> int: + proxy = _make_proxy(self._bus, path, NM_DEVICE_IFACE) + return _get(proxy, "DeviceType") or 0 + + # Access points + + def wireless_access_points(self) -> list[AccessPointData]: + """Get all wireless access points across all wifi devices. + + Port of dbus.rs wireless_access_points(). + """ + aps: dict[str, AccessPointData] = {} + for dev_path in self.wifi_devices(): + wdev = _make_proxy(self._bus, dev_path, NM_WIRELESS_IFACE) + ap_paths = _get(wdev, "AccessPoints") or [] + device_state = self._device_state_name(dev_path) + active_ap_path = _get(wdev, "ActiveAccessPoint") + + for ap_path in ap_paths: + is_active = ap_path == active_ap_path + ap = self._read_ap(ap_path, dev_path, device_state, is_active) + if ap is None: + continue + # Deduplicate by ssid, keep best (mirrors Rust logic) + existing = aps.get(ap.ssid) + if existing and AccessPointData.is_better( + existing.max_bitrate, + existing.frequency, + existing.strength, + ap.max_bitrate, + ap.frequency, + ap.strength, + ): + continue + aps[ap.ssid] = ap + + return sorted(aps.values(), key=lambda a: a.strength, reverse=True) + + def _read_ap( + self, ap_path: str, dev_path: str, device_state: str, is_active: bool = False + ) -> AccessPointData | None: + try: + proxy = _make_proxy(self._bus, ap_path, NM_AP_IFACE) + ssid_data = _get(proxy, "Ssid") + ssid = _ssid_bytes_to_str(ssid_data) + if ssid == "Unknown": + return None + strength = _get(proxy, "Strength") or 0 + flags = _get(proxy, "Flags") or 0 + return AccessPointData( + ssid=ssid, + bssid=None, + strength=strength, + max_bitrate=_get(proxy, "MaxBitrate") or 0, + frequency=_get(proxy, "Frequency") or 0, + state=device_state, + requires_password=flags != 0, + icon=_signal_icon(strength), + path=ap_path, + device_path=dev_path, + is_active=is_active, + ) + except GLib.Error as e: + logger.debug(f"[Network] Failed to read AP {ap_path}: {e}") + return None + + def _device_state_name(self, dev_path: str) -> str: + proxy = _make_proxy(self._bus, dev_path, NM_DEVICE_IFACE) + state = _get(proxy, "State") or 0 + return _DEVICE_STATE_NAMES.get(state, "unknown") + + def active_access_point_path(self, dev_path: str) -> str | None: + wdev = _make_proxy(self._bus, dev_path, NM_WIRELESS_IFACE) + return _get(wdev, "ActiveAccessPoint") + + def active_connections_info(self) -> list[ActiveConnectionInfo]: + """Get info about active connections. + + Port of dbus.rs active_connections_info(). + """ + result: list[ActiveConnectionInfo] = [] + for ac_path in self.active_connection_paths: + try: + ac = _make_proxy(self._bus, ac_path, NM_ACTIVE_CONN_IFACE) + conn_type = _get(ac, "Type") or "" + name = _get(ac, "Id") or "" + + if _get(ac, "Vpn"): + result.append( + ActiveConnectionInfo( + name=name, connection_type="vpn", object_path=ac_path + ) + ) + continue + + is_vpn = "vpn" in conn_type.lower() + if is_vpn: + result.append( + ActiveConnectionInfo( + name=name, connection_type="vpn", object_path=ac_path + ) + ) + continue + + is_wireless = "wireless" in conn_type + if is_wireless: + strength = -1 + for dev_path in _get(ac, "Devices") or []: + if self._device_type(dev_path) == DEVICE_TYPE_WIFI: + ap_path = self.active_access_point_path(dev_path) + if ap_path: + ap = _make_proxy(self._bus, ap_path, NM_AP_IFACE) + strength = _get(ap, "Strength") or 0 + result.append( + ActiveConnectionInfo( + name=name, connection_type="wifi", strength=strength + ) + ) + elif "wireguard" in conn_type.lower(): + result.append( + ActiveConnectionInfo( + name=name, connection_type="vpn", object_path=ac_path + ) + ) + else: + result.append( + ActiveConnectionInfo(name=name, connection_type="wired") + ) + except GLib.Error as e: + logger.debug( + f"[Network] Failed to read active connection {ac_path}: {e}" + ) + + result.sort( + key=lambda c: {"vpn": 0, "wired": 1, "wifi": 2}.get(c.connection_type, 3) + ) + return result + + def request_scan(self, dev_path: str) -> bool: + try: + wdev = _make_proxy(self._bus, dev_path, NM_WIRELESS_IFACE) + wdev.call_sync( + "RequestScan", + GLib.Variant("(a{sv})", ({},)), + Gio.DBusCallFlags.NONE, + -1, + None, + ) + return True + except GLib.Error as e: + logger.warning(f"[Network] Scan request failed for {dev_path}: {e}") + return False + + def scan_nearby_wifi(self) -> list[str]: + """Request scan on all wireless devices. Returns paths of devices that accepted.""" + requested: list[str] = [] + for dev_path in self.wifi_devices(): + if self.request_scan(dev_path): + requested.append(dev_path) + return requested + + # Known connections + + def known_connections( + self, wireless_aps: list[AccessPointData] | None = None + ) -> list[KnownConnection]: + known_ssid: list[str] = [] + known_vpn: list[Vpn] = [] + conn_paths = _get(self._settings, "Connections") or [] + + for conn_path in conn_paths: + try: + cs = _make_proxy(self._bus, conn_path, NM_CONN_SETTINGS_IFACE) + settings = _call(cs, "GetSettings") + if not settings: + continue + # settings is a{sa{sv}} — dict of section -> dict of key -> variant + if isinstance(settings, tuple) and settings: + settings = settings[0] + sections = settings if isinstance(settings, dict) else {} + + if "802-11-wireless" in sections: + conn_section = sections.get("connection", {}) + conn_id = conn_section.get("id") + if isinstance(conn_id, GLib.Variant): + conn_id = conn_id.unpack() + if conn_id: + known_ssid.append(str(conn_id)) + elif "vpn" in sections or "wireguard" in sections: + conn_section = sections.get("connection", {}) + conn_id = conn_section.get("id") + if isinstance(conn_id, GLib.Variant): + conn_id = conn_id.unpack() + if conn_id: + known_vpn.append(Vpn(name=str(conn_id), path=conn_path)) + except GLib.Error as e: + logger.debug(f"[Network] Failed to read connection {conn_path}: {e}") + + if wireless_aps is None: + wireless_aps = self.wireless_access_points() + + result: list[KnownConnection] = [] + for ap in wireless_aps: + if ap.ssid in known_ssid: + result.append( + KnownConnection(connection_type="access_point", access_point=ap) + ) + for vpn in known_vpn: + result.append(KnownConnection(connection_type="vpn", vpn=vpn)) + return result + + def find_connection(self, name: str) -> str | None: + """Find a connection path by SSID/id.""" + conn_paths = _get(self._settings, "Connections") or [] + for conn_path in conn_paths: + try: + cs = _make_proxy(self._bus, conn_path, NM_CONN_SETTINGS_IFACE) + settings = _call(cs, "GetSettings") + if not settings: + continue + if isinstance(settings, tuple) and settings: + settings = settings[0] + sections = settings if isinstance(settings, dict) else {} + conn_section = sections.get("connection", {}) + conn_id = conn_section.get("id") + if isinstance(conn_id, GLib.Variant): + conn_id = conn_id.unpack() + if conn_id == name: + return conn_path + except GLib.Error: + continue + return None + + def update_connection_password(self, conn_path: str, password: str): + """Update the PSK in an existing connection.""" + try: + cs = _make_proxy(self._bus, conn_path, NM_CONN_SETTINGS_IFACE) + settings = _call(cs, "GetSettings") + if not settings: + return + if isinstance(settings, tuple) and settings: + settings = settings[0] + sections = dict(settings) if isinstance(settings, dict) else {} + sec = dict(sections.get("802-11-wireless-security", {})) + sec["psk"] = GLib.Variant("s", password) + sections["802-11-wireless-security"] = sec + _call(cs, "Update", GLib.Variant("(a{sa{sv}})", (sections,))) + except GLib.Error as e: + logger.warning(f"[Network] Failed to update connection password: {e}") class Wifi(Service): - """A service to manage the wifi connection.""" + """A service to manage the wifi connection via D-Bus.""" @Signal def changed(self) -> None: ... - def __init__(self, client: NM.Client, device: NM.DeviceWifi, **kwargs): - self._client: NM.Client = client - self._device: NM.DeviceWifi = device - self._ap: NM.AccessPoint | None = None - self._ap_signal: int | None = None + def __init__(self, nm: NmProxy, device_path: str, **kwargs): + self._nm = nm + self._device_path = device_path + self._ap_path: str | None = None + self._handler_id: int | None = None + self._nm_handler_id: int | None = None + self._device_handler_id: int | None = None super().__init__(**kwargs) - self._client_handler = self._client.connect( - "notify::wireless-enabled", - lambda *args: self.notifier("enabled", args), + # Watch for active access point changes on this device + wdev = _make_proxy(nm._bus, device_path, NM_WIRELESS_IFACE) + self._handler_id = wdev.connect( + "g-properties-changed", self._on_device_props_changed ) - if self._device: - self._device_handlers = bulk_connect( - self._device, - { - "notify::active-access-point": lambda *args: self._activate_ap(), - "access-point-added": lambda *args: self.emit("changed"), - "access-point-removed": lambda *args: self.emit("changed"), - "state-changed": lambda *args: self.ap_update(), - }, - ) - self._activate_ap() + + # Watch WirelessEnabled on the NM main proxy so the toggle stays in sync + self._nm_handler_id = nm._nm.connect( + "g-properties-changed", self._on_nm_props_changed + ) + + self._refresh_active_ap() def close(self): - """Disconnect all signal handlers so the service can be garbage collected.""" - try: - self._client.disconnect(self._client_handler) - except Exception as e: - logger.warning( - f"[network] self._client.disconnect(self._client_handler) failed: {e}" - ) - for hid in getattr(self, "_device_handlers", ()): + if self._nm_handler_id is not None: try: - self._device.disconnect(hid) - except Exception as e: - logger.warning(f"[network] self._device.disconnect(hid) failed: {e}") - if getattr(self, "_ap", None) and getattr(self, "_ap_signal", None): + self._nm._nm.disconnect(self._nm_handler_id) + except Exception: + pass + self._nm_handler_id = None + if self._handler_id is not None: try: - self._ap.disconnect(self._ap_signal) - except Exception as e: - logger.warning( - f"[network] self._ap.disconnect(self._ap_signal) failed: {e}" - ) + wdev = _make_proxy(self._nm._bus, self._device_path, NM_WIRELESS_IFACE) + wdev.disconnect(self._handler_id) + except Exception: + pass + self._handler_id = None - def ap_update(self): - self.emit("changed") - for sn in [ - "enabled", - "internet", - "strength", - "frequency", - "access-points", - "ssid", - "state", - "icon-name", - ]: - self.notify(sn) - - def _activate_ap(self): - if self._ap: - self._ap.disconnect(self._ap_signal) - self._ap = self._device.get_active_access_point() - if not self._ap: - return - - self._ap_signal = self._ap.connect( - "notify::strength", lambda *args: self.ap_update() - ) - - def toggle_wifi(self): - self._client.wireless_set_enabled(not self._client.wireless_get_enabled()) - - def scan(self): - 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}") + def _on_nm_props_changed(self, proxy, changed, invalidated): + """React to WirelessEnabled changes on the NM main proxy.""" + if "WirelessEnabled" in changed: + self.notify("enabled") self.emit("changed") - try: - self._device.request_scan_async(None, _on_scan_finish) - except Exception as e: - logger.warning(f"[network] request_scan_async failed: {e}") - self._scanning = False + def _on_device_props_changed(self, proxy, changed, invalidated): + if "ActiveAccessPoint" in changed or "AccessPoints" in changed: + self._refresh_active_ap() + self.emit("changed") + for sn in ( + "strength", + "frequency", + "access-points", + "ssid", + "internet", + "icon-name", + ): + self.notify(sn) - def notifier(self, name: str, *args): - self.notify(name) + def _refresh_active_ap(self): + self._ap_path = self._nm.active_access_point_path(self._device_path) + + def _get_active_ap_prop(self, prop): + if not self._ap_path: + return None + try: + proxy = _make_proxy(self._nm._bus, self._ap_path, NM_AP_IFACE) + return _get(proxy, prop) + except GLib.Error: + return None + + def get_interface_name(self) -> str | None: + """Return the kernel interface name (e.g. 'wlan0') for this device.""" + try: + proxy = _make_proxy(self._nm._bus, self._device_path, NM_DEVICE_IFACE) + return _get(proxy, "Interface") or None + except GLib.Error: + return None + + def toggle_wifi(self): + self._nm.wireless_enabled = not self._nm.wireless_enabled + + def scan(self): + if getattr(self, "_scanning", False): + return + self._scanning = True + self.notify("scanning") self.emit("changed") - return + + def _finish(): + self._scanning = False + self.notify("scanning") + self.emit("changed") + return False + + def _do_scan(): + self._nm.request_scan(self._device_path) + GLib.timeout_add(3000, _finish) + return False + + GLib.idle_add(_do_scan) @Property(bool, "read-write", default_value=False) def enabled(self) -> bool: - return bool(self._client.wireless_get_enabled()) + return self._nm.wireless_enabled @enabled.setter def enabled(self, value: bool): - self._client.wireless_set_enabled(value) + self._nm.wireless_enabled = value @Property(int, "readable") - def strength(self): - return self._ap.get_strength() if self._ap else -1 + def strength(self) -> int: + v = self._get_active_ap_prop("Strength") + return v if v is not None else -1 @Property(str, "readable") - def icon_name(self): - if not self._ap: + def icon_name(self) -> str: + if not self._ap_path: return "network-wireless-disabled-symbolic" - if self.internet == "activated": - return { - 80: "network-wireless-signal-excellent-symbolic", - 60: "network-wireless-signal-good-symbolic", - 40: "network-wireless-signal-ok-symbolic", - 20: "network-wireless-signal-weak-symbolic", - 00: "network-wireless-signal-none-symbolic", - }.get( - min(80, 20 * round(self._ap.get_strength() / 20)), - "network-wireless-no-route-symbolic", - ) + return _signal_icon(self.strength) if self.internet == "activating": return "network-wireless-acquiring-symbolic" - return "network-wireless-offline-symbolic" @Property(int, "readable") - def frequency(self): - return self._ap.get_frequency() if self._ap else -1 + def frequency(self) -> int: + v = self._get_active_ap_prop("Frequency") + return v if v is not None else -1 @Property(str, "readable") - def internet(self): - conn = self._device.get_active_connection() - if not conn: - return "unknown" - return { - NM.ActiveConnectionState.ACTIVATED: "activated", - NM.ActiveConnectionState.ACTIVATING: "activating", - NM.ActiveConnectionState.DEACTIVATING: "deactivating", - NM.ActiveConnectionState.DEACTIVATED: "deactivated", - }.get(conn.get_state(), "unknown") + def internet(self) -> str: + ac_paths = self._nm.active_connection_paths + for ac_path in ac_paths: + try: + ac = _make_proxy(self._nm._bus, ac_path, NM_ACTIVE_CONN_IFACE) + devices = _get(ac, "Devices") or [] + if self._device_path not in devices: + continue + state = _get(ac, "State") or 0 + return _AC_STATE_NAMES.get(state, "unknown") + except GLib.Error: + continue + return "unknown" @Property(object, "readable") - def access_points(self) -> List[object]: - points: list[NM.AccessPoint] = self._device.get_access_points() - active_bssid = self._ap.get_bssid() if self._ap else None - - def make_ap(ap: NM.AccessPoint): - ssid_bytes = ap.get_ssid() - ssid = ( - NM.utils_ssid_to_utf8(ssid_bytes.get_data()) - if ssid_bytes - else "Unknown" - ) - bssid = ap.get_bssid() - strength = ap.get_strength() - is_active = bssid == active_bssid if active_bssid else False - flags = ap.get_rsn_flags() or ap.get_wpa_flags() - requires_password = bool(flags) - icon = { - 80: "network-wireless-signal-excellent-symbolic", - 60: "network-wireless-signal-good-symbolic", - 40: "network-wireless-signal-ok-symbolic", - 20: "network-wireless-signal-weak-symbolic", - 0: "network-wireless-signal-none-symbolic", - }.get( - min(80, 20 * round(strength / 20)), - "network-wireless-no-route-symbolic", - ) - return SimpleNamespace( - ssid=ssid, - bssid=bssid, - strength=strength, - is_active=is_active, - requires_password=requires_password, - icon=icon, - ) - - return list(map(make_ap, points)) + def access_points(self) -> List[AccessPointData]: + return self._nm.wireless_access_points() @Property(str, "readable") - def ssid(self): - if not self._ap: - return "Disconnected" - ssid = self._ap.get_ssid().get_data() - return NM.utils_ssid_to_utf8(ssid) if ssid else "Unknown" + def ssid(self) -> str: + ssid_data = self._get_active_ap_prop("Ssid") + return _ssid_bytes_to_str(ssid_data) if ssid_data else "Disconnected" @Property(int, "readable") - def state(self): - return { - NM.DeviceState.UNMANAGED: "unmanaged", - NM.DeviceState.UNAVAILABLE: "unavailable", - NM.DeviceState.DISCONNECTED: "disconnected", - NM.DeviceState.PREPARE: "prepare", - NM.DeviceState.CONFIG: "config", - NM.DeviceState.NEED_AUTH: "need_auth", - NM.DeviceState.IP_CONFIG: "ip_config", - NM.DeviceState.IP_CHECK: "ip_check", - NM.DeviceState.SECONDARIES: "secondaries", - NM.DeviceState.ACTIVATED: "activated", - NM.DeviceState.DEACTIVATING: "deactivating", - NM.DeviceState.FAILED: "failed", - }.get(self._device.get_state(), "unknown") + def state(self) -> str: + return self._nm._device_state_name(self._device_path) + + @Property(bool, "readable", default_value=False) + def scanning(self) -> bool: + return getattr(self, "_scanning", False) class Ethernet(Service): - """A service to manage the ethernet connection.""" + """A service to manage the ethernet connection via D-Bus.""" @Signal def changed(self) -> None: ... - @Property(int, "readable") - def speed(self) -> int: - return self._device.get_speed() + def __init__(self, nm: NmProxy, device_path: str, **kwargs) -> None: + super().__init__(**kwargs) + self._nm = nm + self._device_path = device_path + self._handler_id: int | None = None + dev = _make_proxy(nm._bus, device_path, NM_DEVICE_IFACE) + self._handler_id = dev.connect( + "g-properties-changed", lambda *_: self.emit("changed") + ) + + def close(self): + if self._handler_id is not None: + try: + dev = _make_proxy(self._nm._bus, self._device_path, NM_DEVICE_IFACE) + dev.disconnect(self._handler_id) + except Exception: + pass + self._handler_id = None @Property(int, "readable") - def internet(self): - conn = self._device.get_active_connection() - if not conn: - return "unknown" - return { - NM.ActiveConnectionState.ACTIVATED: "activated", - NM.ActiveConnectionState.ACTIVATING: "activating", - NM.ActiveConnectionState.DEACTIVATING: "deactivating", - NM.ActiveConnectionState.DEACTIVATED: "deactivated", - }.get(conn.get_state(), "unknown") + def speed(self) -> int: + dev = _make_proxy(self._nm._bus, self._device_path, NM_DEVICE_IFACE) + return _get(dev, "Speed") or 0 + + @Property(int, "readable") + def internet(self) -> str: + ac_paths = self._nm.active_connection_paths + for ac_path in ac_paths: + try: + ac = _make_proxy(self._nm._bus, ac_path, NM_ACTIVE_CONN_IFACE) + devices = _get(ac, "Devices") or [] + if self._device_path not in devices: + continue + state = _get(ac, "State") or 0 + return _AC_STATE_NAMES.get(state, "unknown") + except GLib.Error: + continue + return "unknown" @Property(str, "readable") def icon_name(self) -> str: - network = self.internet - if network == "activated": + internet = self.internet + if internet == "activated": return "network-wired-symbolic" - - elif network == "activating": + if internet == "activating": return "network-wired-acquiring-symbolic" - - elif self._device.get_connectivity() != NM.ConnectivityState.FULL: + if self._nm.connectivity != "full": return "network-wired-no-route-symbolic" - return "network-wired-disconnected-symbolic" - def __init__(self, client: NM.Client, device: NM.DeviceEthernet, **kwargs) -> None: - super().__init__(**kwargs) - self._client: NM.Client = client - self._device: NM.DeviceEthernet = device - - self._handlers: list[int] = [] - for pn in ( - "active-connection", - "icon-name", - "internet", - "speed", - "state", - ): - self._handlers.append( - self._device.connect(f"notify::{pn}", lambda *_: self.notifier(pn)) - ) - - self._handlers.append( - self._device.connect("notify::speed", lambda *_: self.notifier("speed")) - ) - - def notifier(self, pn): - self.notify(pn) - self.emit("changed") - - def close(self): - """Disconnect all signal handlers so the service can be garbage collected.""" - for hid in getattr(self, "_handlers", ()): - try: - self._device.disconnect(hid) - except Exception as e: - logger.warning(f"[network] self._device.disconnect(hid) failed: {e}") - class NetworkClient(Service): @Signal @@ -309,55 +780,88 @@ class NetworkClient(Service): def device_removed(self, iface: str) -> None: ... def __init__(self, **kwargs): - self._client: NM.Client | None = None + self._nm: NmProxy | None = None self.wifi_device: Wifi | None = None self.wifi_devices: dict[str, Wifi] = {} self.ethernet_device: Ethernet | None = None + self._airplane_mode: bool = False + self._nm_proxy_handler: int | None = None super().__init__(**kwargs) - NM.Client.new_async( - cancellable=None, - callback=self._init_network_client, + bus = Gio.bus_get_sync(Gio.BusType.SYSTEM, None) + if bus is None: + logger.error("[Network] Failed to connect to system bus") + return + self._nm = NmProxy(bus) + self._init_devices() + + def _init_devices(self): + if not self._nm: + return + + for dev_path in self._nm.devices: + self._handle_device_added(dev_path) + + # Watch for device additions/removals via PropertiesChanged on NM + self._nm_proxy_handler = self._nm._nm.connect( + "g-properties-changed", self._on_nm_props_changed ) - def _init_network_client(self, client: NM.Client, task: Gio.Task, **kwargs): - self._client = client - - for device in self._client.get_devices(): - self._handle_device_added(device) - - self._client.connect("device-added", lambda _, d: self._handle_device_added(d)) - self._client.connect( - "device-removed", lambda _, d: self._handle_device_removed(d) - ) + # Initial airplane mode + bluetooth_blocked = _rfkill_soft_blocked() + wifi_enabled = self._nm.wireless_enabled + self._airplane_mode = bluetooth_blocked and not wifi_enabled self.notify("primary-device") + self.notify("airplane-mode") - def _handle_device_added(self, nm_device): - dtype = nm_device.get_device_type() + def _on_nm_props_changed(self, proxy, changed, invalidated): + if "Devices" not in changed: + return + new_devices = set(self._nm.devices) + # Track by old wifi device paths + old_paths = {w._device_path for w in self.wifi_devices.values()} + if self.ethernet_device: + old_paths.add(self.ethernet_device._device_path) - if dtype == NM.DeviceType.WIFI: - iface = nm_device.get_iface() - if iface in self.wifi_devices: + new_paths = set(new_devices) + added = new_paths - old_paths + removed = old_paths - new_paths + + for dev_path in added: + self._handle_device_added(dev_path) + for dev_path in removed: + self._handle_device_removed(dev_path) + + def _handle_device_added(self, dev_path: str): + if not self._nm: + return + dtype = self._nm._device_type(dev_path) + iface = dev_path.split("/")[-1] + + if dtype == DEVICE_TYPE_WIFI: + if dev_path in {w._device_path for w in self.wifi_devices.values()}: return - wifi = Wifi(self._client, nm_device) - self.wifi_devices[iface] = wifi + wifi = Wifi(self._nm, dev_path) + self.wifi_devices[dev_path] = wifi if self.wifi_device is None: self.wifi_device = wifi logger.info(f"[Network] Wifi device added: {iface}") self.emit("device-added", iface) self.emit("device-ready") - elif dtype == NM.DeviceType.ETHERNET: + elif dtype == DEVICE_TYPE_ETHERNET: if self.ethernet_device is None: - self.ethernet_device = Ethernet(client=self._client, device=nm_device) + self.ethernet_device = Ethernet(self._nm, dev_path) self.emit("device-ready") - def _handle_device_removed(self, nm_device): - dtype = nm_device.get_device_type() + def _handle_device_removed(self, dev_path: str): + if not self._nm: + return + dtype = self._nm._device_type(dev_path) + iface = dev_path.split("/")[-1] - if dtype == NM.DeviceType.WIFI: - iface = nm_device.get_iface() - wifi = self.wifi_devices.pop(iface, None) + if dtype == DEVICE_TYPE_WIFI: + wifi = self.wifi_devices.pop(dev_path, None) if wifi is None: return if self.wifi_device is wifi: @@ -366,131 +870,221 @@ class NetworkClient(Service): logger.info(f"[Network] Wifi device removed: {iface}") self.emit("device-removed", iface) - elif dtype == NM.DeviceType.ETHERNET: + elif dtype == DEVICE_TYPE_ETHERNET: if self.ethernet_device is not None: self.ethernet_device.close() self.ethernet_device = None - self.emit("device-removed", nm_device.get_iface()) + self.emit("device-removed", iface) def _get_primary_device(self) -> Literal["wifi", "wired"] | None: - if not self._client: + if not self._nm: return None try: - conn_type = self._client.get_primary_connection().get_connection_type() - if "wireless" in str(conn_type): - return "wifi" - if "ethernet" in str(conn_type): - return "wired" + for ac_path in self._nm.active_connection_paths: + ac = _make_proxy(self._nm._bus, ac_path, NM_ACTIVE_CONN_IFACE) + conn_type = _get(ac, "Type") or "" + if "wireless" in conn_type: + return "wifi" + if "ethernet" in conn_type: + return "wired" except Exception as e: - logger.error(f"An error occurred: {e}") + logger.error(f"[Network] _get_primary_device failed: {e}") return None def is_network_saved(self, ssid: str) -> bool: - client = self._client or NM.Client.new(None) - if not client: + if not self._nm: return False - for conn in client.get_connections(): - s_wifi = conn.get_setting_wireless() - if s_wifi: - saved_ssid = s_wifi.get_ssid() - if saved_ssid: - if NM.utils_ssid_to_utf8(saved_ssid.get_data()) == ssid: - return True + conn_paths = _get(self._nm._settings, "Connections") or [] + for conn_path in conn_paths: + try: + cs = _make_proxy(self._nm._bus, conn_path, NM_CONN_SETTINGS_IFACE) + settings = _call(cs, "GetSettings") + if not settings: + continue + if isinstance(settings, tuple) and settings: + settings = settings[0] + sections = settings if isinstance(settings, dict) else {} + if "802-11-wireless" not in sections: + continue + conn_section = sections.get("connection", {}) + conn_id = conn_section.get("id") + if isinstance(conn_id, GLib.Variant): + conn_id = conn_id.unpack() + if conn_id == ssid: + return True + except GLib.Error: + continue return False - def connect_wifi_bssid(self, bssid: str, callback=None): - if not self._client or not self.wifi_device: + def connect_wifi(self, ap: AccessPointData, callback=None): + if not self._nm or not self.wifi_device: if callback: callback(False, "No wifi device") return - if not bssid: - if callback: - callback(False, "No bssid") - return - - device = self.wifi_device._device - handler_id: list[int] = [] - _fired = [False] - - def _on_state_changed(_dev, new_state, _old_state, reason): - if _fired[0]: - return - state = NM.DeviceState(new_state) - if state == NM.DeviceState.ACTIVATED: - _fired[0] = True - _cleanup() - if callback: - callback(True, "Connected") - elif state == NM.DeviceState.FAILED: - _fired[0] = True - _cleanup() - reason_str = NM.DeviceStateReason(reason).value_nick - if callback: - if "secret" in reason_str or "auth" in reason_str: - callback(False, "Wrong password") - else: - callback(False, f"Failed ({reason_str})") - - def _cleanup(): - if handler_id: - try: - device.disconnect_by_func(_on_state_changed) - except Exception as e: - logger.error(f"An error occurred: {e}") - - if callback: - handler_id.append(device.connect("state-changed", _on_state_changed)) - - exec_shell_command_async( - f"nmcli device wifi connect {shlex.quote(bssid)}", - lambda *args: logger.debug(f"connect result: {args}"), - ) - - def connect_wifi_with_password(self, bssid: str, password: str, callback): - if not self._client or not self.wifi_device: - callback(False, "No wifi device") - return - - if not bssid: - callback(False, "No bssid") - return - - device = self.wifi_device._device - handler_id: list[int] = [] - _fired = [False] - - def _on_state_changed(_dev, new_state, _old_state, reason): - if _fired[0]: - return - state = NM.DeviceState(new_state) - if state == NM.DeviceState.ACTIVATED: - _fired[0] = True - _cleanup() - callback(True, "Connected") - elif state in (NM.DeviceState.FAILED, NM.DeviceState.DISCONNECTED): - _fired[0] = True - _cleanup() - reason_str = NM.DeviceStateReason(reason).value_nick - if "secret" in reason_str or "auth" in reason_str: - callback(False, "Wrong password") + def _do_connect(): + try: + conn_path = self._nm.find_connection(ap.ssid) + if conn_path: + self._nm.activate_connection(conn_path, ap.device_path, ap.path) else: - callback(False, f"Failed ({reason_str})") - elif state == NM.DeviceState.NEED_AUTH: - _fired[0] = True - _cleanup() - callback(False, "Wrong password") + settings = { + "802-11-wireless": { + "ssid": GLib.Variant("ay", ap.ssid.encode("utf-8")) + }, + "connection": { + "id": GLib.Variant("s", ap.ssid), + "type": GLib.Variant("s", "802-11-wireless"), + }, + } + self._nm.add_and_activate_connection( + settings, ap.device_path, ap.path + ) + if callback: + GLib.idle_add(callback, True, "") + except Exception as e: + logger.error(f"[Network] Connect failed: {e}") + if callback: + GLib.idle_add(callback, False, str(e)) - def _cleanup(): - if handler_id: + import threading + + threading.Thread(target=_do_connect, daemon=True).start() + + def connect_wifi_with_password( + self, ap: AccessPointData, password: str, callback=None + ): + if not self._nm or not self.wifi_device: + if callback: + callback(False, "No wifi device") + return + + def _do_connect(): + try: + conn_path = self._nm.find_connection(ap.ssid) + if conn_path: + self._nm.update_connection_password(conn_path, password) + self._nm.activate_connection(conn_path, ap.device_path, ap.path) + else: + settings = { + "802-11-wireless": { + "ssid": GLib.Variant("ay", ap.ssid.encode("utf-8")) + }, + "connection": { + "id": GLib.Variant("s", ap.ssid), + "type": GLib.Variant("s", "802-11-wireless"), + }, + "802-11-wireless-security": { + "key-mgmt": GLib.Variant("s", "wpa-psk"), + "psk": GLib.Variant("s", password), + }, + } + self._nm.add_and_activate_connection( + settings, ap.device_path, ap.path + ) + if callback: + GLib.idle_add(callback, True, "") + except Exception as e: + logger.error(f"[Network] Connect with password failed: {e}") + if callback: + GLib.idle_add(callback, False, str(e)) + + import threading + + threading.Thread(target=_do_connect, daemon=True).start() + + def disconnect_wifi(self): + if not self._nm or not self.wifi_device: + return + + def _do_disconnect(): + for ac_path in self._nm.active_connection_paths: try: - device.disconnect_by_func(_on_state_changed) - except Exception as e: - logger.error(f"An error occurred: {e}") + ac = _make_proxy(self._nm._bus, ac_path, NM_ACTIVE_CONN_IFACE) + devices = _get(ac, "Devices") or [] + if self.wifi_device._device_path in devices: + self._nm.deactivate_connection(ac_path) + break + except Exception: + continue - handler_id.append(device.connect("state-changed", _on_state_changed)) + import threading - exec_shell_command_async( - f"nmcli device wifi connect {shlex.quote(bssid)} password {shlex.quote(password)}", - lambda *args: logger.debug(f"connect_with_password result: {args}"), + threading.Thread(target=_do_disconnect, daemon=True).start() + + # Airplane mode + + @Property(bool, "read-write", default_value=False) + def airplane_mode(self) -> bool: + if self._nm: + bluetooth_blocked = _rfkill_soft_blocked() + wifi_enabled = self._nm.wireless_enabled + self._airplane_mode = bluetooth_blocked and not wifi_enabled + return self._airplane_mode + + @airplane_mode.setter + def airplane_mode(self, value: bool): + self.set_airplane_mode(value) + + def set_airplane_mode(self, enable: bool): + """Toggle airplane mode: block/unblock bluetooth rfkill + toggle wifi radio. + + Port of dbus.rs NetworkBackend::set_airplane_mode. + """ + try: + action = "block" if enable else "unblock" + subprocess.run( + ["rfkill", action, "bluetooth"], + capture_output=True, + timeout=5, + ) + except FileNotFoundError: + logger.warning("[Network] rfkill not found") + except Exception as e: + logger.warning(f"[Network] rfkill {action} failed: {e}") + + if self._nm: + self._nm.wireless_enabled = not enable + + self._airplane_mode = enable + self.notify("airplane-mode") + self.emit("changed") + + @Property(str, "readable") + def connectivity(self) -> str: + if not self._nm: + return "unknown" + return self._nm.connectivity + + # Known connections + @Property(object, "readable") + def known_connections(self) -> List[KnownConnection]: + if not self._nm: + return [] + return self._nm.known_connections() + + # Active connections info + + @Property(object, "readable") + def active_connections(self) -> List[ActiveConnectionInfo]: + if not self._nm: + return [] + return self._nm.active_connections_info() + + # Network data snapshot + + def get_network_data(self) -> NetworkData: + return NetworkData( + wifi_present=self.wifi_device is not None, + wireless_access_points=( + self.wifi_device.access_points if self.wifi_device else [] + ), + active_connections=self.active_connections, + known_connections=self.known_connections, + wifi_enabled=(self.wifi_device.enabled if self.wifi_device else False), + airplane_mode=self.airplane_mode, + connectivity=self.connectivity, + scanning_nearby_wifi=( + self.wifi_device.scanning if self.wifi_device else False + ), ) diff --git a/src/window/controlcenter/bluetooth.py b/src/window/controlcenter/bluetooth.py index 5551bde..1744ef7 100644 --- a/src/window/controlcenter/bluetooth.py +++ b/src/window/controlcenter/bluetooth.py @@ -1,6 +1,6 @@ from enum import Enum, auto -from fabric.utils import Gdk, GLib, Gtk, logger +from fabric.utils import Gdk, GLib, Gtk, idle_add, logger from fabric.widgets.box import Box from fabric.widgets.button import Button from fabric.widgets.centerbox import CenterBox @@ -518,7 +518,7 @@ class BluetoothConnections(Box): ): self.client.scan() # Defer refresh until after the height animation completes - GLib.idle_add(self._refresh_after_animation) + idle_add(self._refresh_after_animation) def _cancel_pending_refresh(self): if hasattr(self, "_anim_finished_handler") and self._anim_finished_handler: diff --git a/src/window/controlcenter/main.py b/src/window/controlcenter/main.py index 1bae039..d63dd8c 100644 --- a/src/window/controlcenter/main.py +++ b/src/window/controlcenter/main.py @@ -767,23 +767,7 @@ class ModusControlCenter(AppletWindow): def toggle_flight_mode(self, *_): try: self.flight_mode = not self.flight_mode - - if self.flight_mode: - # Turn off WiFi and Bluetooth - if self.wifi_service: - self.wifi_service.wireless_enabled = False - if hasattr(self, "bluetooth_man") and hasattr( - self.bluetooth_man, "client" - ): - self.bluetooth_man.client.set_enabled(False) - else: - # Turn on WiFi and Bluetooth - if self.wifi_service: - self.wifi_service.wireless_enabled = True - if hasattr(self, "bluetooth_man") and hasattr( - self.bluetooth_man, "client" - ): - self.bluetooth_man.client.set_enabled(True) + self.network_service.set_airplane_mode(self.flight_mode) # Update icon self.flight_icon.dynamic_file( diff --git a/src/window/controlcenter/wifi.py b/src/window/controlcenter/wifi.py index b335c39..5dbd55e 100644 --- a/src/window/controlcenter/wifi.py +++ b/src/window/controlcenter/wifi.py @@ -1,4 +1,4 @@ -from fabric.utils import Gdk, GLib, logger +from fabric.utils import Gdk, GLib, idle_add, logger from fabric.widgets.box import Box from fabric.widgets.button import Button from fabric.widgets.centerbox import CenterBox @@ -35,9 +35,8 @@ class WifiNetworkSlot(Box): self.is_connected = access_point.is_active - self.styles = [ - "connected" if self.is_connected else "", - ] + if self.is_connected: + self.add_style_class("connected") wifi_icon_path = get_wifi_icon_for_strength(self.strength) self.dimage = svg_file(wifi_icon_path, size=28) @@ -108,15 +107,8 @@ class WifiNetworkSlot(Box): self.dimage.set_from_file(connecting_icon) self.dimage.add_style_class("disconnecting") - if self.wifi_service and self.wifi_service._device: - spawn_detached( - [ - "nmcli", - "device", - "disconnect", - self.wifi_service._device.get_iface(), - ] - ) + if self.network_service: + self.network_service.disconnect_wifi() self.is_connected = False GLib.timeout_add(500, lambda: self._reset_disconnect_state()) else: @@ -150,8 +142,8 @@ class WifiNetworkSlot(Box): try: if self.network_service: - self.network_service.connect_wifi_bssid( - self.access_point.bssid, callback=on_open_connection_result + self.network_service.connect_wifi( + self.access_point, callback=on_open_connection_result ) except Exception as e: logger.error( @@ -180,9 +172,10 @@ class WifiNetworkSlot(Box): def on_changed(self, *_): self.is_connected = self.access_point.is_active - self.styles = [ - "connected" if self.is_connected else "", - ] + if self.is_connected: + self.add_style_class("connected") + else: + self.remove_style_class("connected") if self.is_connected: self.wifi_icon_box.add_style_class("wifi-icon-box-connected") @@ -220,12 +213,7 @@ class WifiNetworkSlot(Box): GLib.timeout_add(500, lambda: self._reset_connect_state()) - if ( - self.password_dialog - and self.password_dialog.connection_timeout_id - ): - GLib.source_remove(self.password_dialog.connection_timeout_id) - self.password_dialog.connection_timeout_id = None + if self.password_dialog: self.password_dialog.is_connecting = False else: self._reset_connect_state() @@ -237,7 +225,7 @@ class WifiNetworkSlot(Box): try: if self.network_service: self.network_service.connect_wifi_with_password( - self.access_point.bssid, password, callback=on_connection_result + self.access_point, password, callback=on_connection_result ) except Exception as e: logger.error( @@ -278,8 +266,8 @@ class WifiConnections(Box): self._network_ready_fired = False # Guard against duplicate device-ready self._pending_toggle = None # Force toggle to target after stale revert - if network_service and network_service.wifi_device: - self.on_network_ready() + if self.network_service and self.network_service.wifi_device: + idle_add(self.on_network_ready) else: self._signal_ids.append( ( @@ -411,7 +399,7 @@ class WifiConnections(Box): if self.wifi_service: self.wifi_service.scan() # Defer refresh until after the height animation completes - GLib.idle_add(self._refresh_after_animation) + idle_add(self._refresh_after_animation) def _cancel_pending_refresh(self): if hasattr(self, "_anim_finished_handler") and self._anim_finished_handler: @@ -449,7 +437,16 @@ class WifiConnections(Box): return self._network_ready_fired = True self.wifi_service = wifi - self.toggle_button.set_active(self.wifi_service.enabled) + # Defer the initial sync: the D-Bus cached property may not yet be + # populated when device-ready fires inside _init_devices, causing + # enabled to read as False even when WiFi is on. + idle_add( + lambda: ( + self.toggle_button.set_active(self.wifi_service.enabled) + if self.wifi_service + else False + ) + ) self._signal_ids.append( ( @@ -469,6 +466,9 @@ class WifiConnections(Box): # Initial network update self.update_networks() + # Trigger initial scan so AP list is populated + self.wifi_service.scan() + def on_toggle_changed(self, toggle_button, *_): """Handle WiFi toggle button changes""" if self.wifi_service: @@ -508,109 +508,82 @@ class WifiConnections(Box): self._update_in_progress = True try: - current_known_ssids = { - child.ssid - for child in self.known_networks.get_children() - if hasattr(child, "ssid") - } - current_other_ssids = { - child.ssid - for child in self.other_networks.get_children() - if hasattr(child, "ssid") - } - access_points = self.wifi_service.access_points known_networks = [] other_networks = [] - new_known_ssids = set() - new_other_ssids = set() for access_point in access_points: - try: - if access_point.ssid and access_point.ssid != "Unknown": - if access_point.is_active or self._is_saved_network( - access_point - ): - known_networks.append(access_point) - new_known_ssids.add(access_point.ssid) - else: - other_networks.append(access_point) - new_other_ssids.add(access_point.ssid) - except Exception as e: - logger.warning( - f"[WiFi] Failed to process access point {access_point}: {e}" - ) - continue + if access_point.ssid and access_point.ssid != "Unknown": + if access_point.is_active or self._is_saved_network(access_point): + known_networks.append(access_point) + else: + other_networks.append(access_point) - known_changed = current_known_ssids != new_known_ssids - other_changed = current_other_ssids != new_other_ssids + # Get existing networks + existing_known = { + child.ssid: child for child in self.known_networks.get_children() + } + existing_other = { + child.ssid: child for child in self.other_networks.get_children() + } - if known_changed or other_changed: - # Get existing networks - existing_known = { - child.ssid: child for child in self.known_networks.get_children() - } - existing_other = { - child.ssid: child for child in self.other_networks.get_children() - } + for access_point in known_networks: + if not self._destroyed: + if access_point.ssid in existing_known: + slot = existing_known.pop(access_point.ssid) + slot.update_ap(access_point) + elif access_point.ssid in existing_other: + slot = existing_other.pop(access_point.ssid) + slot.update_ap(access_point) + self.other_networks.remove(slot) + self.known_networks.add(slot) + else: + network_slot = WifiNetworkSlot( + access_point, + self.wifi_service, + network_service=self.network_service, + parent=self.parent, + ) + self.known_networks.add(network_slot) - for access_point in known_networks: - if not self._destroyed: - if access_point.ssid in existing_known: - slot = existing_known.pop(access_point.ssid) - slot.update_ap(access_point) - elif access_point.ssid in existing_other: - slot = existing_other.pop(access_point.ssid) - slot.update_ap(access_point) - self.other_networks.remove(slot) - self.known_networks.add(slot) - else: - network_slot = WifiNetworkSlot( - access_point, - self.wifi_service, - network_service=self.network_service, - parent=self.parent, - ) - self.known_networks.add(network_slot) + for access_point in other_networks: + if not self._destroyed: + if access_point.ssid in existing_other: + slot = existing_other.pop(access_point.ssid) + slot.update_ap(access_point) + elif access_point.ssid in existing_known: + slot = existing_known.pop(access_point.ssid) + slot.update_ap(access_point) + self.known_networks.remove(slot) + self.other_networks.add(slot) + else: + network_slot = WifiNetworkSlot( + access_point, + self.wifi_service, + network_service=self.network_service, + parent=self.parent, + ) + self.other_networks.add(network_slot) - for access_point in other_networks: - if not self._destroyed: - if access_point.ssid in existing_other: - slot = existing_other.pop(access_point.ssid) - slot.update_ap(access_point) - elif access_point.ssid in existing_known: - slot = existing_known.pop(access_point.ssid) - slot.update_ap(access_point) - self.known_networks.remove(slot) - self.other_networks.add(slot) - else: - network_slot = WifiNetworkSlot( - access_point, - self.wifi_service, - network_service=self.network_service, - parent=self.parent, - ) - self.other_networks.add(network_slot) + for slot in existing_known.values(): + if hasattr(slot, "access_point"): + slot.strength = 0 + from shared.widgets.wifi_icon import get_wifi_icon_for_strength - for slot in existing_known.values(): - if hasattr(slot, "access_point"): - slot.strength = 0 - from shared.widgets.wifi_icon import get_wifi_icon_for_strength + if not slot.dimage.has_style_class( + "connecting" + ) and not slot.dimage.has_style_class("disconnecting"): + slot.dimage.set_from_file(get_wifi_icon_for_strength(0)) - if not slot.dimage.has_style_class( - "connecting" - ) and not slot.dimage.has_style_class("disconnecting"): - slot.dimage.set_from_file(get_wifi_icon_for_strength(0)) + for slot in existing_other.values(): + if hasattr(slot, "access_point"): + slot.strength = 0 + from shared.widgets.wifi_icon import get_wifi_icon_for_strength - for slot in existing_other.values(): - if hasattr(slot, "access_point"): - slot.strength = 0 - from shared.widgets.wifi_icon import get_wifi_icon_for_strength - - if not slot.dimage.has_style_class( - "connecting" - ) and not slot.dimage.has_style_class("disconnecting"): - slot.dimage.set_from_file(get_wifi_icon_for_strength(0)) + if not slot.dimage.has_style_class( + "connecting" + ) and not slot.dimage.has_style_class("disconnecting"): + slot.dimage.set_from_file(get_wifi_icon_for_strength(0)) if not self._destroyed: has_known_networks = len(known_networks) > 0