diff --git a/modules/launcher/plugins/bluetooth.py b/modules/launcher/plugins/bluetooth.py deleted file mode 100644 index c6989aa..0000000 --- a/modules/launcher/plugins/bluetooth.py +++ /dev/null @@ -1,594 +0,0 @@ -from typing import List - -from fabric.bluetooth import BluetoothClient, BluetoothDevice - -import utils.icons as icons -from modules.launcher.plugin_base import PluginBase -from modules.launcher.result import Result - - -class BluetoothPlugin(PluginBase): - """ - Bluetooth management plugin for the launcher. - """ - - def __init__(self): - super().__init__() - self.display_name = "Bluetooth" - self.description = "Manage Bluetooth devices and connections" - - # Initialize bluetooth client with timeout protection - self.bluetooth_client = None - self._bluetooth_available = False - self._last_query_time = 0 # Track when plugin was last used - self._scan_timeout_id = None # Track auto-stop scan timeout - self._init_bluetooth_client() - - def initialize(self): - """Initialize the bluetooth plugin.""" - self.set_triggers(["bt"]) - - def cleanup(self): - """Cleanup the bluetooth plugin.""" - # Stop any ongoing scanning when plugin is cleaned up - if self.bluetooth_client and self.bluetooth_client.scanning: - self.bluetooth_client.scanning = False - - # Cancel any pending scan timeout - if self._scan_timeout_id: - try: - from gi.repository import GLib - - GLib.source_remove(self._scan_timeout_id) - self._scan_timeout_id = None - except Exception: - pass - - def _init_bluetooth_client(self): - """Initialize bluetooth client with proper error handling.""" - try: - # Set up timeout for bluetooth client creation - import signal - - def timeout_handler(_signum, _frame): - raise TimeoutError("Bluetooth client initialization timeout") - - old_handler = signal.signal(signal.SIGALRM, timeout_handler) - signal.alarm(3) # 3 second timeout - - try: - self.bluetooth_client = BluetoothClient() - self.bluetooth_client.connect("changed", self._on_bluetooth_changed) - self.bluetooth_client.connect("device-added", self._on_device_changed) - self.bluetooth_client.connect("device-removed", self._on_device_changed) - self._bluetooth_available = True - except TimeoutError: - self.bluetooth_client = None - except Exception as e: - print(f"Warning: Failed to initialize BluetoothClient: {e}") - self.bluetooth_client = None - finally: - signal.alarm(0) - signal.signal(signal.SIGALRM, old_handler) - - except Exception as e: - print(f"Warning: Bluetooth initialization failed: {e}") - self.bluetooth_client = None - - def _on_bluetooth_changed(self, *_args): - """Called when bluetooth state changes.""" - pass - - def _on_device_changed(self, *_args): - """Called when devices are added or removed.""" - pass - - def _get_bluetooth_status_icon(self) -> str: - """Get icon representing current bluetooth status.""" - if not self.bluetooth_client: - return icons.bluetooth_off - - if not self.bluetooth_client.enabled: - return icons.bluetooth_off - elif self.bluetooth_client.connected_devices: - return icons.bluetooth_connected - elif self.bluetooth_client.scanning: - return icons.loader - else: - return icons.bluetooth - - def _format_bluetooth_status(self) -> tuple[str, str]: - """Get formatted bluetooth status title and subtitle.""" - if not self.bluetooth_client: - return "Bluetooth Unavailable", "Bluetooth service not available" - - if not self.bluetooth_client.enabled: - return "Bluetooth Disabled", "Click to enable Bluetooth" - - connected_devices = self.bluetooth_client.connected_devices - if connected_devices: - if len(connected_devices) == 1: - device = connected_devices[0] - battery_info = "" - if device.battery_percentage > 0: - battery_info = f" ({device.battery_percentage:.0f}%)" - return ( - f"Connected to {device.alias}", - f"Device: {device.type}{battery_info}", - ) - else: - return ( - f"Connected to {len(connected_devices)} devices", - "Multiple devices connected", - ) - elif self.bluetooth_client.scanning: - return "Scanning for devices...", "Looking for nearby Bluetooth devices" - else: - return "Bluetooth Enabled", "No devices connected" - - def _get_device_icon(self, device: BluetoothDevice) -> str: - """Get appropriate icon for a bluetooth device based on its type.""" - device_type = device.type.lower() - - if ( - "audio" in device_type - or "headphone" in device_type - or "headset" in device_type - ): - return icons.headphones - elif "mouse" in device_type or "keyboard" in device_type: - return icons.keyboard - elif "phone" in device_type: - return icons.bluetooth - else: - return icons.bluetooth - - def _toggle_bluetooth(self): - """Toggle Bluetooth on/off and keep launcher open.""" - if self.bluetooth_client: - self.bluetooth_client.toggle_power() - # Force launcher refresh to show updated status - self._force_launcher_refresh() - return None # Keep launcher open - - def _scan_devices(self): - """Scan for available devices and keep launcher open (like network plugin).""" - if self.bluetooth_client: - # Start scanning - self.bluetooth_client.scan() - - # Cancel any existing auto-stop timeout - if self._scan_timeout_id: - try: - from gi.repository import GLib - - GLib.source_remove(self._scan_timeout_id) - self._scan_timeout_id = None - except Exception: - pass - - # Auto-stop scanning after 30 seconds to prevent continuous scanning - try: - from gi.repository import GLib - - def auto_stop_scan(): - if self.bluetooth_client and self.bluetooth_client.scanning: - print("Auto-stopping bluetooth scan after 30 seconds...") - self.bluetooth_client.scanning = False - self._scan_timeout_id = None - return False # Don't repeat - - # Schedule auto-stop after 30 seconds - self._scan_timeout_id = GLib.timeout_add_seconds(30, auto_stop_scan) - - except Exception as e: - print(f"Could not schedule auto-stop scan: {e}") - - return None # Keep launcher open - - def _stop_scanning(self): - """Stop bluetooth scanning.""" - if self.bluetooth_client and self.bluetooth_client.scanning: - self.bluetooth_client.scanning = False - - # Cancel any pending auto-stop timeout - if self._scan_timeout_id: - try: - from gi.repository import GLib - - GLib.source_remove(self._scan_timeout_id) - self._scan_timeout_id = None - except Exception: - pass - - return None # Keep launcher open - - def _check_and_stop_inactive_scanning(self): - """Check if plugin is inactive and stop scanning if needed.""" - if not self.bluetooth_client or not self.bluetooth_client.scanning: - return - - # Stop scanning if plugin hasn't been used for 10 seconds - import time - - current_time = time.time() - if current_time - self._last_query_time > 10: - self._stop_scanning() - - def _connect_device(self, device: BluetoothDevice): - """Connect to a bluetooth device.""" - if device and not device.connected: - device.connected = True # This will trigger the connection - # Force launcher refresh to show updated status - self._force_launcher_refresh() - return None # Keep launcher open - - def _disconnect_device(self, device: BluetoothDevice): - """Disconnect from a bluetooth device.""" - if device and device.connected: - device.connected = False # This will trigger the disconnection - # Force launcher refresh to show updated status - self._force_launcher_refresh() - return None # Keep launcher open - - def _create_connect_action(self, device: BluetoothDevice): - """Create a connect action function with proper closure.""" - - def connect_action(): - return self._connect_device(device) - - return connect_action - - def _create_disconnect_action(self, device: BluetoothDevice): - """Create a disconnect action function with proper closure.""" - - def disconnect_action(): - return self._disconnect_device(device) - - return disconnect_action - - def _force_launcher_refresh(self): - """Force the launcher to refresh and show updated bluetooth status.""" - try: - from gi.repository import GLib - - # Use a small delay to ensure the action completes first - def trigger_refresh(): - try: - # Try to access the launcher through the fabric Application - from fabric import Application - - app = Application.get_default() - - if app and hasattr(app, "launcher"): - launcher = app.launcher - if launcher and hasattr(launcher, "_perform_search"): - # Get current query and trigger a search - current_query = ( - launcher.query if hasattr(launcher, "query") else "" - ) - if not current_query: - # If no query, use the trigger to force refresh - current_query = "bt list" - - # Force a search which will refresh the results - launcher._perform_search(current_query) - return False - - # Fallback: try to find launcher instance through other means - import gc - - for obj in gc.get_objects(): - if ( - hasattr(obj, "__class__") - and obj.__class__.__name__ == "Launcher" - ): - if hasattr(obj, "_perform_search") and hasattr( - obj, "query" - ): - current_query = obj.query if obj.query else "bt list" - obj._perform_search(current_query) - return False - - except Exception as e: - print(f"Error forcing launcher refresh: {e}") - - return False # Don't repeat - - # Use a small delay to ensure the bluetooth action completes first - GLib.timeout_add(100, trigger_refresh) - - except Exception as e: - print(f"Could not trigger refresh: {e}") - - def query(self, query_string: str) -> List[Result]: - """Process bluetooth queries.""" - # Update last query time to track plugin activity - import time - - self._last_query_time = time.time() - - # Schedule periodic check for inactive scanning - try: - from gi.repository import GLib - - GLib.timeout_add_seconds( - 15, lambda: (self._check_and_stop_inactive_scanning(), False) - ) - except Exception: - pass - - results = [] - query = query_string.strip() - query_lower = query.lower() - - if not query_lower: - # Check if bluetooth client is available - if not self.bluetooth_client: - results.append( - Result( - title="Bluetooth Service Unavailable", - subtitle="Bluetooth service is not available", - icon_markup=icons.bluetooth_off, - action=lambda: None, - relevance=1.0, - plugin_name=self.display_name, - data={"type": "error"}, - ) - ) - return results - - # Show current bluetooth status - title, subtitle = self._format_bluetooth_status() - status_icon = self._get_bluetooth_status_icon() - - results.append( - Result( - title=title, - subtitle=subtitle, - icon_markup=status_icon, - action=lambda: None, - relevance=1.0, - plugin_name=self.display_name, - data={"type": "status"}, - ) - ) - - # Show available actions - if self.bluetooth_client.enabled: - results.append( - Result( - title="Disable Bluetooth", - subtitle="Turn off Bluetooth adapter", - icon_markup=icons.bluetooth_off, - action=lambda: self._toggle_bluetooth(), - relevance=0.9, - plugin_name=self.display_name, - data={"type": "toggle", "keep_launcher_open": True}, - ) - ) - - else: - results.append( - Result( - title="Enable Bluetooth", - subtitle="Turn on Bluetooth adapter", - icon_markup=icons.bluetooth, - action=lambda: self._toggle_bluetooth(), - relevance=0.9, - plugin_name=self.display_name, - data={"type": "toggle", "keep_launcher_open": True}, - ) - ) - - return results - - # Handle specific queries - if query_lower in ["list", "devices"]: - if not self.bluetooth_client or not self.bluetooth_client.enabled: - results.append( - Result( - title="Bluetooth Disabled", - subtitle="Enable Bluetooth to see devices", - icon_markup=icons.bluetooth_off, - action=lambda: self._toggle_bluetooth(), - relevance=1.0, - plugin_name=self.display_name, - data={"type": "enable", "keep_launcher_open": True}, - ) - ) - return results - - # Automatically scan when listing devices (like network plugin) - self._scan_devices() - - devices = self.bluetooth_client.devices - - # Show scanning message if no devices available yet (like network plugin) - if not devices: - results.append( - Result( - title="Scanning for Devices...", - subtitle="Please wait while scanning for Bluetooth devices", - icon_markup=icons.radar, - action=lambda: None, # Don't close launcher - relevance=1.0, - plugin_name=self.display_name, - data={"type": "scanning", "keep_launcher_open": True}, - ) - ) - else: - # Show all devices (limit to 10 like network plugin) - for device in devices[:10]: - device_icon = self._get_device_icon(device) - - # Add connection icon for connected devices - if device.connected: - title = f"🔗 {device.alias}" - else: - title = device.alias - - if device.connected: - battery_info = "" - if device.battery_percentage > 0: - battery_info = f" • {device.battery_percentage:.0f}%" - - subtitle = f"Connected • {device.type}{battery_info}" - action = self._create_disconnect_action(device) - relevance = 1.0 - elif device.connecting: - subtitle = f"Connecting... • {device.type}" - - def action(): - return None - - relevance = 0.9 - else: - subtitle = f"Disconnected • {device.type} • Click to connect" - action = self._create_connect_action(device) - relevance = 0.8 - - results.append( - Result( - title=title, - subtitle=subtitle, - icon_markup=device_icon, - action=action, - relevance=relevance, - plugin_name=self.display_name, - data={ - "type": "device", - "device_address": device.address, - "device_name": device.alias, - "connected": device.connected, - "keep_launcher_open": True, - }, - ) - ) - - return results - - # Handle power/toggle commands - if query_lower in ["toggle"]: - if not self.bluetooth_client: - return results - - if self.bluetooth_client.enabled: - results.append( - Result( - title="Disable Bluetooth", - subtitle="Turn off Bluetooth adapter", - icon_markup=icons.bluetooth_off, - action=lambda: self._toggle_bluetooth(), - relevance=1.0, - plugin_name=self.display_name, - data={"type": "toggle", "keep_launcher_open": True}, - ) - ) - else: - results.append( - Result( - title="Enable Bluetooth", - subtitle="Turn on Bluetooth adapter", - icon_markup=icons.bluetooth, - action=lambda: self._toggle_bluetooth(), - relevance=1.0, - plugin_name=self.display_name, - data={"type": "toggle", "keep_launcher_open": True}, - ) - ) - - return results - - # Search for devices by name (like network plugin search) - if self.bluetooth_client and self.bluetooth_client.enabled: - devices = self.bluetooth_client.devices - matching_devices = [ - device - for device in devices - if query_lower in device.alias.lower() - or query_lower in device.name.lower() - ] - - if matching_devices: - # Show matching devices (limit to 5 like network plugin) - for device in matching_devices[:5]: - device_icon = self._get_device_icon(device) - - # Add connection icon for connected devices - if device.connected: - title = f"🔗 {device.alias}" - else: - title = device.alias - - if device.connected: - battery_info = "" - if device.battery_percentage > 0: - battery_info = f" • {device.battery_percentage:.0f}%" - - subtitle = f"Connected • {device.type}{battery_info}" - action = self._create_disconnect_action(device) - relevance = 1.0 - elif device.connecting: - subtitle = f"Connecting... • {device.type}" - - def action(): - return None - - relevance = 0.9 - else: - subtitle = f"Disconnected • {device.type} • Click to connect" - action = self._create_connect_action(device) - relevance = 0.9 - - results.append( - Result( - title=title, - subtitle=subtitle, - icon_markup=device_icon, - action=action, - relevance=relevance, - plugin_name=self.display_name, - data={ - "type": "device", - "device_address": device.address, - "device_name": device.alias, - "connected": device.connected, - "keep_launcher_open": True, - }, - ) - ) - - return results - else: - # No matching devices found (like network plugin) - results.append( - Result( - title=f"No devices matching '{query}'", - subtitle="Try scanning for more devices", - icon_markup=icons.radar, - action=lambda: self._scan_devices(), - relevance=0.5, - plugin_name=self.display_name, - data={"type": "scan", "keep_launcher_open": True}, - ) - ) - return results - - # Fallback for when Bluetooth is disabled or not available (like network plugin) - if not self.bluetooth_client or not self.bluetooth_client.enabled: - results.append( - Result( - title="Bluetooth Not Available", - subtitle="Enable Bluetooth to search for devices", - icon_markup=icons.bluetooth_off, - action=lambda: self._toggle_bluetooth() - if self.bluetooth_client - else None, - relevance=0.5, - plugin_name=self.display_name, - data={"type": "info"}, - ) - ) - - return results diff --git a/modules/launcher/plugins/network.py b/modules/launcher/plugins/network.py deleted file mode 100644 index 643bdaf..0000000 --- a/modules/launcher/plugins/network.py +++ /dev/null @@ -1,862 +0,0 @@ -from typing import List - -from fabric.widgets.box import Box -from fabric.widgets.entry import Entry -from gi.repository import GLib - -import utils.icons as icons -from modules.launcher.plugin_base import PluginBase -from modules.launcher.result import Result -from services.network import NetworkClient - - -class NetworkPasswordEntry(Box): - """ - Custom password entry widget for network connections. - """ - - def __init__(self, access_point, network_plugin, **kwargs): - super().__init__( - name="network-password-entry", orientation="h", spacing=8, **kwargs - ) - self.access_point = access_point - self.network_plugin = network_plugin - - # Make widget focusable for keyboard events - self.set_can_focus(True) - self.connect("key-press-event", self.on_key_press) - - # Password entry field - self.password_entry = Entry( - name="network-password-field", - placeholder="Enter password", - h_expand=True, - h_align="fill", - visibility=False, - ) - - # Connect Enter key to connect action - self.password_entry.connect("activate", lambda *args: self.connect_to_network()) - - # Auto-focus the password entry when widget is created - GLib.timeout_add(100, self._focus_password_entry) - - # Add widgets to layout - self.add(self.password_entry) - - def _focus_password_entry(self): - """Focus the password entry field.""" - try: - self.password_entry.grab_focus() - except Exception as e: - print(f"Could not focus password entry: {e}") - return False # Don't repeat - - def on_key_press(self, widget, event): - """Handle keyboard events for password entry.""" - from gi.repository import Gdk - - keyval = event.keyval - - # Enter - connect to network - if keyval == Gdk.KEY_Return: - self.connect_to_network() - return True - - # Escape - cancel password entry - if keyval == Gdk.KEY_Escape: - self.cancel_password_entry() - return True - - return False - - def connect_to_network(self): - """Connect to network with entered password.""" - password = self.password_entry.get_text() - if password: - self.network_plugin._connect_to_network(self.access_point, password) - self.network_plugin.showing_password_for = None - # Force refresh to show normal network list - self.network_plugin._force_launcher_refresh() - else: - print("Please enter a password") - - def cancel_password_entry(self): - """Cancel password entry and return to network list.""" - self.network_plugin.showing_password_for = None - # Force refresh to show normal network list - self.network_plugin._force_launcher_refresh() - - -class NetworkPlugin(PluginBase): - """ - Network management plugin for the launcher. - """ - - def __init__(self): - super().__init__() - self.display_name = "Network" - self.description = "Manage WiFi networks and connections" - - # Track if we're ready - self._ready = False - - # Track which network is showing password entry - self.showing_password_for = None - - # Initialize network client safely - try: - self.network_client = NetworkClient() - - # Connect to network client signals - self.network_client.connect("ready", self._on_network_ready) - self.network_client.connect("changed", self._on_network_changed) - - if self.network_client.is_ready: - self._setup_network_signals() - except Exception as e: - print(f"Warning: Failed to initialize NetworkClient: {e}") - self.network_client = None - - def initialize(self): - """Initialize the network plugin.""" - self.set_triggers(["net"]) - self.description = "Manage WiFi networks and connections. Keyboard: Enter to connect, Esc to cancel password entry" - - def cleanup(self): - """Cleanup the network plugin.""" - pass - - def _on_network_ready(self, *args): - """Called when network client is ready.""" - self._ready = True - self._setup_network_signals() - - def _on_network_changed(self, *args): - """Called when network state changes.""" - pass - - def _setup_network_signals(self): - """Setup signals for WiFi and Ethernet devices.""" - if self.network_client and self.network_client.wifi_device: - self.network_client.wifi_device.connect("changed", self._on_wifi_changed) - self.network_client.wifi_device.connect("ap-added", self._on_wifi_changed) - self.network_client.wifi_device.connect("ap-removed", self._on_wifi_changed) - - def _on_wifi_changed(self, *args): - """Called when WiFi state changes.""" - pass - - def _get_wifi_icon(self, strength: int) -> str: - """Get WiFi icon based on signal strength.""" - if strength >= 75: - return icons.wifi_3 - elif strength >= 50: - return icons.wifi_2 - elif strength >= 25: - return icons.wifi_1 - else: - return icons.wifi_0 - - def _get_connection_status_icon(self) -> str: - """Get icon representing current connection status.""" - if not self._ready or not self.network_client: - return icons.loader - - # Check ethernet first - if ( - self.network_client.ethernet_device - and self.network_client.ethernet_device.state == "activated" - ): - return icons.world - - # Check WiFi - if not self.network_client.wifi_device: - return icons.cloud_off - elif not self.network_client.wifi_device.wireless_enabled: - return icons.wifi_off - else: - active_ap = self.network_client.wifi_device.active_access_point - if active_ap: - return self._get_wifi_icon(active_ap.strength) - else: - return icons.world_off - - def _format_connection_status(self) -> tuple[str, str]: - """Get formatted connection status title and subtitle.""" - if not self._ready or not self.network_client: - return "Network Loading...", "Initializing network client" - - # Check ethernet first - if ( - self.network_client.ethernet_device - and self.network_client.ethernet_device.state == "activated" - ): - speed = self.network_client.ethernet_device.speed - return "Connected via Ethernet", f"Speed: {speed}" - - # Check WiFi - if not self.network_client.wifi_device: - return "No WiFi Device", "WiFi hardware not found" - elif not self.network_client.wifi_device.wireless_enabled: - return "WiFi Disabled", "Click to enable WiFi" - else: - active_ap = self.network_client.wifi_device.active_access_point - if active_ap: - return ( - f"Connected to {active_ap.ssid}", - f"Signal: {active_ap.strength}%", - ) - else: - return "WiFi Disconnected", "No active connection" - - def _toggle_wifi(self): - """Toggle WiFi on/off and keep launcher open.""" - if self.network_client and self.network_client.wifi_device: - self.network_client.wifi_device.toggle_wifi() - return None # Keep launcher open to show new state - - def _disconnect_wifi(self): - """Disconnect from current WiFi network and keep launcher open.""" - if ( - self.network_client - and self.network_client.wifi_device - and self.network_client.wifi_device.active_access_point - ): - self.network_client.wifi_device.disconnect_wifi() - return None # Keep launcher open to show disconnected state - - def _scan_networks(self): - """Scan for available networks and keep launcher open.""" - if self.network_client and self.network_client.wifi_device: - self.network_client.wifi_device.scan() - # Return None to keep launcher open - return None - - def _connect_to_network(self, access_point, password: str = None): - """Connect to a WiFi network with improved error handling.""" - try: - if not self.network_client: - return None - - if not self.network_client.wifi_device: - return None - - # Validate access point - if not access_point or not hasattr(access_point, "ssid"): - return None - - # Validate password for secured networks - if access_point.requires_password and not password: - return None - - # Use our own safer connection method to avoid NetworkManager errors - self._safe_connect_to_wifi(access_point, password) - - except Exception as e: - print(f"Error connecting to network {access_point.ssid}: {e}") - return None # Keep launcher open to show connection progress - - def _safe_connect_to_wifi(self, access_point, password: str = None): - """Safer WiFi connection method to avoid NetworkManager assertion errors.""" - try: - from gi.repository import NM, GLib - - ssid = access_point.ssid - client = self.network_client._client - device = self.network_client.wifi_device._device - - if not client or not device: - print("NetworkManager client or device not available") - return False - - # Check for existing connections first - existing_connection = None - for connection in client.get_connections(): - wifi_setting = connection.get_setting_wireless() - if wifi_setting: - conn_ssid = NM.utils_ssid_to_utf8( - wifi_setting.get_ssid().get_data() - ) - if conn_ssid == ssid: - existing_connection = connection - break - - if existing_connection: - # Use existing connection - def activate_existing(): - try: - client.activate_connection_async( - existing_connection, - device, - None, - None, - self._on_connection_result, - ) - except Exception as e: - print(f"Error activating existing connection: {e}") - return False - - GLib.timeout_add(200, activate_existing) - else: - # Create new connection with delay to prevent assertion errors - def create_new_connection(): - try: - self._create_and_activate_connection( - ssid, password, client, device - ) - except Exception as e: - print(f"Error creating new connection: {e}") - return False - - GLib.timeout_add(200, create_new_connection) - - except Exception as e: - print(f"Error in safe WiFi connection: {e}") - return False - - def _create_and_activate_connection(self, ssid, password, client, device): - """Create and activate a new WiFi connection.""" - try: - from gi.repository import NM, GLib - - # Create connection - connection = NM.SimpleConnection.new() - - # 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, device.get_iface()) - 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") - connection.add_setting(s_wifi) - - # Security settings if password provided - if password: - 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) - - # IP settings - s_ipv4 = NM.SettingIP4Config.new() - s_ipv4.set_property("method", "auto") - connection.add_setting(s_ipv4) - - s_ipv6 = NM.SettingIP6Config.new() - s_ipv6.set_property("method", "auto") - connection.add_setting(s_ipv6) - - # Add connection and activate - def on_connection_added(client_obj, result): - try: - new_connection = client_obj.add_connection_finish(result) - if new_connection: - # Activate with delay - def activate_new(): - try: - client_obj.activate_connection_async( - new_connection, - device, - None, - None, - self._on_connection_result, - ) - except Exception as e: - print(f"Error activating new connection: {e}") - return False - - GLib.timeout_add(100, activate_new) - else: - print(f"✗ Failed to create connection profile for {ssid}") - except Exception as e: - print(f"Error in connection callback: {e}") - - client.add_connection_async(connection, True, None, on_connection_added) - - except Exception as e: - print(f"Error creating connection: {e}") - - def _on_connection_result(self, client, result): - """Handle connection activation result.""" - try: - active_connection = client.activate_connection_finish(result) - if active_connection: - print("✓ WiFi connection successful") - else: - print("✗ WiFi connection failed") - except Exception as e: - print(f"Connection result error: {e}") - - def _show_password_entry_for_network(self, access_point): - """Show password entry for a network that requires password.""" - # Set which network is showing password entry - self.showing_password_for = access_point.ssid - - # Force launcher refresh by triggering a search - self._force_launcher_refresh() - - return None # Keep launcher open - - def _force_launcher_refresh(self): - """Force the launcher to refresh and show the password entry.""" - try: - from gi.repository import GLib - - # Use a small delay to ensure the action completes first - def trigger_refresh(): - try: - # Try to access the launcher through the fabric Application - from fabric import Application - - app = Application.get_default() - - if app and hasattr(app, "launcher"): - launcher = app.launcher - if launcher and hasattr(launcher, "_perform_search"): - # Get current query and trigger a search - current_query = ( - launcher.query if hasattr(launcher, "query") else "" - ) - if not current_query: - # If no query, use the trigger to force refresh - current_query = "net list" - - # Force a search which will refresh the results - launcher._perform_search(current_query) - return False - - # Fallback: try to find launcher instance through other means - import gc - - for obj in gc.get_objects(): - if ( - hasattr(obj, "__class__") - and obj.__class__.__name__ == "Launcher" - ): - if hasattr(obj, "_perform_search") and hasattr( - obj, "query" - ): - current_query = obj.query if obj.query else "net list" - obj._perform_search(current_query) - return False - - except Exception as e: - print(f"Error forcing launcher refresh: {e}") - - return False # Don't repeat - - # Use a small delay to ensure the action completes first - GLib.timeout_add(50, trigger_refresh) - - except Exception as e: - print(f"Could not trigger refresh: {e}") - - def _cancel_password_entry(self): - """Cancel password entry and return to network list.""" - self.showing_password_for = None - return None # Keep launcher open - - def query(self, query_string: str) -> List[Result]: - """Process network queries.""" - results = [] - query = query_string.strip() - query_lower = query.lower() - - # No more text-based password handling - using custom widgets instead - - if not query_lower: - # Check if network client is available - if not self.network_client: - results.append( - Result( - title="Network Service Unavailable", - subtitle="NetworkManager service is not available", - icon_markup=icons.cloud_off, - action=lambda: None, - relevance=1.0, - plugin_name=self.display_name, - data={"type": "error"}, - ) - ) - return results - - # Show current connection status - title, subtitle = self._format_connection_status() - status_icon = self._get_connection_status_icon() - - results.append( - Result( - title=title, - subtitle=subtitle, - icon_markup=status_icon, - action=lambda: None, - relevance=1.0, - plugin_name=self.display_name, - data={"type": "status"}, - ) - ) - - # Show available actions - if not self._ready: - return results - - # WiFi toggle option - if self.network_client.wifi_device: - if self.network_client.wifi_device.wireless_enabled: - results.append( - Result( - title="Disable WiFi", - subtitle="Turn off wireless networking", - icon_markup=icons.wifi_off, - action=lambda: self._toggle_wifi(), - relevance=0.9, - plugin_name=self.display_name, - data={"type": "toggle_wifi", "keep_launcher_open": True}, - ) - ) - - # Disconnect option if connected - if self.network_client.wifi_device.active_access_point: - results.append( - Result( - title="Disconnect WiFi", - subtitle=f"Disconnect from { - self.network_client.wifi_device.active_access_point.ssid - }", - icon_markup=icons.world_off, - action=lambda: self._disconnect_wifi(), - relevance=0.8, - plugin_name=self.display_name, - data={"type": "disconnect", "keep_launcher_open": True}, - ) - ) - - else: - results.append( - Result( - title="Enable WiFi", - subtitle="Turn on wireless networking", - icon_markup=icons.wifi_3, - action=lambda: self._toggle_wifi(), - relevance=0.9, - plugin_name=self.display_name, - data={"type": "toggle_wifi", "keep_launcher_open": True}, - ) - ) - - return results - - # Handle specific commands - if query_lower in ["scan", "refresh"]: - # Trigger scan immediately - self._scan_networks() - - results.append( - Result( - title="Scanning for Networks...", - subtitle="Refreshing available WiFi networks", - icon_markup=icons.radar, - action=lambda: None, # Keep launcher open - relevance=1.0, - plugin_name=self.display_name, - data={"type": "scanning", "keep_launcher_open": True}, - ) - ) - - # Show current networks if any - if ( - self._ready - and self.network_client.wifi_device - and self.network_client.wifi_device.wireless_enabled - ): - access_points = self.network_client.wifi_device.access_points - if access_points: - results.append( - Result( - title=f"Found {len(access_points)} Networks", - subtitle="Scan complete - networks updated", - icon_markup=icons.wifi_3, - action=lambda: None, - relevance=0.9, - plugin_name=self.display_name, - data={"type": "scan_complete", "keep_launcher_open": True}, - ) - ) - - return results - - if query_lower in ["disconnect", "off"]: - if ( - self.network_client.wifi_device - and self.network_client.wifi_device.active_access_point - ): - ap = self.network_client.wifi_device.active_access_point - results.append( - Result( - title=f"Disconnect from {ap.ssid}", - subtitle=f"Currently connected with {ap.strength}% signal", - icon_markup=icons.world_off, - action=lambda: self._disconnect_wifi(), - relevance=1.0, - plugin_name=self.display_name, - data={"type": "disconnect", "keep_launcher_open": True}, - ) - ) - else: - results.append( - Result( - title="Not Connected", - subtitle="No active WiFi connection to disconnect", - icon_markup=icons.world_off, - action=lambda: None, - relevance=0.5, - plugin_name=self.display_name, - data={"type": "info"}, - ) - ) - return results - - if query_lower in ["toggle"]: - if self.network_client.wifi_device: - if self.network_client.wifi_device.wireless_enabled: - action_text = "Disable WiFi" - icon = icons.wifi_off - else: - action_text = "Enable WiFi" - icon = icons.wifi_3 - - results.append( - Result( - title=action_text, - subtitle="Toggle wireless networking", - icon_markup=icon, - action=lambda: self._toggle_wifi(), - relevance=1.0, - plugin_name=self.display_name, - data={"type": "toggle_wifi", "keep_launcher_open": True}, - ) - ) - return results - - # Search for specific networks or show available networks - if ( - self._ready - and self.network_client.wifi_device - and self.network_client.wifi_device.wireless_enabled - ): - access_points = self.network_client.wifi_device.access_points - - if query_lower in ["list"]: - # Automatically scan when listing networks - self._scan_networks() - - # Show all available networks - if not access_points: - results.append( - Result( - title="Scanning for Networks...", - subtitle="Please wait while scanning for WiFi networks", - icon_markup=icons.radar, - action=lambda: None, # Don't close launcher - relevance=1.0, - plugin_name=self.display_name, - data={"type": "scanning", "keep_launcher_open": True}, - ) - ) - else: - for ap in access_points[:10]: # Limit to 10 networks - # Add lock icon for secured networks - if ap.requires_password: - title = f"🔒 {ap.ssid}" - else: - title = ap.ssid - - # Check if this network should show password entry instead - if ( - ap.requires_password - and self.showing_password_for == ap.ssid - ): - # REPLACE network with password entry Result using our own Entry widget - password_widget = NetworkPasswordEntry( - access_point=ap, network_plugin=self - ) - - results.append( - Result( - title=f"🔑 Enter password for {ap.ssid}", - subtitle="Type password and click connect • Click cancel to go back", - icon_markup=icons.lock, - action=lambda: None, # Widget handles the action - relevance=1.0, # High relevance to appear at top - plugin_name=self.display_name, - custom_widget=password_widget, # Use our custom Entry widget - data={ - "type": "password_entry", - "ssid": ap.ssid, - "keep_launcher_open": True, - }, - ) - ) - else: - # Show normal network Result - if ap.is_active: - subtitle = f"Connected • {ap.strength}% signal" - - def action(): - return self._disconnect_wifi() - else: - if ap.requires_password: - subtitle = f"{ - ap.strength - }% signal • Click to enter password" - - def action(current_ap=ap): - return self._show_password_entry_for_network( - current_ap - ) - else: - subtitle = ( - f"{ap.strength}% signal • Click to connect" - ) - - def action(current_ap=ap): - return self._connect_to_network(current_ap) - - results.append( - Result( - title=title, - subtitle=subtitle, - icon_markup=self._get_wifi_icon(ap.strength), - action=action, - relevance=0.9 if ap.is_active else 0.8, - plugin_name=self.display_name, - data={ - "type": "network", - "ssid": ap.ssid, - "requires_password": ap.requires_password, - "keep_launcher_open": True, - }, - ) - ) - - return results - - else: - # Search for networks matching the query - matching_networks = [ - ap for ap in access_points if query.lower() in ap.ssid.lower() - ] - - if matching_networks: - for ap in matching_networks[:5]: # Limit to 5 matches - if ap.requires_password: - title = f"🔒 {ap.ssid}" - else: - title = ap.ssid - - # Check if this network should show password entry instead - if ( - ap.requires_password - and self.showing_password_for == ap.ssid - ): - # REPLACE network with password entry Result using our own Entry widget - password_widget = NetworkPasswordEntry( - access_point=ap, network_plugin=self - ) - - results.append( - Result( - title=f"🔑 Enter password for {ap.ssid}", - subtitle="Type password and click connect • Click cancel to go back", - icon_markup=icons.lock, - action=lambda: None, # Widget handles the action - relevance=1.0, # High relevance to appear at top - plugin_name=self.display_name, - custom_widget=password_widget, # Use our custom Entry widget - data={ - "type": "password_entry", - "ssid": ap.ssid, - "keep_launcher_open": True, - }, - ) - ) - else: - # Show normal network Result - if ap.is_active: - subtitle = f"Connected • {ap.strength}% signal" - - def action(): - return self._disconnect_wifi() - else: - if ap.requires_password: - subtitle = f"{ - ap.strength - }% signal • Click to enter password" - - def action(current_ap=ap): - return self._show_password_entry_for_network( - current_ap - ) - else: - subtitle = ( - f"{ap.strength}% signal • Click to connect" - ) - - def action(current_ap=ap): - return self._connect_to_network(current_ap) - - results.append( - Result( - title=title, - subtitle=subtitle, - icon_markup=self._get_wifi_icon(ap.strength), - action=action, - relevance=1.0 if ap.is_active else 0.9, - plugin_name=self.display_name, - data={ - "type": "network", - "ssid": ap.ssid, - "requires_password": ap.requires_password, - "keep_launcher_open": True, - }, - ) - ) - - return results - else: - # No matching networks found - results.append( - Result( - title=f"No networks matching '{query}'", - subtitle="Try scanning for more networks", - icon_markup=icons.radar, - action=lambda: self._scan_networks(), - relevance=0.5, - plugin_name=self.display_name, - data={"type": "scan", "keep_launcher_open": True}, - ) - ) - return results - - # Fallback for when WiFi is disabled or not available - results.append( - Result( - title="WiFi Not Available", - subtitle="Enable WiFi to search for networks", - icon_markup=icons.wifi_off, - action=lambda: self._toggle_wifi() - if self.network_client.wifi_device - else None, - relevance=0.5, - plugin_name=self.display_name, - data={"type": "info"}, - ) - ) - - return results