From 9533d15ce0949673b7aa6b931d376042bc917dfd Mon Sep 17 00:00:00 2001 From: S4NKALP Date: Tue, 24 Jun 2025 02:53:47 +0545 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20feat(plugins):=20add=20Process=20pl?= =?UTF-8?q?ugin=20for=20managing=20running=20processes=20with=20live=20upd?= =?UTF-8?q?ates?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- modules/launcher/main.py | 89 +------ modules/launcher/plugin_manager.py | 1 + modules/launcher/plugins/process.py | 370 ++++++++++++++++++++++++++++ utils/icons.py | 1 + 4 files changed, 381 insertions(+), 80 deletions(-) create mode 100644 modules/launcher/plugins/process.py diff --git a/modules/launcher/main.py b/modules/launcher/main.py index b032a8f..64194d2 100644 --- a/modules/launcher/main.py +++ b/modules/launcher/main.py @@ -74,9 +74,8 @@ class Launcher(Window): self.visible = False # Launcher visibility state self.opened_with_trigger = False # Whether launcher was opened with a trigger - # Focus management for header buttons - self.focus_mode = "search" # "search", "results", "header" - self.header_button_index = 0 # 0 = config, 1 = close + # Focus management + self.focus_mode = "search" # "search", "results" # Setup UI main_box = Box( @@ -190,10 +189,6 @@ class Launcher(Window): # Reset focus mode to search self.focus_mode = "search" - self.header_button_index = 0 - - # Remove focus styling from header buttons - self._clear_header_focus() # Focus search entry without selecting text if trigger_keyword: @@ -280,7 +275,6 @@ class Launcher(Window): # Reset focus to search when user types if self.focus_mode != "search": self.focus_mode = "search" - self._clear_header_focus() query = entry.get_text().strip() self.query = query @@ -884,13 +878,8 @@ class Launcher(Window): self._update_selection() return True - # Enter - activate selected result or header button + # Enter - activate selected result if keyval == Gdk.KEY_Return: - # Check if we're in header mode - if self.focus_mode == "header": - # Activate the selected header button - self.header_buttons[self.header_button_index].emit("clicked") - return True # Check for Shift+Enter for alternative actions if event.state & Gdk.ModifierType.SHIFT_MASK: @@ -925,24 +914,18 @@ class Launcher(Window): self._activate_selected() return True - # Tab - cycle through focus areas, results, and header buttons + # Tab - cycle through focus areas and results if keyval == Gdk.KEY_Tab: if event.state & Gdk.ModifierType.SHIFT_MASK: # Shift+Tab - reverse direction - if self.focus_mode == "header": - # Navigate between header buttons in reverse - self._navigate_header_buttons_backward() - elif self.focus_mode == "results": + if self.focus_mode == "results": # Navigate through results in reverse self._navigate_results_backward() else: self._cycle_focus_backward() else: # Tab - forward direction - if self.focus_mode == "header": - # Navigate between header buttons forward - self._navigate_header_buttons_forward() - elif self.focus_mode == "results": + if self.focus_mode == "results": # Navigate through results forward self._navigate_results_forward() else: @@ -1173,45 +1156,19 @@ class Launcher(Window): return False # Don't repeat the idle callback def _cycle_focus_forward(self): - """Cycle focus forward: search -> results -> header (first button)""" + """Cycle focus forward: search -> results""" if self.focus_mode == "search": if self.results: self.focus_mode = "results" self._update_selection() - else: - self.focus_mode = "header" - self.header_button_index = 0 - self._update_header_focus() - elif self.focus_mode == "results": - self.focus_mode = "header" - self.header_button_index = 0 - self._update_header_focus() def _cycle_focus_backward(self): - """Cycle focus backward: search -> header (last button) -> results -> search""" - if self.focus_mode == "search": - self.focus_mode = "header" - self.header_button_index = len(self.header_buttons) - 1 - self._update_header_focus() - elif self.focus_mode == "results": + """Cycle focus backward: results -> search""" + if self.focus_mode == "results": self.focus_mode = "search" - self._clear_header_focus() self._focus_search_entry_without_selection() - def _update_header_focus(self): - """Update focus to the selected header button.""" - # Remove focus from all buttons first - self._clear_header_focus() - # Add focus style to selected button - selected_button = self.header_buttons[self.header_button_index] - selected_button.add_style_class("focused") - selected_button.grab_focus() - - def _clear_header_focus(self): - """Remove focus styling from all header buttons.""" - for button in self.header_buttons: - button.remove_style_class("focused") def _focus_search_entry_without_selection(self): """Focus search entry and position cursor at end without selecting text.""" @@ -1259,35 +1216,7 @@ class Launcher(Window): # Also try with a small delay as backup GLib.timeout_add(CURSOR_POSITION_DELAY_MS, clear_selection) - def _navigate_header_buttons_forward(self): - """Navigate to next header button or exit header mode.""" - if self.header_button_index < len(self.header_buttons) - 1: - # Move to next button - self.header_button_index += 1 - self._update_header_focus() - else: - # Exit header mode and go to search - self.focus_mode = "search" - self._clear_header_focus() - self._focus_search_entry_without_selection() - def _navigate_header_buttons_backward(self): - """Navigate to previous header button or exit header mode.""" - if self.header_button_index > 0: - # Move to previous button - self.header_button_index -= 1 - self._update_header_focus() - else: - # Exit header mode and go to results or search - if self.results: - self.focus_mode = "results" - self.selected_index = len(self.results) - 1 # Go to last result - self._clear_header_focus() - self._update_selection() - else: - self.focus_mode = "search" - self._clear_header_focus() - self._focus_search_entry_without_selection() def _navigate_results_forward(self): """Navigate to next result or wrap around to first result.""" diff --git a/modules/launcher/plugin_manager.py b/modules/launcher/plugin_manager.py index 3686a7a..56c89ff 100644 --- a/modules/launcher/plugin_manager.py +++ b/modules/launcher/plugin_manager.py @@ -96,6 +96,7 @@ class PluginManager: "notifications", "password", "bookmarks", + "process", ] for plugin_name in default_plugins: diff --git a/modules/launcher/plugins/process.py b/modules/launcher/plugins/process.py new file mode 100644 index 0000000..a733637 --- /dev/null +++ b/modules/launcher/plugins/process.py @@ -0,0 +1,370 @@ +import time +import threading +from typing import List + +import psutil +import utils.icons as icons +from modules.launcher.plugin_base import PluginBase +from modules.launcher.result import Result + + +class ProcessPlugin(PluginBase): + """ + Plugin for viewing and managing running processes with live CPU/RAM updates. + """ + + def __init__(self): + super().__init__() + self.display_name = "Process Manager" + self.description = "View and manage running processes" + + + + # Threading for auto-refresh + self.refresh_thread = None + self.stop_refresh = threading.Event() + self.last_update = 0 + + # CPU monitoring cache to get meaningful readings + self._cpu_cache = {} + self._last_cpu_update = 0 + + def initialize(self): + """Initialize the process plugin.""" + self.set_triggers(["ps"]) + # Pre-populate CPU cache for immediate meaningful readings + self._initialize_cpu_monitoring() + self._start_refresh_thread() + + def _initialize_cpu_monitoring(self): + """Initialize CPU monitoring for all processes to get meaningful readings immediately.""" + try: + for proc in psutil.process_iter(['pid']): + try: + pid = proc.info['pid'] + # Initialize CPU monitoring - this establishes baseline for future readings + proc.cpu_percent() + self._cpu_cache[pid] = proc + except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess): + continue + self._last_cpu_update = time.time() + except Exception as e: + print(f"ProcessPlugin: Error initializing CPU monitoring: {e}") + + def cleanup(self): + """Cleanup the process plugin.""" + self.stop_refresh.set() + if self.refresh_thread: + self.refresh_thread.join(timeout=1) + # Clear CPU cache + self._cpu_cache.clear() + + def _get_live_process_data(self): + """Get fresh live process data with CPU/RAM information.""" + try: + processes = [] + current_time = time.time() + + # Initialize CPU monitoring if this is the first call or enough time has passed + if current_time - self._last_cpu_update > 1.0: # Update every second + # Pre-populate CPU cache for all processes + for proc in psutil.process_iter(['pid']): + try: + pid = proc.info['pid'] + # Initialize CPU monitoring - this call establishes baseline + proc.cpu_percent() + self._cpu_cache[pid] = proc + except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess): + continue + self._last_cpu_update = current_time + + # Get process data with meaningful CPU readings + for proc in psutil.process_iter(['pid', 'name', 'status']): + try: + info = proc.info + if info['status'] == psutil.STATUS_ZOMBIE: + continue + + pid = info['pid'] + + # Get CPU percentage - use cached process if available for better readings + if pid in self._cpu_cache: + try: + cpu_percent = self._cpu_cache[pid].cpu_percent() + except (psutil.NoSuchProcess, psutil.AccessDenied): + cpu_percent = proc.cpu_percent() + self._cpu_cache[pid] = proc + else: + cpu_percent = proc.cpu_percent() + self._cpu_cache[pid] = proc + + # Normalize CPU to 0-100% range (divide by number of cores) + # psutil returns total CPU across all cores, so divide by core count + cpu_cores = psutil.cpu_count() + if cpu_cores and cpu_percent > 0: + cpu_percent = min(100.0, cpu_percent / cpu_cores) + + memory_info = proc.memory_info() + memory_percent = proc.memory_percent() + memory_mb = memory_info.rss / (1024 * 1024) if memory_info else 0 + + process_data = { + 'pid': pid, + 'name': info['name'] or 'Unknown', + 'cpu_percent': cpu_percent, + 'memory_percent': memory_percent, + 'memory_mb': memory_mb, + 'status': info['status'] + } + processes.append(process_data) + + except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess): + continue + + # Clean up old entries from cache + current_pids = {proc['pid'] for proc in processes} + self._cpu_cache = {pid: proc for pid, proc in self._cpu_cache.items() if pid in current_pids} + + # Sort by combined CPU and memory usage for better display + processes.sort(key=lambda x: (x['cpu_percent'] + x['memory_percent']), reverse=True) + return processes + + except Exception as e: + print(f"ProcessPlugin: Error getting process data: {e}") + return [] + + def query(self, query_string: str) -> List[Result]: + """Search for processes matching the query with live updates - ALWAYS FRESH DATA.""" + # ALWAYS get completely fresh data - NO CACHING EVER + processes = self._get_live_process_data() + + query = query_string.strip().lower() + results = [] + + # Filter processes based on query + filtered_processes = processes + if query: + filtered_processes = [ + proc for proc in processes + if query in proc['name'].lower() or query in str(proc['pid']) + ] + + # Limit to top 20 processes to avoid overwhelming the UI + for proc in filtered_processes[:20]: + result = self._create_process_result(proc) + if result: + results.append(result) + + return results + + def _create_process_result(self, proc_data: dict) -> Result: + """Create a Result object for a process.""" + try: + pid = proc_data['pid'] + name = proc_data['name'] + cpu_percent = proc_data['cpu_percent'] + memory_percent = proc_data['memory_percent'] + memory_mb = proc_data['memory_mb'] + + # Format the title and subtitle with live indicator + title = f"{name} (PID: {pid})" + + # Add visual indicators for high usage + cpu_indicator = "🔥" if cpu_percent > 80 else "âš¡" if cpu_percent > 50 else "" + mem_indicator = "💾" if memory_percent > 80 else "" + + subtitle = f"CPU: {cpu_percent:.1f}%{cpu_indicator} | Memory: {memory_mb:.1f}MB ({memory_percent:.1f}%){mem_indicator}" + + # Choose icon based on CPU usage + if cpu_percent > 50: + icon_markup = icons.cpu + elif memory_percent > 50: + icon_markup = icons.memory + else: + icon_markup = icons.terminal + + return Result( + title=title, + subtitle=subtitle, + icon_markup=icon_markup, + action=lambda: None, # No action on Enter - just show process info + relevance=min(1.0, (cpu_percent + memory_percent) / 100.0), + plugin_name=self.display_name, + data={ + "type": "process", + "pid": pid, + "name": name, + "cpu_percent": cpu_percent, + "memory_percent": memory_percent, + "memory_mb": memory_mb, + "keep_launcher_open": True, # Keep launcher open to see live updates + "alt_action": lambda p=pid: self._kill_process(p), + }, + ) + except Exception as e: + print(f"ProcessPlugin: Error creating process result: {e}") + return None + + + + def _kill_process(self, pid: int): + """Kill a process by PID.""" + try: + proc = psutil.Process(pid) + proc_name = proc.name() + + # Try graceful termination first + proc.terminate() + + # Wait a bit for graceful termination + try: + proc.wait(timeout=3) + print(f"✓ Successfully terminated process '{proc_name}' (PID: {pid})") + except psutil.TimeoutExpired: + # Force kill if graceful termination failed + proc.kill() + print(f"✓ Force killed process '{proc_name}' (PID: {pid})") + + + + except psutil.NoSuchProcess: + print(f"✗ Process {pid} not found - may have already been terminated") + except psutil.AccessDenied: + print(f"✗ Access denied - cannot kill process {pid} (insufficient permissions)") + except Exception as e: + print(f"✗ Error killing process {pid}: {e}") + + def _start_refresh_thread(self): + """Start background thread for auto-refreshing process data (optimized).""" + + def refresh_loop(): + while not self.stop_refresh.wait(2.0): # Check every 2 seconds to reduce load + current_time = time.time() + if current_time - self.last_update >= 2.0: # Update every 2 seconds + self.last_update = current_time + # Update existing process labels without full refresh + try: + self._selective_force_refresh() + except Exception: + pass + + self.refresh_thread = threading.Thread(target=refresh_loop, daemon=True) + self.refresh_thread.start() + + def _selective_force_refresh(self): + """Update process data in existing result items (like OTP plugin).""" + try: + import gc + from gi.repository import GLib + + def do_update(): + try: + for obj in gc.get_objects(): + if ( + hasattr(obj, "__class__") + and obj.__class__.__name__ == "Launcher" + and hasattr(obj, "results_box") + and hasattr(obj, "visible") + and obj.visible + and hasattr(obj, "results") + and obj.results + ): + has_process_results = any( + result.data and result.data.get("type") == "process" + for result in obj.results + if hasattr(result, "data") and result.data + ) + + if has_process_results: + self._update_existing_process_labels(obj.results_box) + return False + except Exception: + pass + return False + + GLib.idle_add(do_update) + except Exception: + pass + + def _update_existing_process_labels(self, results_box): + """Update subtitle labels in existing ResultItem widgets with FRESH data (optimized).""" + try: + # Get fresh process data in background thread - optimized + def get_data_async(): + return self._get_live_process_data() + + # Run data collection in background to avoid blocking UI + import concurrent.futures + with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor: + future = executor.submit(get_data_async) + try: + current_processes = future.result(timeout=1.0) # 1 second timeout + process_dict = {proc['pid']: proc for proc in current_processes} + + for child in results_box.get_children(): + if ( + hasattr(child, "__class__") + and child.__class__.__name__ == "ResultItem" + and hasattr(child, "result") + and hasattr(child.result, "data") + and child.result.data + and child.result.data.get("type") == "process" + ): + self._update_process_result_item(child, process_dict) + except concurrent.futures.TimeoutError: + # Skip update if it takes too long + pass + except Exception as e: + print(f"Error updating process labels: {e}") + + def _update_process_result_item(self, result_item, process_dict): + """Update both the title and subtitle of a specific process ResultItem.""" + try: + pid = result_item.result.data.get("pid") + if not pid or pid not in process_dict: + return + + proc_data = process_dict[pid] + cpu_percent = proc_data['cpu_percent'] + memory_percent = proc_data['memory_percent'] + memory_mb = proc_data['memory_mb'] + + # CPU is already normalized in _get_live_process_data, so no need to normalize again + + # Update the subtitle with live data + cpu_indicator = "🔥" if cpu_percent > 80 else "âš¡" if cpu_percent > 50 else "" + mem_indicator = "💾" if memory_percent > 80 else "" + new_subtitle = f"CPU: {cpu_percent:.1f}%{cpu_indicator} | Memory: {memory_mb:.1f}MB ({memory_percent:.1f}%){mem_indicator}" + + # Find and update the subtitle label widget + self._find_and_update_subtitle_label(result_item, new_subtitle) + + # Update the result data for consistency + result_item.result.data.update({ + "cpu_percent": cpu_percent, + "memory_percent": memory_percent, + "memory_mb": memory_mb, + }) + + except Exception as e: + print(f"Error updating process result item: {e}") + + def _find_and_update_subtitle_label(self, result_item, new_subtitle): + """Find the subtitle label widget and update its text.""" + + def find_subtitle_label(widget): + if ( + hasattr(widget, "get_name") + and widget.get_name() == "result-item-subtitle" + ): + return widget + if hasattr(widget, "get_children"): + for child in widget.get_children(): + found = find_subtitle_label(child) + if found: + return found + return None + + subtitle_label = find_subtitle_label(result_item) + if subtitle_label and hasattr(subtitle_label, "set_text"): + subtitle_label.set_text(new_subtitle) diff --git a/utils/icons.py b/utils/icons.py index 7139536..c08e4f2 100644 --- a/utils/icons.py +++ b/utils/icons.py @@ -185,6 +185,7 @@ disconnect: str = "ﴖ" auth: str = "" bookmark: str = "" star: str = "" +refresh: str = "" # otp