diff --git a/modules/controlcenter/wifi.py b/modules/controlcenter/wifi.py index 8fc9267..8aa44d7 100644 --- a/modules/controlcenter/wifi.py +++ b/modules/controlcenter/wifi.py @@ -1,13 +1,12 @@ -from gi.repository import Gtk, Gdk, GLib - -from services.network import NetworkService from fabric.widgets.box import Box from fabric.widgets.button import Button from fabric.widgets.centerbox import CenterBox from fabric.widgets.image import Image from fabric.widgets.label import Label from fabric.widgets.scrolledwindow import ScrolledWindow -from fabric.widgets.eventbox import EventBox +from gi.repository import Gdk, GLib, Gtk + +from services.network import NetworkService class WifiNetworkSlot(CenterBox): @@ -20,14 +19,16 @@ class WifiNetworkSlot(CenterBox): self.ssid = network_data.get("ssid", "Unknown") self.bssid = network_data.get("bssid", "") self.strength = network_data.get("strength", 0) - self.icon_name = network_data.get("icon-name", "network-wireless-signal-none-symbolic") + self.icon_name = network_data.get( + "icon-name", "network-wireless-signal-none-symbolic" + ) # The network service puts the same active-ap reference in all networks # We need to check if THIS specific network's BSSID matches the active AP's BSSID self.is_connected = False if self.wifi_service and self.wifi_service._ap: active_bssid = self.wifi_service._ap.get_bssid() - self.is_connected = (active_bssid == self.bssid) + self.is_connected = active_bssid == self.bssid # Initialize styles based on connection state self.styles = [ @@ -59,29 +60,71 @@ class WifiNetworkSlot(CenterBox): is_currently_connected = False if self.wifi_service and self.wifi_service._ap: active_bssid = self.wifi_service._ap.get_bssid() - is_currently_connected = (active_bssid == self.bssid) + is_currently_connected = active_bssid == self.bssid if is_currently_connected: + # Show disconnecting state + self.dimage.set_property("icon-name", "network-wireless-acquiring-symbolic") + self.dimage.add_style_class("disconnecting") + # Disconnect from network success = self.wifi_service.disconnect_network(self.ssid) if success: self.is_connected = False + # Remove disconnecting state after a short delay to show feedback + from gi.repository import GLib + + GLib.timeout_add(500, lambda: self._reset_disconnect_state()) + else: + # Reset state if disconnect failed + self._reset_disconnect_state() else: + # Show connecting state + self.dimage.set_property("icon-name", "network-wireless-acquiring-symbolic") + self.dimage.add_style_class("connecting") + # Connect to network (try without password first, could be enhanced for password input) success = self.wifi_service.connect_network(self.ssid) if success: self.is_connected = True + # Remove connecting state after a short delay + from gi.repository import GLib + + GLib.timeout_add(500, lambda: self._reset_connect_state()) + else: + # Reset state if connect failed + self._reset_connect_state() # Update display after connection attempt self.on_changed() + def _reset_disconnect_state(self): + """Reset visual state after disconnect operation""" + self.dimage.remove_style_class("disconnecting") + self.dimage.set_property("icon-name", "network-wireless-symbolic") + self.on_changed() + return False # Remove timeout + + def _reset_connect_state(self): + """Reset visual state after connect operation""" + self.dimage.remove_style_class("connecting") + self.dimage.set_property("icon-name", "network-wireless-symbolic") + self.on_changed() + return False # Remove timeout + def on_changed(self, *_): """Update display when connection state changes""" # Check if this network is currently connected by comparing BSSIDs is_currently_connected = False if self.wifi_service and self.wifi_service._ap: - active_bssid = self.wifi_service._ap.get_bssid() - is_currently_connected = (active_bssid == self.bssid) + # Also check the internet connection state to ensure we're actually connected + internet_state = self.wifi_service.internet + if internet_state == "activated": + active_bssid = self.wifi_service._ap.get_bssid() + is_currently_connected = active_bssid == self.bssid + # If deactivating or deactivated, definitely not connected + elif internet_state in ["deactivating", "deactivated", "disconnected"]: + is_currently_connected = False self.is_connected = is_currently_connected self.styles = [ @@ -114,7 +157,7 @@ class WifiConnections(Box): label="↓ Pull to scan for networks", h_align="center", visible=False, - style="color: #fff; font-size: 12px; padding: 5px;" + style="color: #fff; font-size: 12px; padding: 5px;", ) self.title = Box( @@ -135,7 +178,7 @@ class WifiConnections(Box): self.device_box = Box( spacing=2, orientation="vertical", - children=[self.refresh_indicator, self.available_networks] + children=[self.refresh_indicator, self.available_networks], ) # Create scrolled window with pull-to-refresh @@ -170,7 +213,9 @@ class WifiConnections(Box): self.toggle_button.set_active(self.wifi_service.enabled) self.toggle_button.connect( "notify::active", - lambda *_: setattr(self.wifi_service, 'enabled', self.toggle_button.get_active()), + lambda *_: setattr( + self.wifi_service, "enabled", self.toggle_button.get_active() + ), ) # Connect to WiFi service signals @@ -178,7 +223,12 @@ class WifiConnections(Box): self.wifi_service.connect("notify::scanning", self.on_scanning_changed) self.wifi_service.connect("changed", self.update_networks) # Also connect to access point changes to update connection states - self.wifi_service.connect("notify::access-points", self.refresh_network_states) + self.wifi_service.connect( + "notify::access-points", self.refresh_network_states + ) + # Connect to internet state changes to handle disconnections + self.wifi_service.connect("notify::internet", self.refresh_network_states) + self.wifi_service.connect("notify::ssid", self.refresh_network_states) # Initial network update self.update_networks() @@ -220,7 +270,7 @@ class WifiConnections(Box): def refresh_network_states(self, *_): """Refresh connection states for all network slots""" for child in self.available_networks.get_children(): - if hasattr(child, 'on_changed'): + if hasattr(child, "on_changed"): child.on_changed() def setup_pull_to_refresh(self): @@ -237,7 +287,8 @@ class WifiConnections(Box): self.bounce_timeout_id = None self.bounce_frame = 0 self.bounce_duration = 60 # Total frames for animation - self.bounce_amplitude = 30 # Maximum bounce height in pixels (increased for visibility) + # Maximum bounce height in pixels (increased for visibility) + self.bounce_amplitude = 30 # Connect to scroll events self.scrolled_window.connect("scroll-event", self.on_scroll_event) @@ -247,10 +298,10 @@ class WifiConnections(Box): # Enable events self.scrolled_window.set_events( - Gdk.EventMask.SCROLL_MASK | - Gdk.EventMask.BUTTON_PRESS_MASK | - Gdk.EventMask.BUTTON_RELEASE_MASK | - Gdk.EventMask.POINTER_MOTION_MASK + Gdk.EventMask.SCROLL_MASK + | Gdk.EventMask.BUTTON_PRESS_MASK + | Gdk.EventMask.BUTTON_RELEASE_MASK + | Gdk.EventMask.POINTER_MOTION_MASK ) def on_scroll_event(self, widget, event): @@ -332,22 +383,22 @@ class WifiConnections(Box): def ease_out_bounce(self, t): """Smooth bounce easing function (0 to 1)""" - import math - if t < 1/2.75: + if t < 1 / 2.75: return 7.5625 * t * t - elif t < 2/2.75: - t -= 1.5/2.75 + elif t < 2 / 2.75: + t -= 1.5 / 2.75 return 7.5625 * t * t + 0.75 - elif t < 2.5/2.75: - t -= 2.25/2.75 + elif t < 2.5 / 2.75: + t -= 2.25 / 2.75 return 7.5625 * t * t + 0.9375 else: - t -= 2.625/2.75 + t -= 2.625 / 2.75 return 7.5625 * t * t + 0.984375 def ease_out_elastic(self, t): """Elastic easing function for smoother bounce""" import math + if t == 0 or t == 1: return t @@ -371,7 +422,10 @@ class WifiConnections(Box): self.refresh_indicator.set_margin_top(0) self.bounce_timeout_id = None # Hide indicator after animation if not scanning and not pulling - if not (self.wifi_service and self.wifi_service.scanning) and not self.is_pulling: + if ( + not (self.wifi_service and self.wifi_service.scanning) + and not self.is_pulling + ): self.refresh_indicator.set_visible(False) return False @@ -395,7 +449,8 @@ class WifiConnections(Box): self.bounce_frame = 0 self.bounce_duration = 80 # Longer for elastic effect - self.bounce_amplitude = 40 # Higher amplitude (increased for visibility) + # Higher amplitude (increased for visibility) + self.bounce_amplitude = 40 self.bounce_timeout_id = GLib.timeout_add(16, self.animate_elastic) def animate_elastic(self): @@ -405,7 +460,10 @@ class WifiConnections(Box): self.refresh_indicator.set_margin_top(0) self.bounce_timeout_id = None # Hide indicator after animation if not scanning and not pulling - if not (self.wifi_service and self.wifi_service.scanning) and not self.is_pulling: + if ( + not (self.wifi_service and self.wifi_service.scanning) + and not self.is_pulling + ): self.refresh_indicator.set_visible(False) return False @@ -419,7 +477,6 @@ class WifiConnections(Box): bounce_offset = int(self.bounce_amplitude * (1 - eased_progress)) self.refresh_indicator.set_margin_top(max(0, bounce_offset)) - self.bounce_frame += 1 return True @@ -432,4 +489,5 @@ class WifiConnections(Box): def on_destroy(self, widget): """Cleanup animations when widget is destroyed""" - self.stop_bounce_animation() \ No newline at end of file + self.stop_bounce_animation() + diff --git a/modules/panel/components/indicators.py b/modules/panel/components/indicators.py index 9aea6c3..44dd374 100644 --- a/modules/panel/components/indicators.py +++ b/modules/panel/components/indicators.py @@ -3,8 +3,9 @@ from fabric.utils import get_relative_path from fabric.widgets.box import Box from fabric.widgets.button import Button from fabric.widgets.svg import Svg -from services.network import NetworkService + from services.battery import Battery +from services.network import NetworkService from utils.roam import modus_service @@ -155,6 +156,8 @@ class NetworkIndicator(Box): wlan_state += f":{wifi.strength}%" elif wifi.internet == "activating": wlan_state = f"connecting:{wifi.ssid}" + elif wifi.internet in ["deactivating", "deactivated", "disconnected"]: + wlan_state = "No Connection" else: wlan_state = "enabled" @@ -189,6 +192,9 @@ class NetworkIndicator(Box): elif wifi.internet == "activating": icon_file = "wifi.svg" tooltip = f"Connecting to {wifi.ssid}..." + elif wifi.internet in ["deactivating", "deactivated", "disconnected"]: + icon_file = "wifi-off.svg" + tooltip = "WiFi disconnected" else: icon_file = "wifi-off.svg" tooltip = "WiFi disconnected" diff --git a/services/network.py b/services/network.py index c6fcf4b..0a1a85e 100644 --- a/services/network.py +++ b/services/network.py @@ -221,13 +221,95 @@ class Wifi(Service): if not ssid: logger.error("[NetworkService] SSID cannot be empty") return False + try: - subprocess.run(["nmcli", "con", "down", ssid], check=True) - return True - except subprocess.CalledProcessError as e: + # Method 1: Try to disconnect using device-based approach (more reliable) + if self._device and self._device.get_active_connection(): + active_connection = self._device.get_active_connection() + # Verify this is the connection we want to disconnect + if self._ap and NM.utils_ssid_to_utf8(self._ap.get_ssid().get_data()) == ssid: + try: + # Use NetworkManager API to deactivate the connection + self._client.deactivate_connection_async( + active_connection, + None, + lambda client, result: self._disconnect_finished(client, result, ssid) + ) + logger.info(f"[NetworkService] Initiated disconnect from {ssid}") + + # Force immediate state update to reflect disconnection + from gi.repository import GLib + GLib.timeout_add(100, lambda: self._force_state_update()) + + return True + except Exception as e: + logger.warning(f"[NetworkService] API disconnect failed: {e}, trying nmcli") + + # Method 2: Fallback to nmcli with proper connection ID resolution + try: + # First, get the connection ID that matches the SSID + result = subprocess.check_output( + ["nmcli", "-t", "-f", "NAME,TYPE", "connection", "show", "--active"], + text=True + ) + + connection_id = None + for line in result.strip().split('\n'): + if line: + name, conn_type = line.split(':', 1) + if conn_type == "802-11-wireless" and name == ssid: + connection_id = name + break + + if connection_id: + subprocess.run(["nmcli", "con", "down", connection_id], check=True) + logger.info(f"[NetworkService] Disconnected from {ssid} using connection ID") + + # Force immediate state update + from gi.repository import GLib + GLib.timeout_add(100, lambda: self._force_state_update()) + + return True + else: + # Method 3: Try disconnecting by device + device_name = self._device.get_iface() if self._device else None + if device_name: + subprocess.run(["nmcli", "device", "disconnect", device_name], check=True) + logger.info(f"[NetworkService] Disconnected device {device_name}") + + # Force immediate state update + from gi.repository import GLib + GLib.timeout_add(100, lambda: self._force_state_update()) + + return True + else: + logger.error(f"[NetworkService] No active connection found for SSID: {ssid}") + return False + + except subprocess.CalledProcessError as e: + logger.error(f"[NetworkService] nmcli disconnect failed: {e}") + return False + + except Exception as e: logger.error(f"[NetworkService] Failed disconnecting from network: {e}") return False + def _disconnect_finished(self, client, result, ssid): + """Handle async disconnect completion""" + try: + client.deactivate_connection_finish(result) + logger.info(f"[NetworkService] Successfully disconnected from {ssid}") + # Force state update after async completion + self._force_state_update() + except Exception as e: + logger.error(f"[NetworkService] Async disconnect failed: {e}") + + def _force_state_update(self): + """Force an immediate state update to reflect changes""" + # Trigger all relevant property notifications + self.ap_update() + return False # Remove timeout + @Property(bool, "read-write", default_value=False) def enabled(self) -> bool: # noqa: F811 return bool(self._client.wireless_get_enabled()) @@ -267,13 +349,17 @@ class Wifi(Service): @Property(int, "readable") def internet(self): + active_connection = self._device.get_active_connection() + if not active_connection: + return "disconnected" + return { NM.ActiveConnectionState.ACTIVATED: "activated", NM.ActiveConnectionState.ACTIVATING: "activating", NM.ActiveConnectionState.DEACTIVATING: "deactivating", NM.ActiveConnectionState.DEACTIVATED: "deactivated", }.get( - self._device.get_active_connection().get_state(), + active_connection.get_state(), "unknown", ) @@ -307,6 +393,14 @@ class Wifi(Service): @Property(str, "readable") def ssid(self): + # Check if we have an active connection first + active_connection = self._device.get_active_connection() + if not active_connection or active_connection.get_state() in [ + NM.ActiveConnectionState.DEACTIVATED, + NM.ActiveConnectionState.DEACTIVATING + ]: + return "Disconnected" + if not self._ap: return "Disconnected" ssid = self._ap.get_ssid().get_data()