From 25d12a838bc8c712e057b898c822242ab10f46e6 Mon Sep 17 00:00:00 2001 From: S4NKALP Date: Tue, 5 Aug 2025 08:57:44 +0545 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20feat(controlcenter):=20integrate=20?= =?UTF-8?q?network=20and=20bluetooth=20indicators=20with=20control=20cente?= =?UTF-8?q?r?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- modules/controlcenter/bluetooth.py | 19 +- modules/controlcenter/main.py | 1 + modules/controlcenter/wifi.py | 205 ++-- modules/panel/components/indicators.py | 151 ++- services/network.py | 1363 +++++++++--------------- styles/controlcenter.css | 9 + 6 files changed, 732 insertions(+), 1016 deletions(-) diff --git a/modules/controlcenter/bluetooth.py b/modules/controlcenter/bluetooth.py index d1c7761..9f97f79 100644 --- a/modules/controlcenter/bluetooth.py +++ b/modules/controlcenter/bluetooth.py @@ -54,7 +54,7 @@ class BluetoothDeviceSlot(CenterBox): class BluetoothConnections(Box): - def __init__(self, parent, show_hidden_devices: bool = False, **kwargs): + def __init__(self, parent, show_hidden_devices: bool = False, show_back_button=True, **kwargs): super().__init__( spacing=4, orientation="vertical", @@ -67,15 +67,20 @@ class BluetoothConnections(Box): self.client = BluetoothClient(on_device_added=self.on_device_added) - self.title = Box( - orientation="h", - children=[ + # Create title with optional back button + title_children = [] + if show_back_button: + title_children.append( Button( image=Image(icon_name="back", size=10), on_clicked=lambda *_: self.parent.close_bluetooth(), - ), - Label("Bluetooth"), - ], + ) + ) + title_children.append(Label("Bluetooth")) + + self.title = Box( + orientation="h", + children=title_children, ) self.toggle_button = Gtk.Switch(visible=True, name="toggle-button") diff --git a/modules/controlcenter/main.py b/modules/controlcenter/main.py index 8837d02..71f3a82 100644 --- a/modules/controlcenter/main.py +++ b/modules/controlcenter/main.py @@ -104,6 +104,7 @@ class ModusControlCenter(Window): # Disable brightness scale if no backlight device available self.brightness_scale.set_sensitive(False) + # Create music widget with ultra-lazy player container self.music_widget = Box( name="music-widget", children=[PlayerBoxStack(MprisPlayerManager())], diff --git a/modules/controlcenter/wifi.py b/modules/controlcenter/wifi.py index 7c7dd58..a7c4945 100644 --- a/modules/controlcenter/wifi.py +++ b/modules/controlcenter/wifi.py @@ -6,31 +6,25 @@ 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 services.network import NetworkService +from services.network import NetworkClient from widgets.wifi_password_dialog import WiFiPasswordDialog class WifiNetworkSlot(CenterBox): - def __init__(self, network_data: dict, wifi_service, parent=None, **kwargs): + def __init__(self, access_point, wifi_service, parent=None, **kwargs): super().__init__(**kwargs) - self.network_data = network_data + self.access_point = access_point self.wifi_service = wifi_service self.parent = parent # Reference to control center - # Get network info - 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" - ) + # Get network info from AccessPoint object + self.ssid = access_point.ssid + self.bssid = access_point.bssid + self.strength = access_point.strength + self.icon_name = access_point.icon - # 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 + # Check if this network is currently connected + self.is_connected = access_point.is_active # Initialize styles based on connection state self.styles = [ @@ -61,11 +55,8 @@ class WifiNetworkSlot(CenterBox): def toggle_connecting(self): """Toggle WiFi network connection""" - # 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 + # Check if this network is currently connected + is_currently_connected = self.access_point.is_active if is_currently_connected: # Show disconnecting state @@ -73,32 +64,34 @@ class WifiNetworkSlot(CenterBox): 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 + self.wifi_service.disconnect_wifi() + 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() + GLib.timeout_add(500, lambda: self._reset_disconnect_state()) else: - # Try to connect without password first (for open networks or saved connections) - self.dimage.set_property("icon-name", "network-wireless-acquiring-symbolic") - self.dimage.add_style_class("connecting") - - 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: - # Connection failed, likely needs password - show password dialog - self._reset_connect_state() + # Try to connect - check if password is required + if self.access_point.requires_password: + # Show password dialog immediately self._show_password_dialog() + else: + # Try to connect without password (for open networks) + self.dimage.set_property( + "icon-name", "network-wireless-acquiring-symbolic" + ) + self.dimage.add_style_class("connecting") + + success = self.wifi_service.connect_to_wifi(self.access_point) + 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: + # Connection failed + self._reset_connect_state() # Update display after connection attempt self.on_changed() @@ -119,19 +112,8 @@ class WifiNetworkSlot(CenterBox): 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: - # 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 + # Check if this network is currently connected using the access point's is_active property + self.is_connected = self.access_point.is_active self.styles = [ "connected" if self.is_connected else "", ] @@ -165,7 +147,7 @@ class WifiNetworkSlot(CenterBox): # Try to connect with password try: - success = self.wifi_service.connect_network(ssid, password) + success = self.wifi_service.connect_to_wifi(self.access_point, password) if success: self.is_connected = True # Remove connecting state after a short delay @@ -201,7 +183,7 @@ class WifiNetworkSlot(CenterBox): class WifiConnections(Box): - def __init__(self, parent, **kwargs): + def __init__(self, parent, show_back_button=True, **kwargs): super().__init__( spacing=4, orientation="vertical", @@ -210,12 +192,12 @@ class WifiConnections(Box): ) self.parent = parent - self.network_service = NetworkService() + self.network_service = NetworkClient() self.wifi_service = None self.is_scanning = False # Track scanning state # Wait for network service to be ready - self.network_service.connect("device-ready", self.on_network_ready) + self.network_service.connect("wifi-device-added", self.on_network_ready) # Create pull-to-refresh indicator self.refresh_indicator = Label( @@ -226,15 +208,20 @@ class WifiConnections(Box): style="color: #fff; font-size: 12px; padding: 5px;", ) - self.title = Box( - orientation="h", - children=[ + # Create title with optional back button + title_children = [] + if show_back_button: + title_children.append( Button( image=Image(icon_name="back", size=10), on_clicked=lambda *_: self.parent.close_wifi(), - ), - Label("WiFi"), - ], + ) + ) + title_children.append(Label("WiFi")) + + self.title = Box( + orientation="h", + children=title_children, ) self.toggle_button = Gtk.Switch(visible=True, name="toggle-button") @@ -276,43 +263,31 @@ class WifiConnections(Box): self.wifi_service = self.network_service.wifi_device if self.wifi_service: # Set up WiFi toggle - self.toggle_button.set_active(self.wifi_service.enabled) + self.toggle_button.set_active(self.wifi_service.wireless_enabled) self.toggle_button.connect( "notify::active", lambda *_: setattr( - self.wifi_service, "enabled", self.toggle_button.get_active() + self.wifi_service, + "wireless_enabled", + self.toggle_button.get_active(), ), ) # Connect to WiFi service signals - self.wifi_service.connect("notify::enabled", self.on_wifi_enabled_changed) - 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 + "notify::wireless-enabled", self.on_wifi_enabled_changed ) - # 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) + self.wifi_service.connect("changed", self.update_networks) + self.wifi_service.connect("ap-added", self.update_networks) + self.wifi_service.connect("ap-removed", self.update_networks) # Initial network update self.update_networks() - # Initialize scanning state - self.wifi_service.notify("scanning") - def on_wifi_enabled_changed(self, *_): """Handle WiFi enabled state changes""" if self.wifi_service: - self.toggle_button.set_active(self.wifi_service.enabled) - - def on_scanning_changed(self, *_): - """Handle scanning state changes""" - if self.wifi_service: - is_scanning = self.wifi_service.scanning - self.is_scanning = is_scanning # Update our tracking state - self.update_scan_label() + self.toggle_button.set_active(self.wifi_service.wireless_enabled) def update_networks(self, *_): """Update the list of available networks""" @@ -325,10 +300,10 @@ class WifiConnections(Box): # Add current networks access_points = self.wifi_service.access_points - for ap_data in access_points: - if ap_data.get("ssid") and ap_data.get("ssid") != "Unknown": + for access_point in access_points: + if access_point.ssid and access_point.ssid != "Unknown": network_slot = WifiNetworkSlot( - ap_data, self.wifi_service, parent=self.parent + access_point, self.wifi_service, parent=self.parent ) self.available_networks.add(network_slot) @@ -377,8 +352,8 @@ class WifiConnections(Box): # Only handle pull-to-refresh when at the top if self.vadjustment.get_value() <= 0: if event.direction == Gdk.ScrollDirection.UP: - # Scrolling up at the top - start scan (only if not already scanning) - if self.wifi_service and not self.is_scanning: + # Scrolling up at the top - start scan + if self.wifi_service: self.wifi_service.scan() # Trigger smooth bounce animation for scroll self.start_bounce_animation() @@ -397,8 +372,8 @@ class WifiConnections(Box): if self.is_pulling: pull_distance = event.y - self.pull_start_y if pull_distance > self.pull_threshold: - # Start scan (only if not already scanning) - if self.wifi_service and not self.is_scanning: + # Start scan + if self.wifi_service: self.wifi_service.scan() # Trigger elastic animation for pull gesture (most satisfying) self.start_elastic_animation() @@ -419,36 +394,15 @@ class WifiConnections(Box): # Show refresh indicator when pulling down self.refresh_indicator.set_visible(True) if pull_distance >= self.pull_threshold: - if self.is_scanning: - self.refresh_indicator.set_label("↑ Already scanning...") - else: - self.refresh_indicator.set_label("↑ Release to scan") + self.refresh_indicator.set_label("↑ Release to scan") self.refresh_indicator.add_style_class("ready-to-refresh") else: - if self.is_scanning: - self.refresh_indicator.set_label("↓ Scanning in progress...") - else: - self.refresh_indicator.set_label("↓ Pull to scan for networks") + self.refresh_indicator.set_label("↓ Pull to scan for networks") self.refresh_indicator.remove_style_class("ready-to-refresh") else: self.refresh_indicator.set_visible(False) return False - def update_scan_label(self): - """Update scanning state appearance""" - if self.wifi_service and self.wifi_service.scanning: - # Show scanning feedback in refresh indicator - - self.refresh_indicator.set_label("Scanning for networks...") - self.refresh_indicator.set_visible(True) - self.refresh_indicator.add_style_class("scanning") - self.refresh_indicator.show_all() - else: - # Hide scanning feedback only if not animating and not pulling - if not self.is_pulling and not self.bounce_timeout_id: - self.refresh_indicator.set_visible(False) - self.refresh_indicator.remove_style_class("scanning") - def ease_out_bounce(self, t): """Smooth bounce easing function (0 to 1)""" if t < 1 / 2.75: @@ -489,11 +443,8 @@ class WifiConnections(Box): # Animation finished 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 - ): + # Hide indicator after animation if not pulling + if not self.is_pulling: self.refresh_indicator.set_visible(False) return False @@ -527,11 +478,8 @@ class WifiConnections(Box): # Animation finished 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 - ): + # Hide indicator after animation if not pulling + if not self.is_pulling: self.refresh_indicator.set_visible(False) return False @@ -558,4 +506,3 @@ class WifiConnections(Box): def on_destroy(self, widget): """Cleanup animations when widget is destroyed""" self.stop_bounce_animation() - diff --git a/modules/panel/components/indicators.py b/modules/panel/components/indicators.py index 4a46eb6..02831c4 100644 --- a/modules/panel/components/indicators.py +++ b/modules/panel/components/indicators.py @@ -4,8 +4,12 @@ from fabric.widgets.box import Box from fabric.widgets.button import Button from fabric.widgets.svg import Svg from services.battery import Battery -from services.network import NetworkService +from services.network import NetworkClient from utils.roam import modus_service +from modules.controlcenter.wifi import WifiConnections +from modules.controlcenter.bluetooth import BluetoothConnections +from widgets.mousecapture import DropDownMouseCapture +from widgets.wayland import WaylandWindow as Window class BluetoothIndicator(Box): @@ -21,10 +25,32 @@ class BluetoothIndicator(Box): ), ) - self.bt_button = Button(name="bt-button", child=self.bt_icon) + self.bt_button = Button( + name="bt-button", child=self.bt_icon, on_clicked=self.on_bluetooth_clicked + ) self.add(self.bt_button) + # Create Bluetooth control center widget + self.bluetooth_window = Window( + layer="overlay", + title="modus", + anchor="top right", + margin="2px 10px 0px 0px", + exclusivity="auto", + keyboard_mode="on-demand", + name="bluetooth-control-window", + visible=False, + ) + + self.bluetooth_widget = BluetoothConnections(self, show_back_button=False) + self.bluetooth_window.children = [self.bluetooth_widget] + + # Create mouse capture for Bluetooth widget + self.bluetooth_mousecapture = DropDownMouseCapture( + layer="top", child_window=self.bluetooth_window + ) + modus_service.connect("bluetooth-changed", self.on_bluetooth_changed) self.bluetooth.connect("changed", self.on_bluetooth_direct_changed) self.bluetooth.connect("device-added", self.on_device_added) @@ -107,12 +133,24 @@ class BluetoothIndicator(Box): modus_service.bluetooth = bluetooth_state + def on_bluetooth_clicked(self, *args): + """Handle Bluetooth indicator click""" + self.bluetooth_mousecapture.toggle_mousecapture() + + def close_bluetooth(self, *args): + """Close Bluetooth control center""" + self.bluetooth_mousecapture.hide_child_window() + + def hide_controlcenter(self, *args): + """Hide Bluetooth control center""" + self.bluetooth_mousecapture.hide_child_window() + class NetworkIndicator(Box): def __init__(self, **kwargs): super().__init__(name="network-indicator", orientation="h", **kwargs) - self.network_service = NetworkService() + self.network_service = NetworkClient() self.network_icon = Svg( name="network-icon", @@ -122,11 +160,40 @@ class NetworkIndicator(Box): ), ) - self.network_button = Button(name="network-button", child=self.network_icon) + self.network_button = Button( + name="network-button", + child=self.network_icon, + on_clicked=self.on_wifi_clicked, + ) self.add(self.network_button) + + # Create WiFi control center widget + self.wifi_window = Window( + layer="overlay", + title="modus", + anchor="top right", + margin="2px 10px 0px 0px", + exclusivity="auto", + keyboard_mode="on-demand", + name="wifi-control-window", + visible=False, + ) + + self.wifi_widget = WifiConnections(self, show_back_button=False) + self.wifi_window.children = [self.wifi_widget] + + # Create mouse capture for WiFi widget + self.wifi_mousecapture = DropDownMouseCapture( + layer="top", child_window=self.wifi_window + ) + modus_service.connect("wlan-changed", self.on_wlan_changed) - self.network_service.connect("device-ready", self.on_device_ready) + self.network_service.connect("wifi-device-added", self.on_wifi_device_added) + self.network_service.connect( + "ethernet-device-added", self.on_ethernet_device_added + ) + self.network_service.connect("changed", self.on_network_changed) self.update_modus_service_wlan_state() self.update_state() @@ -134,17 +201,21 @@ class NetworkIndicator(Box): def on_wlan_changed(self, service, new_wlan_state): self.update_state() - def on_device_ready(self, *args): + def on_wifi_device_added(self, *args): + """Called when WiFi device is added""" if self.network_service.wifi_device: self.network_service.wifi_device.connect( "changed", self.on_network_direct_changed ) + self.update_modus_service_wlan_state() + self.update_state() + def on_ethernet_device_added(self, *args): + """Called when Ethernet device is added""" if self.network_service.ethernet_device: self.network_service.ethernet_device.connect( "changed", self.on_network_direct_changed ) - self.update_modus_service_wlan_state() self.update_state() @@ -157,30 +228,28 @@ class NetworkIndicator(Box): self.update_state() def update_modus_service_wlan_state(self): - primary_device = self.network_service.primary_device wlan_state = "disconnected" - if primary_device == "wifi" and self.network_service.wifi_device: + # Check WiFi first (prioritize WiFi over Ethernet) + if self.network_service.wifi_device: wifi = self.network_service.wifi_device - if not wifi.enabled: + if not wifi.wireless_enabled: wlan_state = "disabled" - elif wifi.internet == "activated": - wlan_state = f"connected:{wifi.ssid}" - if wifi.strength >= 0: - 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" + elif wifi.active_access_point: + ap = wifi.active_access_point + wlan_state = f"connected:{ap.ssid}" + if ap.strength >= 0: + wlan_state += f":{ap.strength}%" else: wlan_state = "enabled" - elif primary_device == "wired" and self.network_service.ethernet_device: + # Check Ethernet if WiFi is not connected + elif self.network_service.ethernet_device: ethernet = self.network_service.ethernet_device if ethernet.internet == "activated": wlan_state = "ethernet:connected" - if hasattr(ethernet, "speed") and ethernet.speed > 0: - wlan_state += f":{ethernet.speed}Mbps" + if hasattr(ethernet, "speed") and ethernet.speed: + wlan_state += f":{ethernet.speed}" elif ethernet.internet == "activating": wlan_state = "ethernet:connecting" else: @@ -189,37 +258,33 @@ class NetworkIndicator(Box): modus_service.wlan = wlan_state def update_state(self): - primary_device = self.network_service.primary_device tooltip = "No network connection" icon_file = "wifi-off-clear.svg" - if primary_device == "wifi" and self.network_service.wifi_device: + # Check WiFi first (prioritize WiFi over Ethernet) + if self.network_service.wifi_device: wifi = self.network_service.wifi_device - if not wifi.enabled: + if not wifi.wireless_enabled: icon_file = "wifi-off-clear.svg" tooltip = "WiFi disabled" - elif wifi.internet == "activated": + elif wifi.active_access_point: + ap = wifi.active_access_point icon_file = "wifi-clear.svg" - tooltip = f"Connected to {wifi.ssid}" - if wifi.strength >= 0: - tooltip += f" ({wifi.strength}%)" - elif wifi.internet == "activating": - icon_file = "wifi-clear.svg" - tooltip = f"Connecting to {wifi.ssid}..." - elif wifi.internet in ["deactivating", "deactivated", "disconnected"]: - icon_file = "wifi-off-clear.svg" - tooltip = "WiFi disconnected" + tooltip = f"Connected to {ap.ssid}" + if ap.strength >= 0: + tooltip += f" ({ap.strength}%)" else: icon_file = "wifi-off-clear.svg" tooltip = "WiFi disconnected" - elif primary_device == "wired" and self.network_service.ethernet_device: + # Check Ethernet if WiFi is not connected + elif self.network_service.ethernet_device: ethernet = self.network_service.ethernet_device if ethernet.internet == "activated": icon_file = "network-wired.svg" tooltip = "Ethernet connected" - if hasattr(ethernet, "speed") and ethernet.speed > 0: - tooltip += f" ({ethernet.speed} Mbps)" + if hasattr(ethernet, "speed") and ethernet.speed: + tooltip += f" ({ethernet.speed})" elif ethernet.internet == "activating": icon_file = "network-wired.svg" tooltip = "Ethernet connecting..." @@ -232,6 +297,18 @@ class NetworkIndicator(Box): ) self.network_button.set_tooltip_text(tooltip) + def on_wifi_clicked(self, *args): + """Handle WiFi indicator click""" + self.wifi_mousecapture.toggle_mousecapture() + + def close_wifi(self, *args): + """Close WiFi control center""" + self.wifi_mousecapture.hide_child_window() + + def hide_controlcenter(self, *args): + """Hide WiFi control center""" + self.wifi_mousecapture.hide_child_window() + class BatteryIndicator(Box): def __init__(self, **kwargs): diff --git a/services/network.py b/services/network.py index a9d96ec..59f5e01 100644 --- a/services/network.py +++ b/services/network.py @@ -1,21 +1,277 @@ -import os -import subprocess -from typing import Any, List, Literal - +from gi.repository import NM, GLib import gi -from gi.repository import Gio +from typing import List, Optional +from fabric.core.service import Property, Service, Signal +from fabric.utils import bulk_connect, get_enum_member_name, snake_case_to_kebab_case from loguru import logger -from fabric.core.service import Property, Service, Signal -from fabric.utils import bulk_connect +gi.require_version("NM", "1.0") # Ensure the correct version is loaded -# FIX: HAHA Could be better if loc is less -try: - gi.require_version("NM", "1.0") - from gi.repository import NM -except ValueError: - raise logger.error("Failed to start network manager") +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 class Wifi(Service): @@ -25,694 +281,232 @@ class Wifi(Service): def changed(self) -> None: ... @Signal - def enabled(self) -> bool: ... + def ap_added(self, ap: AccessPoint) -> None: ... @Signal - def scanning(self, is_scanning: bool) -> None: ... + def ap_removed(self, ap: AccessPoint) -> 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 - self._scan_cancellable: Gio.Cancellable | None = None - self._is_scanning: bool = False + @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 + self._device: NM.DeviceWifi = device + self._active_access_point: NM.AccessPoint | None = None + self._access_points: dict[str, AccessPoint] = {} - self._client.connect( - "notify::wireless-enabled", - lambda *args: self.notifier("enabled", args), + 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), + }, ) - if self._device: - bulk_connect( - self._device, - { - "notify::active-access-point": self._activate_ap, - "access-point-added": lambda *_: self.emit("changed"), - "access-point-removed": lambda *_: self.emit("changed"), - "state-changed": self.ap_update, - }, - ) - self._activate_ap() - def ap_update(self, *_): + self._client.connect("notify::wireless-enabled", + lambda *args: self.notifier('wireless-enabled')) + + 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") - 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 + def do_get_access_points(self): + return self._device.get_access_points() - self._ap_signal = self._ap.connect( - "notify::strength", lambda *args: self.ap_update() - ) # type: ignore + def on_access_point_added(self, ap): + ssid = ap.get_ssid() + ssid = NM.utils_ssid_to_utf8(ssid.get_data()) if ssid else "Unknown" - def toggle_wifi(self): - self._client.wireless_set_enabled(not self._client.wireless_get_enabled()) + access_point: AccessPoint = AccessPoint( + ap=ap, + device=self + ) - def get_ap_security( - self, nm_ap: NM.AccessPoint - ) -> Literal["WEP", "WPA1", "WPA2", "802.1X", "unsecured"]: - """Parse the security flags to return a string with 'WPA2', etc.""" - flags = nm_ap.get_flags() - wpa_flags = nm_ap.get_wpa_flags() - rsn_flags = nm_ap.get_rsn_flags() - sec_str = "" - if ( - (flags & getattr(NM, "80211ApFlags").PRIVACY) - and (wpa_flags == 0) - and (rsn_flags == 0) - ): - sec_str += " WEP" - if wpa_flags != 0: - sec_str += " WPA1" - if rsn_flags != 0: - sec_str += " WPA2" - if (wpa_flags & getattr(NM, "80211ApSecurityFlags").KEY_MGMT_802_1X) or ( - rsn_flags & getattr(NM, "80211ApSecurityFlags").KEY_MGMT_802_1X - ): - sec_str += " 802.1X" + self._access_points[ssid] = access_point - # If there is no security use "--" - if sec_str == "": - sec_str = "unsecured" - return sec_str.lstrip() + 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 scan(self): - """Start scanning for WiFi networks and emit scanning signal""" - if self._device and not self._is_scanning: - self._is_scanning = True - self._scan_cancellable = Gio.Cancellable() - self.notify("scanning") # Notify property change - self._device.request_scan_async( - self._scan_cancellable, - lambda device, result: self._scan_finished(device, result), + self._device.request_scan_async( + None, + lambda device, result: [ + device.request_scan_finish(result), + self.notifier('access-points'), + ], + ) + 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): + """Connect to a WiFi network.""" + + ssid = ap.ssid + + if ssid == "Unknown": + logger.error("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 + + 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") + 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, ssid, device): + try: + new_connection = client.add_connection_finish(result) + if not new_connection: + logger.error( + f"Failed to create connection for '{ssid}'") + return + + logger.info(f"Connection for '{ssid}' added successfully") + + client.activate_connection_async( + new_connection, device, None, None, + lambda c, r: logger.info(f"Connected to '{ssid}'") if c else logger.error( + f"Failed to connect to '{ssid}'") + ) + except Exception as e: + logger.error(f"Failed to add connection: {e}") + + # Save the new connection + self._client._client.add_connection_async( + connection, + True, + None, + lambda c, r: on_connection_added(c, r, ssid, self._device) ) - def _scan_finished(self, device, result): - """Handle scan completion""" - try: - device.request_scan_finish(result) - except Exception as e: - logger.error(f"WiFi scan error: {e}") - finally: - self._is_scanning = False - self._scan_cancellable = None - self.notify("scanning") # Notify property change + # Activate the connection + self._client._client.activate_connection_async( + connection, self._device, None, None, + lambda client, result: logger.info(f"Connected to '{ssid}'") + ) - def toggle_scan(self): - """Toggle WiFi scanning on/off""" - if self._is_scanning: - # Stop current scan - if self._scan_cancellable: - self._scan_cancellable.cancel() - self._scan_cancellable = None - self._is_scanning = False - self.notify("scanning") # Notify property change - else: - # Start new scan - self.scan() + return True - @Property(bool, "readable", default_value=False) - def scanning(self) -> bool: - """Check if currently scanning""" - return self._is_scanning - - def is_active_ap(self, name) -> bool: - return self._ap.get_bssid() == name if self._ap else False - - def notifier(self, name: str, *_): + def notifier(self, name: str, *args): self.notify(name) self.emit("changed") return - def forget_access_point(self, ssid): - try: - # List all saved connections - result = subprocess.check_output( - "nmcli connection show", shell=True, text=True - ) - - # Find connection ID that matches SSID - for line in result.splitlines(): - if ssid in line: - connection_id = line.split()[0] - subprocess.check_call( - f"nmcli connection delete id '{connection_id}'", shell=True - ) - logger.info( - f"[NetworkService] Deleted saved connection: {connection_id}" - ) - return True - - logger.warning( - f"[NetworkService] No saved connection found for SSID: {ssid}" - ) - return False - - except subprocess.CalledProcessError as e: - logger.error(f"[NetworkService] Error forgetting connection: {e}") - return False - - def connect_network( - self, ssid: str, password: str = "", remember: bool = True - ) -> bool: - """Connect to a WiFi network""" - if not ssid: - logger.error("[NetworkService] SSID cannot be empty") - return False - - # Check if nm-applet is running before we kill it - nm_applet_was_running = self._is_nm_applet_running() - - # Kill any running NetworkManager authentication agents to prevent GUI dialogs - try: - subprocess.run(["pkill", "-f", "nm-applet"], capture_output=True) - subprocess.run(["pkill", "-f", "nm-connection-editor"], capture_output=True) - subprocess.run( - ["pkill", "-f", "polkit-gnome-authentication-agent"], - capture_output=True, - ) - except: - pass # Ignore errors if processes don't exist - - try: - # First try to connect using saved connection (suppress GUI) - try: - env = os.environ.copy() - env.update( - { - "DISPLAY": "", - "WAYLAND_DISPLAY": "", - "XDG_SESSION_TYPE": "tty", - } - ) - subprocess.run( - ["nmcli", "--terse", "con", "up", ssid], - check=True, - capture_output=True, - env=env, - ) - return True - except subprocess.CalledProcessError: - # If saved connection fails, try with password if provided - if password: - # Create environment without GUI components to prevent dialogs - env = os.environ.copy() - env.update( - { - "DISPLAY": "", # No X11 display - "WAYLAND_DISPLAY": "", # No Wayland display - "XDG_SESSION_TYPE": "tty", # Force TTY session - "DESKTOP_SESSION": "", # No desktop session - "NM_EDITOR": "/bin/false", # Disable editor - "NM_POLKIT_AGENT": "/bin/false", # Disable polkit agent - } - ) - - # First, try to delete any existing connection with the same name to avoid conflicts - try: - subprocess.run( - ["nmcli", "connection", "delete", ssid], - capture_output=True, - env=env, - ) - except: - pass # Ignore if connection doesn't exist - - # Create a temporary connection profile with the password - # Use a unique temporary name to avoid conflicts - temp_connection_name = f"temp_{ssid}_{os.getpid()}" - - create_cmd = [ - "nmcli", - "connection", - "add", - "type", - "wifi", - "con-name", - temp_connection_name, - "ssid", - ssid, - "wifi-sec.key-mgmt", - "wpa-psk", - "wifi-sec.psk", - password, - "connection.autoconnect", - "no", # Don't auto-connect - ] - - create_result = subprocess.run( - create_cmd, capture_output=True, text=True, env=env - ) - - if create_result.returncode == 0: - # Try to activate the temporary connection with shorter timeout - activate_cmd = [ - "nmcli", - "--wait", - "3", - "connection", - "up", - temp_connection_name, - ] - try: - activate_result = subprocess.run( - activate_cmd, - capture_output=True, - text=True, - env=env, - timeout=5, # 5 second timeout for faster failure detection - ) - except subprocess.TimeoutExpired: - # Connection timed out - treat as failure - activate_result = subprocess.CompletedProcess( - activate_cmd, 1, "", "Connection timeout" - ) - - if activate_result.returncode == 0: - # Connection successful! - if remember: - # Rename the temporary connection to the final name - rename_cmd = [ - "nmcli", - "connection", - "modify", - temp_connection_name, - "connection.id", - ssid, - "connection.autoconnect", - "yes", - ] - subprocess.run(rename_cmd, capture_output=True, env=env) - else: - # Don't remember - delete the temporary connection - subprocess.run( - [ - "nmcli", - "connection", - "delete", - temp_connection_name, - ], - capture_output=True, - env=env, - ) - return True - else: - # Connection failed - delete the temporary connection immediately - # This ensures incorrect passwords are never saved - subprocess.run( - ["nmcli", "connection", "delete", temp_connection_name], - capture_output=True, - env=env, - ) - return False - else: - # Failed to create connection profile - return False - - # Fallback: try the old method but always use temporary connections first - # This ensures incorrect passwords are never permanently saved - fallback_cmd = [ - "nmcli", - "--terse", - "--wait", - "3", # Reduced wait time for faster response - "device", - "wifi", - "connect", - ssid, - "password", - password, - "--temporary", # Always use temporary for initial attempt - ] - - # Set additional environment variables to suppress dialogs - env.update( - { - "SSH_ASKPASS": "/bin/false", - "GIT_ASKPASS": "/bin/false", - "SUDO_ASKPASS": "/bin/false", - "NM_POLKIT_AGENT": "", - } - ) - - result = subprocess.run( - fallback_cmd, - capture_output=True, - text=True, - env=env, - stdin=subprocess.DEVNULL, - ) - - # If connection succeeded and we want to remember it - if result.returncode == 0 and remember: - # Create a permanent connection profile - save_cmd = [ - "nmcli", - "connection", - "add", - "type", - "wifi", - "con-name", - ssid, - "ssid", - ssid, - "wifi-sec.key-mgmt", - "wpa-psk", - "wifi-sec.psk", - password, - "connection.autoconnect", - "yes", - ] - subprocess.run(save_cmd, capture_output=True, env=env) - - return result.returncode == 0 - return False - except subprocess.CalledProcessError as e: - logger.error(f"[NetworkService] Failed connecting to network: {e}") - return False - finally: - # Cleanup: Remove any leftover temporary connections for this process - self._cleanup_temp_connections() - - # Restart nm-applet if it was running before we killed it - if nm_applet_was_running: - self._restart_nm_applet() - - def _is_nm_applet_running(self): - """Check if nm-applet is currently running""" - try: - result = subprocess.run( - ["pgrep", "-f", "nm-applet"], capture_output=True, text=True - ) - return result.returncode == 0 - except Exception: - return False - - def _restart_nm_applet(self): - """Restart nm-applet with indicator support""" - try: - # Small delay to ensure connection process is complete - import time - - time.sleep(0.5) - - # Start nm-applet in the background with indicator support - subprocess.Popen( - ["nm-applet", "--indicator"], - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - start_new_session=True, # Detach from parent process - ) - logger.debug("[NetworkService] Restarted nm-applet with indicator support") - except Exception as e: - logger.debug(f"[NetworkService] Failed to restart nm-applet: {e}") - - def _cleanup_temp_connections(self): - """Clean up any temporary connections created by this process""" - try: - # List all connections and find temporary ones created by this process - result = subprocess.run( - ["nmcli", "-t", "-f", "NAME", "connection", "show"], - capture_output=True, - text=True, - ) - - if result.returncode == 0: - connections = result.stdout.strip().split("\n") - temp_prefix = "temp_" - process_suffix = f"_{os.getpid()}" - - for connection in connections: - if connection.startswith(temp_prefix) and connection.endswith( - process_suffix - ): - # Delete this temporary connection - subprocess.run( - ["nmcli", "connection", "delete", connection], - capture_output=True, - ) - logger.debug( - f"[NetworkService] Cleaned up temporary connection: { - connection - }" - ) - except Exception as e: - logger.debug(f"[NetworkService] Error during temp connection cleanup: {e}") - - def disconnect_network(self, ssid: str) -> bool: - """Disconnect from a WiFi network""" - if not ssid: - logger.error("[NetworkService] SSID cannot be empty") - return False - - try: - # 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()) - - @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(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( - active_connection.get_state(), - "unknown", - ) - - @Property(object, "readable") - def access_points(self) -> List[object]: - points: list[NM.AccessPoint] = self._device.get_access_points() - - def make_ap_dict(ap: NM.AccessPoint): - return { - "bssid": ap.get_bssid(), - "last_seen": ap.get_last_seen(), - "ssid": ( - NM.utils_ssid_to_utf8(ap.get_ssid().get_data()) - if ap.get_ssid() - else "Unknown" - ), - "active-ap": self._ap, - "strength": ap.get_strength(), - "frequency": ap.get_frequency(), - "icon-name": { - 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(ap.get_strength() / 20)), - "network-wireless-no-route-symbolic", - ), - } - - return list(map(make_ap_dict, points)) - - @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() - 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""" @@ -724,191 +518,74 @@ class Ethernet(Service): def enabled(self) -> bool: ... @Property(int, "readable") - def speed(self) -> int: - return self._device.get_speed() + def speed(self) -> str: + speed_mbps = self._device.get_speed() + return f"{speed_mbps} Mb/s" + + @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: - 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(), - "disconnected", - ) + 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(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 - for names in ( - "active-connection", - "icon-name", - "internet", - "speed", - "state", - ): - self._device.connect(f"notify::{names}", lambda *_: self.notifier(names)) + self._device.connect( + "state-changed", lambda *args: self.on_network_state_changed()) - self._device.connect("notify::speed", lambda *_: print(_)) + self.update_active_connection() - def notifier(self, names): - self.notify(names) - self.emit("changed") + 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 + 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')) -class NetworkService(Service): - """A service to manage network devices""" - - _instance = None - - def __new__(cls): - if cls._instance is None: - cls._instance = super().__new__(cls) - return cls._instance - - @Signal - def device_ready(self) -> None: ... - - def __init__(self, **kwargs): - self._client: NM.Client | None = None - self.wifi_device: Wifi | None = None - self.ethernet_device: Ethernet | None = None - super().__init__(**kwargs) - - NM.Client.new_async( - cancellable=None, - callback=self._init_network_client, - **kwargs, - ) - - def _init_network_client(self, client: NM.Client, task: Gio.Task, **kwargs): - self._client = client - wifi_device: NM.DeviceWifi | None = self._get_device(NM.DeviceType.WIFI) # type: ignore - ethernet_device: NM.DeviceEthernet | None = self._get_device( - NM.DeviceType.ETHERNET - ) - - if wifi_device: - self.wifi_device = Wifi(self._client, wifi_device) - # Clean up any leftover temporary connections from previous runs - self._cleanup_temp_connections() - self.emit("device-ready") - - if ethernet_device: - self.ethernet_device = Ethernet(client=self._client, device=ethernet_device) - self.emit("device-ready") - - self.notify("primary-device") - - def _get_device(self, device_type) -> Any: - devices: List[NM.Device] = self._client.get_devices() # type: ignore - return next( - ( - x - for x in devices - if x.get_device_type() == device_type - and x.get_active_connection() is not None - ), - None, - ) - - def _get_primary_device(self) -> Literal["wifi", "wired"] | None: - if not self._client: - return None - - if self._client.get_primary_connection() is None: - return "wifi" - return ( - "wifi" - if "wireless" - in str(self._client.get_primary_connection().get_connection_type()) - else ( - "wired" - if "ethernet" - in str(self._client.get_primary_connection().get_connection_type()) - else None - ) - ) - - @Property(str, "readable") - def primary_device(self) -> Literal["wifi", "wired"] | None: - return self._get_primary_device() - - def _is_nm_applet_running(self): - """Check if nm-applet is currently running""" + def get_network_stats(self): + """Fetch received and transmitted bytes from the system files""" try: - result = subprocess.run( - ["pgrep", "-f", "nm-applet"], capture_output=True, text=True - ) - return result.returncode == 0 - except Exception: - return False - - def _restart_nm_applet(self): - """Restart nm-applet with indicator support""" - try: - # Small delay to ensure connection process is complete - import time - - time.sleep(0.5) - - # Start nm-applet in the background with indicator support - subprocess.Popen( - ["nm-applet", "--indicator"], - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - start_new_session=True, # Detach from parent process - ) - logger.debug("[NetworkService] Restarted nm-applet with indicator support") - except Exception as e: - logger.debug(f"[NetworkService] Failed to restart nm-applet: {e}") - - def _cleanup_temp_connections(self): - """Clean up any temporary connections created by this process""" - try: - # List all connections and find temporary ones created by this process - result = subprocess.run( - ["nmcli", "-t", "-f", "NAME", "connection", "show"], - capture_output=True, - text=True, - ) - - if result.returncode == 0: - connections = result.stdout.strip().split("\n") - temp_prefix = "temp_" - process_suffix = f"_{os.getpid()}" - - for connection in connections: - if connection.startswith(temp_prefix) and connection.endswith( - process_suffix - ): - # Delete this temporary connection - subprocess.run( - ["nmcli", "connection", "delete", connection], - capture_output=True, - ) - logger.debug( - f"[NetworkService] Cleaned up temporary connection: { - connection - }" - ) - except Exception as e: - logger.debug(f"[NetworkService] Error during temp connection cleanup: {e}") + # 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 diff --git a/styles/controlcenter.css b/styles/controlcenter.css index 9a881d5..98de1fd 100644 --- a/styles/controlcenter.css +++ b/styles/controlcenter.css @@ -5,6 +5,15 @@ margin: 0; } +#bluetooth-control-window, +#wifi-control-window { + margin: 6px; + background-color: alpha(#fff, 0.09); + border: 1px solid alpha(#111, 0.3); + box-shadow: inset 0 0 200px 0 alpha(#111, 0.3); + border-radius: 16px; +} + #control-center-widgets { background-color: alpha(#fff, 0.09); box-shadow: inset 0 0 0 1px alpha(#aaa, 0.4);