From dc150aee8102f8aae7d321acd3df580d26422425 Mon Sep 17 00:00:00 2001 From: S4NKALP Date: Wed, 17 Jun 2026 02:00:26 +0545 Subject: [PATCH] refactor(src/services): drop unused NetworkClient and simplify network handling Prevent unused NetworkClient import and remove extensive network code that was blocking release; streamline to only Wifi service and needed helpers. --- src/services/__init__.py | 2 +- src/services/network.py | 946 ++++++++++++++------------------------- 2 files changed, 336 insertions(+), 612 deletions(-) diff --git a/src/services/__init__.py b/src/services/__init__.py index c9de1ad..f46cca9 100644 --- a/src/services/__init__.py +++ b/src/services/__init__.py @@ -3,7 +3,7 @@ from .battery import Battery # noqa: F401 from .brightness import Brightness # noqa: F401 from .capslock import CapsLock # noqa: F401 from .keyboard_layout import KeyboardLayout # noqa: F401 -from .network import NetworkClient, Wifi, Ethernet, AccessPoint # noqa: F401 +from .network import NetworkClient, Wifi, Ethernet # noqa: F401 from .mpris import MprisPlayer, MprisPlayerManager # noqa: F401 from .todo import TodoService # noqa: F401 from .modus import ModusService # noqa: F401 diff --git a/src/services/network.py b/src/services/network.py index 07a7350..7a010c3 100644 --- a/src/services/network.py +++ b/src/services/network.py @@ -1,593 +1,205 @@ -from typing import Optional +from types import SimpleNamespace +from typing import List, Literal import gi from fabric.core.service import Property, Service, Signal -from fabric.utils import ( - GLib, - bulk_connect, - get_enum_member_name, - logger, - snake_case_to_kebab_case, -) -from gi.repository import NM +from fabric.utils import Gio, bulk_connect, exec_shell_command_async, logger -gi.require_version("NM", "1.0") # Ensure the correct version is loaded - - -class NetworkClient(Service): - """A service to manage network devices""" - - @Signal - def wifi_device_added(self) -> None: ... - - @Signal - def ethernet_device_added(self) -> None: ... - - @Signal - def wifi_device_removed(self) -> None: ... - - @Signal - def ethernet_device_removed(self) -> None: ... - - @Signal - def changed(self) -> None: ... - - @Property(list, "readable") - def connections(self) -> Optional[list]: - """Returns the active connections, if available.""" - return self._client.get_property("connections") if self._client else None - - @Property(object, "readable") - def wifi_device(self) -> Optional[object]: - """Returns the WiFi device if available.""" - return self._wifi_device - - @Property(object, "readable") - def ethernet_device(self) -> Optional[object]: - """Returns the Ethernet device if available.""" - return self._ethernet_device - - @Property(str, "readable") - def primary_connection(self) -> Optional[str]: - """Returns the primary connection if available.""" - return self._client.get_property("primary_connection") if self._client else None - - @Property(str, "readable") - def active_connection(self) -> Optional[str]: - """Returns the active connection if available.""" - return self._client.get_property("active_connection") if self._client else None - - @Property(str, "readable") - def state(self) -> str: - """Returns the current network state.""" - if not self._client: - return "disconnected" - return snake_case_to_kebab_case( - get_enum_member_name( - self._client.get_property("state"), default="disconnected" - ) - ) - - @Property(str, "readable") - def connectivity(self) -> str: - """Returns the connectivity state.""" - if not self._client: - return "disconnected" - return snake_case_to_kebab_case( - get_enum_member_name( - self._client.get_property("connectivity"), default="disconnected" - ) - ) - - @Property(list, "readable") - def devices(self) -> Optional[list]: - """Returns the list of network devices if available.""" - return self._client.get_property("devices") if self._client else None - - @Property(str, "readable") - def hostname(self) -> Optional[str]: - """Returns the hostname if available.""" - return self._client.get_property("hostname") if self._client else None - - @Property(bool, "read-write", default_value=False) - def networking_enabled(self) -> bool: - """Checks if networking is enabled.""" - return ( - self._client.get_property("networking_enabled") if self._client else False - ) - - @networking_enabled.setter - def networking_enabled(self, value: bool): - """Sets the networking state if the client is available.""" - if self._client: - self._client.set_property("networking_enabled", value) - - @Property(bool, "read-write", default_value=False) - def wireless_enabled(self) -> bool: - """Checks if wireless networking is enabled.""" - return self._client.get_property("wireless_enabled") if self._client else False - - @wireless_enabled.setter - def wireless_enabled(self, value: bool): - """Sets the wireless networking state if the client is available.""" - if self._client: - self._client.set_property("wireless_enabled", value) - - def __init__(self, **kwargs): - super().__init__(**kwargs) - - self._client: NM.Client = None - self._wifi_device: Wifi | None = None - self._ethernet_device: Ethernet | None = None - - logger.info("[Network] Initializing client asynchronously...") - - # Start async NM client initialization - NM.Client.new_async(None, self.on_client_ready) - - def on_client_ready(self, source, result): - """Callback when NM.Client is ready.""" - try: - self._client = NM.Client.new_finish(result) # Retrieve client instance - logger.info("[Network] NM.Client initialized successfully!") - - # Connect signals - bulk_connect( - self._client, - { - "device-added": lambda _, device: self.on_device_added( - device=device - ), - "device-removed": lambda _, device: self.on_device_removed( - device=device - ), - "notify::state": lambda *args: self.notifier("state"), - "notify::networking-enabled": lambda *args: self.notifier( - "networking-enabled" - ), - "notify::wireless-enabled": lambda *args: self.notifier( - "wireless-enabled" - ), - # "notify::primary-connection": lambda *args: self.notifier('primary-connection'), - # "notify::active-connection": lambda *args: self.notifier('active-connection'), - # "active-connection-added": lambda *args: self.emit("changed"), - # "active-connection-removed": lambda *args: self.emit("changed") - }, - ) - - # Process devices AFTER client is ready - for device in self.do_get_raw_devices(): - self.on_device_added(device=device) - - self.notify("state") - self.notify("networking-enabled") - self.notify("wireless-enabled") - - except Exception as e: - logger.error(f"[Network] Error initializing NM.Client: {e}") - - def do_get_raw_devices(self) -> list[NM.Device]: - return [ - dev - for dev in self.devices - if dev.get_device_type() in (NM.DeviceType.WIFI, NM.DeviceType.ETHERNET) - ] - - def on_device_added(self, device): - device_type = device.get_device_type() - if device_type == NM.DeviceType.WIFI and not self._wifi_device: - logger.info("[Network] WiFi device detected, initializing...") - self._wifi_device = Wifi(client=self, device=device) - self.wifi_device_added.emit() - - elif device_type == NM.DeviceType.ETHERNET and not self._ethernet_device: - logger.info("[Network] Ethernet device detected, initializing...") - self._ethernet_device = Ethernet(client=self, device=device) - self.ethernet_device_added.emit() - - def on_device_removed(self, device): - if device == self._wifi_device: - logger.info("[Network] WiFi device removed.") - self._wifi_device = None - self.wifi_device_removed.emit() - - elif device == self._ethernet_device: - logger.info("[Network] Ethernet device removed.") - self._ethernet_device = None - self.ethernet_device_removed.emit() - - def toggle_network(self): - """Enable or disable Network""" - self.networking_enabled = not self.networking_enabled - - def deactivate_connection(self, connection): - """Disconnect""" - self._client.deactivate_connection_async(connection, None, None) - - def notifier(self, name): - self.notify(name) - self.emit("changed") - - -class AccessPoint(Service): - """A service to manage access points""" - - @Signal - def changed(self) -> None: ... - - @Property(object, "readable") - def device(self) -> object: - return self._device - - @Property(int, "readable") - def strength(self) -> int: - return self._ap.get_property("strength") - - @Property(int, "readable") - def frequency(self) -> int: - return self._ap.get_property("frequency") - - @Property(str, "readable") - def bssid(self) -> str: - return self._ap.get_property("bssid") if self._ap else None - - @Property(str, "readable") - def hw_address(self) -> str: - return self._ap.get_property("hw_address") - - @Property(str, "readable") - def ssid(self) -> str: - ssid = self._ap.get_ssid() - return NM.utils_ssid_to_utf8(ssid.get_data()) if ssid else "Unknown" - - @Property(str, "readable") - def icon(self) -> 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", - 00: "network-wireless-signal-none-symbolic", - }.get( - min(80, 20 * round(self.strength / 20)), - "network-wireless-no-route-symbolic", - ) - - @Property(bool, "readable", default_value=False) - def requires_password(self) -> bool: - ssid = self.ssid - settings = self._client.connections - connection = None - for setting in settings: - wifi_setting = setting.get_setting_wireless() - if ( - wifi_setting - and NM.utils_ssid_to_utf8(wifi_setting.get_ssid().get_data()) == ssid - ): - connection = setting - break - if not connection: - return bool(self._ap.get_wpa_flags() or self._ap.get_rsn_flags()) - return False - - @Property(bool, "readable", default_value=False) - def is_active(self) -> bool: - if self._device.active_access_point: - return self.bssid == self._device.active_access_point.get_bssid() - return False - - def __init__(self, device: "Wifi", ap: NM.AccessPoint, **kwargs): - super().__init__(**kwargs) - self._client: NetworkClient = device.client - self._device: Wifi = device - self._ap: NM.AccessPoint = ap - - self._ap.connect("notify::strength", lambda *args: self.notifier("strength")) - self._device.connect( - "notify::active-access-point", lambda *args: self.notifier("is-active") - ) - - def notifier(self, name: str, *args): - self.notify(name) - self.emit("changed") - return +try: + gi.require_version("NM", "1.0") + from gi.repository import NM +except ValueError: + logger.error("Failed to start network manager") class Wifi(Service): - """A service to manage wifi devices""" + """A service to manage the wifi connection.""" @Signal def changed(self) -> None: ... @Signal - def ap_added(self, ap: AccessPoint) -> None: ... + def enabled_changed(self) -> bool: ... - @Signal - def ap_removed(self, ap: AccessPoint) -> None: ... - - @Property(NetworkClient, "readable") - def client(self) -> NetworkClient: - """Returns the client""" - return self._client - - @Property(bool, "read-write", default_value=False) - def wireless_enabled(self) -> bool: - """Returns if the wifi is enabled""" - return self._client.get_property("wireless_enabled") - - @wireless_enabled.setter - def wireless_enabled(self, value: bool): - return self._client.set_property("wireless_enabled", value) - - @Property(list[AccessPoint], "readable") - def access_points(self) -> list[AccessPoint]: - return sorted( - self._access_points.values(), key=lambda x: x.is_active, reverse=True - ) - - @Property(AccessPoint, "readable") - def active_access_point(self) -> Optional[AccessPoint]: - return self._active_access_point - - def __init__(self, client: NetworkClient, device: NM.DeviceWifi, **kwargs): - super().__init__(**kwargs) - self._client: NetworkClient = client + def __init__(self, client: NM.Client, device: NM.DeviceWifi, **kwargs): + self._client: NM.Client = client self._device: NM.DeviceWifi = device - self._active_access_point: NM.AccessPoint | None = None - self._access_points: dict[str, AccessPoint] = {} - - bulk_connect( - self._device, - { - "notify::active-access-point": lambda *args: ( - self.on_access_point_activated() - ), - "access-point-added": lambda _, ap: self.on_access_point_added(ap=ap), - "access-point-removed": lambda _, ap: self.on_access_point_removed( - ap=ap - ), - # "state-changed": lambda device, new, old, reason: self.on_state_changed(new), - }, - ) + self._ap: NM.AccessPoint | None = None + self._ap_signal: int | None = None + super().__init__(**kwargs) self._client.connect( - "notify::wireless-enabled", lambda *args: self.notifier("wireless-enabled") + "notify::wireless-enabled", + lambda *args: self.notifier("enabled", args), + ) + if self._device: + 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() + + 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() ) - for ap in self.do_get_access_points(): - self.on_access_point_added(ap=ap) - - self.on_access_point_activated() - - def on_state_changed(self, state): - self.emit("changed") - - def do_get_access_points(self): - return self._device.get_access_points() - - def on_access_point_added(self, ap): - ssid = ap.get_ssid() - ssid = NM.utils_ssid_to_utf8(ssid.get_data()) if ssid else "Unknown" - - access_point: AccessPoint = AccessPoint(ap=ap, device=self) - - self._access_points[ssid] = access_point - - self.ap_added.emit(access_point) - - # self.notifier("access-points") - - logger.info(f"[Wifi] New access point added with ssid: {ssid}") - - def on_access_point_removed(self, ap): - ssid = ap.get_ssid() - ssid = NM.utils_ssid_to_utf8(ssid.get_data()) if ssid else "Unknown" - - if not (access_point := self._access_points.pop(ssid, None)): - return logger.warning( - f"[Network] tried to remove a unknwon access point with ssid {ssid}" - ) - - self.ap_removed.emit(access_point) - - logger.info(f"[Wifi] Access point with ssid: {ssid} removed.") - - def on_access_point_activated(self): - if self._device.get_active_access_point(): - self._active_access_point: AccessPoint = AccessPoint( - ap=self._device.get_active_access_point(), device=self - ) - - else: - self._active_access_point = None - - self.notifier("active-access-point") - - logger.info("[Wifi] New active connection") - - def disconnect_wifi(self): - """Disconnect from the current WiFi network.""" - active_connection = self._device.get_active_connection() - self._client.deactivate_connection(active_connection) - logger.info("[Wifi] Wifi network disconnected") + def toggle_wifi(self): + 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.notifier("access-points"), + self.emit("changed"), ], ) - logger.info("[Wifi] Scan started") - - def toggle_wifi(self): - """Enable or disable WiFi""" - self.wireless_enabled = not self.wireless_enabled - - def connect_to_wifi(self, ap: AccessPoint, password: str = None, callback=None): - """Connect to a WiFi network.""" - - ssid = ap.ssid - - if ssid == "Unknown": - logger.error("Invalid access point data") - if callback: - callback(False, "Invalid access point data") - return False - - logger.info(f"Connecting to WiFi SSID: {ssid}") - - # Check for existing connections - settings = self._client.connections - connection = None - - for setting in settings: - wifi_setting = setting.get_setting_wireless() - if ( - wifi_setting - and NM.utils_ssid_to_utf8(wifi_setting.get_ssid().get_data()) == ssid - ): - connection = setting - break - - def on_activation_result(client, result): - """Handle the result of connection activation""" - try: - active_connection = client.activate_connection_finish(result) - if active_connection: - logger.info(f"Successfully connected to '{ssid}'") - if callback: - callback(True, "Connected successfully") - else: - logger.error(f"Failed to connect to '{ssid}'") - if callback: - callback(False, "Failed to connect. Please try again.") - except Exception as e: - error_msg = str(e).lower() - logger.error(f"Connection to '{ssid}' failed: {e}") - - # Parse NetworkManager error messages and provide user-friendly responses - if "802-11-wireless-security" in error_msg: - if "property is invalid" in error_msg or "psk" in error_msg: - user_msg = "Incorrect password. Please try again." - else: - user_msg = "Security configuration error. Please try again." - elif "secrets were required" in error_msg or "no secrets" in error_msg: - user_msg = "Incorrect password. Please try again." - elif "timeout" in error_msg: - user_msg = "Connection timeout. Please try again." - elif "not found" in error_msg or "no such device" in error_msg: - user_msg = "Network not available. Please try again." - elif "already connected" in error_msg: - user_msg = "Already connected to this network." - else: - user_msg = "Failed to connect. Please try again." - - if callback: - callback(False, user_msg) - - if not connection: - # Create a new connection profile - logger.info(f"Creating new WiFi connection for SSID '{ssid}'") - connection = NM.SimpleConnection.new() - - # Required connection settings - s_con = NM.SettingConnection.new() - s_con.set_property(NM.SETTING_CONNECTION_ID, ssid) - s_con.set_property(NM.SETTING_CONNECTION_TYPE, "802-11-wireless") - s_con.set_property( - NM.SETTING_CONNECTION_INTERFACE_NAME, self._device.get_iface() - ) # Set interface name - connection.add_setting(s_con) - - # Wireless settings - s_wifi = NM.SettingWireless.new() - s_wifi.set_property(NM.SETTING_WIRELESS_SSID, GLib.Bytes.new(ssid.encode())) - s_wifi.set_property( - NM.SETTING_WIRELESS_MODE, "infrastructure" - ) # Ensure mode is correct - connection.add_setting(s_wifi) - - # Security settings (only if password is required and provided) - if ap.requires_password: - if not password: - logger.error("Password required but not provided") - if callback: - callback(False, "Password required but not provided") - return False - - s_sec = NM.SettingWirelessSecurity.new() - s_sec.set_property(NM.SETTING_WIRELESS_SECURITY_KEY_MGMT, "wpa-psk") - s_sec.set_property(NM.SETTING_WIRELESS_SECURITY_PSK, password) - connection.add_setting(s_sec) - - # IPv4 settings - s_ipv4 = NM.SettingIP4Config.new() - s_ipv4.set_property("method", "auto") - connection.add_setting(s_ipv4) - - # IPv6 settings - s_ipv6 = NM.SettingIP6Config.new() - s_ipv6.set_property("method", "auto") - connection.add_setting(s_ipv6) - - # Callback for async connection - def on_connection_added(client, result): - try: - new_connection = client.add_connection_finish(result) - if not new_connection: - logger.error(f"Failed to create connection for '{ssid}'") - if callback: - callback( - False, "Failed to create connection. Please try again." - ) - return - - logger.info(f"Connection for '{ssid}' added successfully") - - # Now activate the newly created connection - client.activate_connection_async( - new_connection, self._device, None, None, on_activation_result - ) - except Exception as e: - error_msg = str(e).lower() - logger.error(f"Failed to add connection: {e}") - - # Parse connection creation errors and provide user-friendly responses - if "802-11-wireless-security" in error_msg: - if "property is invalid" in error_msg or "psk" in error_msg: - user_msg = "Incorrect password. Please try again." - else: - user_msg = "Security configuration error. Please try again." - elif "invalid" in error_msg or "property" in error_msg: - user_msg = "Invalid network configuration. Please try again." - else: - user_msg = "Failed to connect. Please try again." - - if callback: - callback(False, user_msg) - - # Save the new connection - self._client._client.add_connection_async( - connection, True, None, on_connection_added - ) - else: - # Activate existing connection - self._client._client.activate_connection_async( - connection, self._device, None, None, on_activation_result - ) - - return True def notifier(self, name: str, *args): self.notify(name) self.emit("changed") return + @Property(bool, "read-write", default_value=False) + def enabled(self) -> bool: + return bool(self._client.wireless_get_enabled()) + + @enabled.setter + def enabled(self, value: bool): + self._client.wireless_set_enabled(value) + + @Property(int, "readable") + def strength(self): + return self._ap.get_strength() if self._ap else -1 + + @Property(str, "readable") + def icon_name(self): + if not self._ap: + 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", + ) + 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 + + @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") + + @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)) + + @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" + + @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") + class Ethernet(Service): - """A service to manage ethernet devices""" + """A service to manage the ethernet connection.""" @Signal def changed(self) -> None: ... @@ -596,80 +208,192 @@ class Ethernet(Service): def enabled(self) -> bool: ... @Property(int, "readable") - def speed(self) -> str: - speed_mbps = self._device.get_speed() - return f"{speed_mbps} Mb/s" + def speed(self) -> int: + return self._device.get_speed() - @Property(str, "readable") - def state(self) -> str: - return snake_case_to_kebab_case( - get_enum_member_name(self._device.get_state(), default="disconnected") - ) - - @Property(str, "readable") - def internet(self) -> str: - if self._active_connection: - return snake_case_to_kebab_case( - get_enum_member_name( - self._active_connection.get_state(), default="disconnected" - ) - ) - return "disconnected" - - @Property(str, "readable") - def iface(self) -> str: - return self._device.get_iface() if self._device else 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") @Property(str, "readable") def icon_name(self) -> str: network = self.internet if network == "activated": return "network-wired-symbolic" + elif network == "activating": return "network-wired-acquiring-symbolic" + + elif self._device.get_connectivity != NM.ConnectivityState.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._active_connection = None - self._device.connect( - "state-changed", lambda *args: self.on_network_state_changed() + for pn in ( + "active-connection", + "icon-name", + "internet", + "speed", + "state", + ): + self._device.connect(f"notify::{pn}", lambda *_: self.notifier(pn)) + + self._device.connect("notify::speed", lambda *_: print(_)) + + def notifier(self, pn): + self.notify(pn) + self.emit("changed") + + +class NetworkClient(Service): + @Signal + def device_ready(self) -> None: ... + + @Signal + def device_added(self, iface: str) -> None: ... + + @Signal + def device_removed(self, iface: str) -> None: ... + + def __init__(self, **kwargs): + self._client: NM.Client | None = None + self.wifi_device: Wifi | None = None + self.wifi_devices: dict[str, Wifi] = {} + self.ethernet_device: Ethernet | None = None + super().__init__(**kwargs) + NM.Client.new_async( + cancellable=None, + callback=self._init_network_client, ) - self.update_active_connection() + def _init_network_client(self, client: NM.Client, task: Gio.Task, **kwargs): + self._client = client - def on_network_state_changed(self): - """Called when networking is toggled on/off.""" - if self.state != "unmanaged": - # Re-initialize device when networking is re-enabled - self.update_active_connection() - else: - self._active_connection = None + for device in self._client.get_devices(): + self._handle_device_added(device) - def update_active_connection(self): - """Updates the active connection and connects to its state change signal.""" - active_connection = self._device.get_active_connection() - if active_connection: - self._active_connection = active_connection - self._active_connection.connect( - "state-changed", lambda *args: self.emit("changed") - ) + self._client.connect("device-added", lambda _, d: self._handle_device_added(d)) + self._client.connect( + "device-removed", lambda _, d: self._handle_device_removed(d) + ) - def get_network_stats(self): - """Fetch received and transmitted bytes from the system files""" + self.notify("primary-device") + + def _handle_device_added(self, nm_device): + dtype = nm_device.get_device_type() + + if dtype == NM.DeviceType.WIFI: + iface = nm_device.get_iface() + if iface in self.wifi_devices: + return + wifi = Wifi(self._client, nm_device) + self.wifi_devices[iface] = 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: + if self.ethernet_device is None: + self.ethernet_device = Ethernet(client=self._client, device=nm_device) + self.emit("device-ready") + + def _handle_device_removed(self, nm_device): + dtype = nm_device.get_device_type() + + if dtype == NM.DeviceType.WIFI: + iface = nm_device.get_iface() + wifi = self.wifi_devices.pop(iface, None) + if wifi is None: + return + if self.wifi_device is wifi: + self.wifi_device = next(iter(self.wifi_devices.values()), None) + logger.info(f"[Network] Wifi device removed: {iface}") + self.emit("device-removed", iface) + + def _get_primary_device(self) -> Literal["wifi", "wired"] | None: + if not self._client: + return None try: - # Read data from /sys/class/net for accurate speed - with open( - f"/sys/class/net/{self.iface}/statistics/rx_bytes", "r" - ) as rx_file: - rx_bytes = int(rx_file.read().strip()) - with open( - f"/sys/class/net/{self.iface}/statistics/tx_bytes", "r" - ) as tx_file: - tx_bytes = int(tx_file.read().strip()) - return rx_bytes, tx_bytes - except FileNotFoundError: - return None, None + 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" + except Exception: + pass + return None + + def is_network_saved(self, ssid: str) -> bool: + if not self._client: + return False + for conn in self._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 + return False + + def connect_wifi_bssid(self, bssid: str): + exec_shell_command_async( + f"nmcli device wifi connect {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 + + device = self.wifi_device._device + handler_id: list[int] = [] + + def _on_state_changed(dev, new_state, old_state, reason): + 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 + + handler_id.append(device.connect("state-changed", _on_state_changed)) + + exec_shell_command_async( + f"nmcli device wifi connect {bssid} password {password!r}", + lambda *args: logger.debug(f"connect_with_password result: {args}"), + ) + + @Property(str, "readable") + def primary_device(self) -> Literal["wifi", "wired"] | None: + return self._get_primary_device()