diff --git a/assets/default.png b/assets/default.png new file mode 100644 index 0000000..774a4e0 Binary files /dev/null and b/assets/default.png differ diff --git a/assets/modus.png b/assets/modus.png new file mode 100644 index 0000000..eb680cc Binary files /dev/null and b/assets/modus.png differ diff --git a/assets/wallpapers_example/example-1.png b/assets/wallpapers_example/example-1.png new file mode 100644 index 0000000..137ceee Binary files /dev/null and b/assets/wallpapers_example/example-1.png differ diff --git a/assets/wallpapers_example/example-2.png b/assets/wallpapers_example/example-2.png new file mode 100644 index 0000000..6d22cf9 Binary files /dev/null and b/assets/wallpapers_example/example-2.png differ diff --git a/assets/wallpapers_example/example-3.jpg b/assets/wallpapers_example/example-3.jpg new file mode 100755 index 0000000..5c2fd0d Binary files /dev/null and b/assets/wallpapers_example/example-3.jpg differ diff --git a/assets/wallpapers_example/example-4.jpeg b/assets/wallpapers_example/example-4.jpeg new file mode 100644 index 0000000..7ca7ae6 Binary files /dev/null and b/assets/wallpapers_example/example-4.jpeg differ diff --git a/assets/wallpapers_example/example-5.jpg b/assets/wallpapers_example/example-5.jpg new file mode 100644 index 0000000..eced4ab Binary files /dev/null and b/assets/wallpapers_example/example-5.jpg differ diff --git a/config/hypr/modus.conf b/config/hypr/modus.conf index 89ea896..a63e071 100644 --- a/config/hypr/modus.conf +++ b/config/hypr/modus.conf @@ -5,38 +5,32 @@ exec-once = wl-paste --type text --watch cliphist store exec-once = wl-paste --type image --watch cliphist store $fabricSend = fabric-cli exec modus -$axMessage = notify-send "Axenide" "FIRE IN THE HOLE‼️🗣️🔥🕳️" -i "/home/sankalp/.config/Modus/assets/ax.png" -A "🗣️" -A "🔥" -A "🕳️" -a "Source Code" +$axMessage = notify-send "Modus" "FIRE IN THE HOLE‼️🗣️🔥🕳️" -i "/home/sankalp/Modus/assets/modus.png" -A "🗣️" -A "🔥" -A "🕳️" -a "Source Code" # Reload Modus -bind = SUPER ALT, B, exec, killall modus; uwsm-app $(python /home/sankalp/.config/Modus/main.py) +bind = ALT SHIFT, R, exec, killall modus; uwsm-app $(python /home/sankalp/Modus/main.py) # Message bind = SUPER, A, exec, $axMessage # Application Switcher bind = ALT, TAB, exec, $fabricSend 'switcher.show_switcher()' -# Dashboard -bind = SUPER, D, exec, $fabricSend 'notch.open_notch("dashboard")' -# Bluetooth -bind = SUPER SHIFT, B, exec, $fabricSend 'notch.open_notch("bluetooth")' -bind = SUPER, Q, exec, $fabricSend 'notch.open_notch("pins")' # Pins -bind = SUPER, N, exec, $fabricSend 'notch.open_notch("kanban")' # Kanban +# Kanban +bind = SUPER, T, exec, $fabricSend "launcher.show_launcher('kanban ')" # Kanban # App Launcher -bind = SUPER, R, exec, $fabricSend 'notch.open_notch("launcher")' -bind = SUPER, T, exec, $fabricSend 'notch.open_notch("tmux")' # Tmux +bind = SUPER, SPACE, exec, $fabricSend "launcher.show_launcher()" +bind = SUPER, D, exec, $fabricSend "launcher.show_launcher('app ')" # Clipboard History -bind = SUPER, V, exec, $fabricSend 'notch.open_notch("cliphist")' -# Toolbox -bind = SUPER, S, exec, $fabricSend 'notch.open_notch("tools")' -bind = SUPER, TAB, exec, $fabricSend 'notch.open_notch("overview")' # Overview +bind = SUPER, V, exec, $fabricSend "launcher.show_launcher('clip ')" # Wallpapers -bind = SUPER, COMMA, exec, $fabricSend 'notch.open_notch("wallpapers")' -bind = SUPER SHIFT, COMMA, exec, $fabricSend 'notch.dashboard.wallpapers.set_random_wallpaper(None, external=True)' # Random Wallpaper +bind = SUPER, W, exec, $fabricSend "launcher.show_launcher('wall ')" +bind = SHIFT ALT, W, exec, $fabricSend "launcher.show_launcher('wall random', external=True)" # Random Wallpaper # Emoji Picker -bind = SUPER, PERIOD, exec, $fabricSend 'notch.open_notch("emoji")' +bind = SUPER, E, exec, $fabricSend 'notch.open_notch("emoji")' # Power Menu -bind = SUPER, ESCAPE, exec, $fabricSend 'notch.open_notch("power")' -bind = SUPER SHIFT, M, exec, $fabricSend 'notch.dashboard.widgets.buttons.caffeine_button.toggle_inhibit(external=True)' # Toggle Caffeine -# Reload CSS -bind = SUPER SHIFT, B, exec, $fabricSend 'app.set_css()' +bind = SUPER, ESCAPE, exec, $fabricSend "launcher.show_launcher('power ')" +bind = SUPER SHIFT, M, exec, $fabricSend "launcher.show_launcher('caffeine on', external=True)" # Toggle Caffeine +# Settings +bind = SUPER, I, exec, uwsm-app $(python /home/sankalp/Modus/config/config.py) # Settings +# Restart Modus bind = SUPER CTRL ALT, B, exec, killall modus; uwsm-app $(GTK_DEBUG=interactive python /home/sankalp/.config/Modus/main.py) # Restart with inspector # Wallpapers directory: /home/sankalp/Pictures/wallpapers diff --git a/config/launcher.json b/config/launcher.json index 58db1e5..0ebe9b9 100644 --- a/config/launcher.json +++ b/config/launcher.json @@ -105,18 +105,6 @@ "bt": { "icon": "bluetooth", "description": "Bluetooth - Manage Bluetooth devices and connections" - }, - "script": { - "icon": "toolbox", - "description": "Scripts - Execute system scripts and utilities" - }, - "cmd": { - "icon": "terminal", - "description": "Scripts - Quick access to system commands" - }, - "util": { - "icon": "config", - "description": "Utilities - System utilities and toggles" } }, "settings": { diff --git a/config/settings/constants.py b/config/settings/constants.py index b569bda..ce210cb 100644 --- a/config/settings/constants.py +++ b/config/settings/constants.py @@ -12,44 +12,34 @@ source = ~/{APP_NAME_CAP}/config/hypr/{APP_NAME}.conf """ DEFAULTS = { - "prefix_restart": "SUPER ALT", - "suffix_restart": "B", - "prefix_axmsg": "SUPER", - "suffix_axmsg": "A", + "prefix_restart": "ALT SHIFT", + "suffix_restart": "R", + "prefix_msg": "SUPER", + "suffix_msg": "A", "prefix_application_switcher": "ALT", "suffix_application_switcher": "TAB", - "prefix_dash": "SUPER", - "suffix_dash": "D", - "prefix_bluetooth": "SUPER SHIFT", - "suffix_bluetooth": "B", - "prefix_pins": "SUPER", - "suffix_pins": "Q", "prefix_kanban": "SUPER", - "suffix_kanban": "N", + "suffix_kanban": "T", "prefix_launcher": "SUPER", - "suffix_launcher": "R", - "prefix_tmux": "SUPER", - "suffix_tmux": "T", + "suffix_launcher": "SPACE", + "suffix_app_launcher": "D", + "prefix_app_launcher": "SUPER", "prefix_cliphist": "SUPER", "suffix_cliphist": "V", - "prefix_toolbox": "SUPER", - "suffix_toolbox": "S", - "prefix_overview": "SUPER", - "suffix_overview": "TAB", "prefix_wallpapers": "SUPER", - "suffix_wallpapers": "COMMA", - "prefix_randwall": "SUPER SHIFT", - "suffix_randwall": "COMMA", + "suffix_wallpapers": "W", + "prefix_randwall": "ALT SHIFT", + "suffix_randwall": "W", "prefix_emoji": "SUPER", - "suffix_emoji": "PERIOD", + "suffix_emoji": "E", "prefix_power": "SUPER", "suffix_power": "ESCAPE", "prefix_caffeine": "SUPER SHIFT", "suffix_caffeine": "M", - "prefix_css": "SUPER SHIFT", - "suffix_css": "B", "prefix_restart_inspector": "SUPER CTRL ALT", "suffix_restart_inspector": "B", + "prefix_settings": "SUPER", + "suffix_settings": "I", "wallpapers_dir": "/home/sankalp/Pictures/wallpapers", "vertical": True, "dock_enabled": True, diff --git a/config/settings/gui.py b/config/settings/gui.py index 813ad0b..de7ae55 100644 --- a/config/settings/gui.py +++ b/config/settings/gui.py @@ -123,27 +123,22 @@ class HyprConfGUI(Window): self.entries = [] bindings = [ (f"Reload {APP_NAME_CAP}", "prefix_restart", "suffix_restart"), - ("Message", "prefix_axmsg", "suffix_axmsg"), + ("Message", "prefix_msg", "suffix_msg"), ( "Application Switcher", "prefix_application_switcher", "suffix_application_switcher", ), - ("Dashboard", "prefix_dash", "suffix_dash"), - ("Bluetooth", "prefix_bluetooth", "suffix_bluetooth"), - ("Pins", "prefix_pins", "suffix_pins"), - ("Kanban", "prefix_kanban", "suffix_kanban"), - ("App Launcher", "prefix_launcher", "suffix_launcher"), - ("Tmux", "prefix_tmux", "suffix_tmux"), + ("Launcher", "prefix_launcher", "suffix_launcher"), + ("App Launcher", "prefix_app_launcher", "suffix_app_launcher"), ("Clipboard History", "prefix_cliphist", "suffix_cliphist"), - ("Toolbox", "prefix_toolbox", "suffix_toolbox"), - ("Overview", "prefix_overview", "suffix_overview"), ("Wallpapers", "prefix_wallpapers", "suffix_wallpapers"), ("Random Wallpaper", "prefix_randwall", "suffix_randwall"), ("Emoji Picker", "prefix_emoji", "suffix_emoji"), + ("Kanban", "prefix_kanban", "suffix_kanban"), ("Power Menu", "prefix_power", "suffix_power"), ("Toggle Caffeine", "prefix_caffeine", "suffix_caffeine"), - ("Reload CSS", "prefix_css", "suffix_css"), + ("Settings", "prefix_settings", "suffix_settings"), ( "Restart with inspector", "prefix_restart_inspector", diff --git a/config/settings/utils.py b/config/settings/utils.py index f21fea7..63c87d2 100644 --- a/config/settings/utils.py +++ b/config/settings/utils.py @@ -253,38 +253,32 @@ exec-once = wl-paste --type text --watch cliphist store exec-once = wl-paste --type image --watch cliphist store $fabricSend = fabric-cli exec {APP_NAME} -$axMessage = notify-send "Axenide" "FIRE IN THE HOLE‼️🗣️🔥🕳️" -i "{home}/.config/{APP_NAME_CAP}/assets/ax.png" -A "🗣️" -A "🔥" -A "🕳️" -a "Source Code" +$axMessage = notify-send "Modus" "FIRE IN THE HOLE‼️🗣️🔥🕳️" -i "{home}/{APP_NAME_CAP}/assets/modus.png" -A "🗣️" -A "🔥" -A "🕳️" -a "Source Code" # Reload {APP_NAME_CAP} -bind = {bind_vars.get("prefix_restart", "SUPER ALT")}, {bind_vars.get("suffix_restart", "B")}, exec, killall {APP_NAME}; uwsm-app $(python {home}/.config/{APP_NAME_CAP}/main.py) +bind = {bind_vars.get("prefix_restart", "ALT SHIFT")}, {bind_vars.get("suffix_restart", "R")}, exec, killall {APP_NAME}; uwsm-app $(python {home}/{APP_NAME_CAP}/main.py) # Message -bind = {bind_vars.get("prefix_axmsg", "SUPER")}, {bind_vars.get("suffix_axmsg", "A")}, exec, $axMessage +bind = {bind_vars.get("prefix_msg", "SUPER")}, {bind_vars.get("suffix_msg", "A")}, exec, $axMessage # Application Switcher bind = {bind_vars.get("prefix_application_switcher", "ALT")}, {bind_vars.get("suffix_application_switcher", "TAB")}, exec, $fabricSend 'switcher.show_switcher()' -# Dashboard -bind = {bind_vars.get("prefix_dash", "SUPER")}, {bind_vars.get("suffix_dash", "D")}, exec, $fabricSend 'notch.open_notch("dashboard")' -# Bluetooth -bind = {bind_vars.get("prefix_bluetooth", "SUPER")}, {bind_vars.get("suffix_bluetooth", "B")}, exec, $fabricSend 'notch.open_notch("bluetooth")' -bind = {bind_vars.get("prefix_pins", "SUPER")}, {bind_vars.get("suffix_pins", "Q")}, exec, $fabricSend 'notch.open_notch("pins")' # Pins -bind = {bind_vars.get("prefix_kanban", "SUPER")}, {bind_vars.get("suffix_kanban", "N")}, exec, $fabricSend 'notch.open_notch("kanban")' # Kanban +# Kanban +bind = {bind_vars.get("prefix_kanban", "SUPER")}, {bind_vars.get("suffix_kanban", "T")}, exec, $fabricSend "launcher.show_launcher('kanban ')" # Kanban # App Launcher -bind = {bind_vars.get("prefix_launcher", "SUPER")}, {bind_vars.get("suffix_launcher", "R")}, exec, $fabricSend 'notch.open_notch("launcher")' -bind = {bind_vars.get("prefix_tmux", "SUPER")}, {bind_vars.get("suffix_tmux", "T")}, exec, $fabricSend 'notch.open_notch("tmux")' # Tmux +bind = {bind_vars.get("prefix_launcher", "SUPER")}, {bind_vars.get("suffix_launcher", "SPACE")}, exec, $fabricSend "launcher.show_launcher()" +bind = {bind_vars.get("prefix_app_launcher", "SUPER")}, {bind_vars.get("suffix_app_launcher", "D")}, exec, $fabricSend "launcher.show_launcher('app ')" # Clipboard History -bind = {bind_vars.get("prefix_cliphist", "SUPER")}, {bind_vars.get("suffix_cliphist", "V")}, exec, $fabricSend 'notch.open_notch("cliphist")' -# Toolbox -bind = {bind_vars.get("prefix_toolbox", "SUPER")}, {bind_vars.get("suffix_toolbox", "S")}, exec, $fabricSend 'notch.open_notch("tools")' -bind = {bind_vars.get("prefix_overview", "SUPER")}, {bind_vars.get("suffix_overview", "TAB")}, exec, $fabricSend 'notch.open_notch("overview")' # Overview +bind = {bind_vars.get("prefix_cliphist", "SUPER")}, {bind_vars.get("suffix_cliphist", "V")}, exec, $fabricSend "launcher.show_launcher('clip ')" # Wallpapers -bind = {bind_vars.get("prefix_wallpapers", "SUPER")}, {bind_vars.get("suffix_wallpapers", "COMMA")}, exec, $fabricSend 'notch.open_notch("wallpapers")' -bind = {bind_vars.get("prefix_randwall", "SUPER")}, {bind_vars.get("suffix_randwall", "COMMA")}, exec, $fabricSend 'notch.dashboard.wallpapers.set_random_wallpaper(None, external=True)' # Random Wallpaper +bind = {bind_vars.get("prefix_wallpapers", "SUPER")}, {bind_vars.get("suffix_wallpapers", "W")}, exec, $fabricSend "launcher.show_launcher('wall ')" +bind = {bind_vars.get("prefix_randwall", "SUPER")}, {bind_vars.get("suffix_randwall", "W")}, exec, $fabricSend "launcher.show_launcher('wall random', external=True)" # Random Wallpaper # Emoji Picker -bind = {bind_vars.get("prefix_emoji", "SUPER")}, {bind_vars.get("suffix_emoji", "PERIOD")}, exec, $fabricSend 'notch.open_notch("emoji")' +bind = {bind_vars.get("prefix_emoji", "SUPER")}, {bind_vars.get("suffix_emoji", "E")}, exec, $fabricSend 'notch.open_notch("emoji")' # Power Menu -bind = {bind_vars.get("prefix_power", "SUPER")}, {bind_vars.get("suffix_power", "ESCAPE")}, exec, $fabricSend 'notch.open_notch("power")' -bind = {bind_vars.get("prefix_caffeine", "SUPER SHIFT")}, {bind_vars.get("suffix_caffeine", "M")}, exec, $fabricSend 'notch.dashboard.widgets.buttons.caffeine_button.toggle_inhibit(external=True)' # Toggle Caffeine -# Reload CSS -bind = {bind_vars.get("prefix_css", "SUPER SHIFT")}, {bind_vars.get("suffix_css", "B")}, exec, $fabricSend 'app.set_css()' +bind = {bind_vars.get("prefix_power", "SUPER")}, {bind_vars.get("suffix_power", "ESCAPE")}, exec, $fabricSend "launcher.show_launcher('power ')" +bind = {bind_vars.get("prefix_caffeine", "SUPER SHIFT")}, {bind_vars.get("suffix_caffeine", "M")}, exec, $fabricSend "launcher.show_launcher('caffeine on', external=True)" # Toggle Caffeine +# Settings +bind = {bind_vars.get("prefix_settings", "SUPER")}, {bind_vars.get("suffix_settings", "I")}, exec, uwsm-app $(python {home}/{APP_NAME_CAP}/config/config.py) # Settings +# Restart Modus bind = {bind_vars.get("prefix_restart_inspector", "SUPER CTRL ALT")}, {bind_vars.get("suffix_restart_inspector", "B")}, exec, killall {APP_NAME}; uwsm-app $(GTK_DEBUG=interactive python {home}/.config/{APP_NAME_CAP}/main.py) # Restart with inspector # Wallpapers directory: {bind_vars.get("wallpapers_dir", "~/Modus/assets/wallpapers_example")} diff --git a/main.py b/main.py index f343a46..9220dc6 100644 --- a/main.py +++ b/main.py @@ -75,23 +75,5 @@ if __name__ == "__main__": app.set_css() - # Register launcher actions - @Application.action("show-launcher") - def show_launcher(): - """Show the launcher window.""" - launcher.show_launcher() - - @Application.action("hide-launcher") - def hide_launcher(): - """Hide the launcher window.""" - launcher.hide_launcher() - - @Application.action("toggle-launcher") - def toggle_launcher(): - """Toggle the launcher window visibility.""" - if launcher.visible: - launcher.hide_launcher() - else: - launcher.show_launcher() app.run() diff --git a/modules/dock/components/applications.py b/modules/dock/components/applications.py index 45fa6d5..6cf2d7c 100644 --- a/modules/dock/components/applications.py +++ b/modules/dock/components/applications.py @@ -394,16 +394,14 @@ class Applications(Box): for child in children: self.add(child) - # Check if the applications component is enabled in the dock configuration - # Import here to avoid circular imports - import config.data as data - applications_visible = data.DOCK_COMPONENTS_VISIBILITY.get("applications", True) - - # Show or hide based on whether there are any applications AND if component is enabled - if len(children) > 0 and applications_visible: + # Only show children if there are any applications + # Note: Component-level visibility is handled by the parent DockComponents class + if len(children) > 0: self.show_all() else: - self.hide() + # Hide the widget content but don't override component-level visibility + for child in self.get_children(): + child.hide() return True @@ -564,7 +562,7 @@ class Applications(Box): if hasattr(self, '_dragged_widget') and self._dragged_widget: children = self.get_children() source_index = children.index(self._dragged_widget) - + # Find separator index separator_index = -1 for i, child in enumerate(children): diff --git a/modules/launcher/main.py b/modules/launcher/main.py index e4bc0e3..46eff5f 100644 --- a/modules/launcher/main.py +++ b/modules/launcher/main.py @@ -146,6 +146,9 @@ class Launcher(Window): self.show_all() if trigger_keyword: + # Set flag to track that launcher was opened with a trigger keyword + self.opened_with_trigger = True + # Set the trigger keyword with a space and activate trigger mode trigger_text = f"{trigger_keyword} " self.search_entry.set_text(trigger_text) @@ -173,6 +176,7 @@ class Launcher(Window): self._clear_results() else: # Normal launcher opening - clear everything + self.opened_with_trigger = False self.search_entry.set_text("") self._clear_results() @@ -198,6 +202,8 @@ class Launcher(Window): self.triggered_plugin = None self.active_trigger = "" self.visible = False + self.opened_with_trigger = False # Reset the flag when closing + self.opened_with_trigger = False # Track if launcher was opened with a trigger keyword def _on_search_changed(self, entry): """Handle search text changes.""" @@ -680,6 +686,10 @@ class Launcher(Window): # Cancel the password entry password_entry_widget.cancel_password_entry() return True + elif self.opened_with_trigger: + # If launcher was opened with a trigger keyword, close directly + self.close_launcher() + return True elif self.triggered_plugin: # Exit trigger mode self.triggered_plugin = None diff --git a/modules/launcher/plugin_manager.py b/modules/launcher/plugin_manager.py index 5e1c123..b3a6601 100644 --- a/modules/launcher/plugin_manager.py +++ b/modules/launcher/plugin_manager.py @@ -97,7 +97,8 @@ class PluginManager: "calendar", "reminders", "network", - "bluetooth" + "bluetooth", + "otp" ] for plugin_name in default_plugins: diff --git a/modules/launcher/plugins/caffeine.py b/modules/launcher/plugins/caffeine.py index 87c5420..f57ee8b 100644 --- a/modules/launcher/plugins/caffeine.py +++ b/modules/launcher/plugins/caffeine.py @@ -5,12 +5,15 @@ Prevents system from going idle for a specified duration. import subprocess from typing import List +from threading import Timer import gi import utils.icons as icons from fabric.utils import get_relative_path +from fabric.utils.helpers import exec_shell_command_async from modules.launcher.plugin_base import PluginBase from modules.launcher.result import Result +import config.data as data gi.require_version("Gtk", "3.0") @@ -57,11 +60,59 @@ class CaffeinePlugin(PluginBase): stderr=subprocess.DEVNULL, start_new_session=True, # Run in a new session to prevent hanging ) + + # Send notification based on duration + if duration.lower() == "off": + # Deactivation notification + exec_shell_command_async( + f"notify-send '☕ Caffeine' 'Deactivated' -a '{data.APP_NAME_CAP}' -e" + ) + else: + # Activation notification with duration + duration_text = self.durations.get(duration, duration) + exec_shell_command_async( + f"notify-send '☕ Caffeine' 'Activated for {duration_text}' -a '{data.APP_NAME_CAP}' -e" + ) + + # Schedule expiration notification for timed durations + if duration.lower() not in ["on", "off"]: + self._schedule_expiration_notification(duration, duration_text) + except Exception as e: print(f"Error starting inhibit script: {e}") return action + def _parse_duration_to_seconds(self, duration_str: str) -> int: + """Parse duration string into seconds. Same logic as inhibit.py""" + try: + if duration_str.lower() in ['on', 'off']: + return 0 + elif duration_str.endswith('h'): + return int(float(duration_str[:-1]) * 3600) + elif duration_str.endswith('m'): + return int(float(duration_str[:-1]) * 60) + elif duration_str.endswith('s'): + return int(float(duration_str[:-1])) + else: + return int(duration_str) + except ValueError: + return 0 + + def _schedule_expiration_notification(self, duration: str, duration_text: str): + """Schedule a notification for when the caffeine effect expires.""" + seconds = self._parse_duration_to_seconds(duration) + if seconds > 0: + def send_expiration_notification(): + exec_shell_command_async( + f"notify-send '☕ Caffeine' 'Expired after {duration_text}' -a '{data.APP_NAME_CAP}' -e" + ) + + # Schedule the notification + timer = Timer(seconds, send_expiration_notification) + timer.daemon = True # Don't prevent program exit + timer.start() + def _is_valid_duration(self, query: str) -> bool: """Check if the query is a valid duration format.""" if query in self.durations: diff --git a/modules/launcher/plugins/clipboard.py b/modules/launcher/plugins/clipboard.py index 78b119b..f1c3858 100644 --- a/modules/launcher/plugins/clipboard.py +++ b/modules/launcher/plugins/clipboard.py @@ -1,12 +1,16 @@ """ Clipboard plugin using cliphist for clipboard history management. +Optimized for performance with caching, threading, and lazy loading. """ import os import subprocess import sys import tempfile -from typing import List, Dict +import threading +import time +from typing import List, Dict, Optional +from concurrent.futures import ThreadPoolExecutor, as_completed from modules.launcher.plugin_base import PluginBase from modules.launcher.result import Result from gi.repository import GdkPixbuf, GLib @@ -19,41 +23,82 @@ class ClipboardPlugin(PluginBase): self.display_name = "Clipboard History" self.description = "Search and manage clipboard history using cliphist" + # Performance settings + self.max_results = 20 + self.cache_ttl = 30 # Cache clipboard items for 30 seconds + self.image_cache_ttl = 300 # Cache images for 5 minutes + # Initialize cache and temp directory self.tmp_dir = tempfile.mkdtemp(prefix="cliphist-") self.image_cache: Dict[str, GdkPixbuf.Pixbuf] = {} - self.clipboard_items = [] + self.clipboard_items_cache: List[str] = [] + self.cache_timestamp = 0 + self.image_cache_timestamps: Dict[str, float] = {} + + # Threading + self.executor = ThreadPoolExecutor(max_workers=2, thread_name_prefix="clipboard") + self.cache_lock = threading.Lock() + + # State tracking self._loading = False self._pending_updates = False def initialize(self): - self.set_triggers(["clip", "clip "]) """Initialize the plugin.""" + self.set_triggers(["clip", "clip "]) try: - subprocess.run(["cliphist", "list"], capture_output=True, check=True) + subprocess.run(["cliphist", "list"], capture_output=True, check=True, timeout=5) except (subprocess.SubprocessError, FileNotFoundError): raise RuntimeError("cliphist is not installed or not working properly") + # Pre-warm cache in background + self.executor.submit(self._load_clipboard_items_cached) + def cleanup(self): """Cleanup the plugin.""" try: + # Shutdown executor + if hasattr(self, 'executor'): + self.executor.shutdown(wait=False) + + # Clean up temp files if os.path.exists(self.tmp_dir): import shutil - shutil.rmtree(self.tmp_dir) - self.image_cache.clear() + + # Clear caches + with self.cache_lock: + self.image_cache.clear() + self.clipboard_items_cache.clear() + self.image_cache_timestamps.clear() except Exception as e: print(f"Error cleaning up temporary files: {e}", file=sys.stderr) - def _load_clipboard_items(self) -> List[str]: - """Load clipboard items from cliphist.""" + def _load_clipboard_items_cached(self) -> List[str]: + """Load clipboard items from cliphist with caching.""" + current_time = time.time() + + # Check if cache is still valid + with self.cache_lock: + if (self.clipboard_items_cache and + current_time - self.cache_timestamp < self.cache_ttl): + return self.clipboard_items_cache.copy() + + # Load fresh data try: result = subprocess.run( - ["cliphist", "list"], capture_output=True, check=True + ["cliphist", "list"], capture_output=True, check=True, timeout=5 ) stdout_str = result.stdout.decode("utf-8", errors="replace") lines = stdout_str.strip().split("\n") - return [line for line in lines if line and " List[str]: + """Load clipboard items (legacy method for compatibility).""" + return self._load_clipboard_items_cached() + def _create_pixbuf_from_bytes( self, image_data: bytes, max_size: int = 100 ) -> GdkPixbuf.Pixbuf: @@ -98,25 +147,37 @@ class ClipboardPlugin(PluginBase): return content[:37] + "..." return content - def _load_image_preview_async(self, item_id: str) -> GdkPixbuf.Pixbuf: - """Load image preview asynchronously.""" - if item_id in self.image_cache: - return self.image_cache[item_id] + def _load_image_preview_cached(self, item_id: str) -> Optional[GdkPixbuf.Pixbuf]: + """Load image preview with caching and timeout.""" + current_time = time.time() + + # Check cache first + with self.cache_lock: + if (item_id in self.image_cache and + item_id in self.image_cache_timestamps and + current_time - self.image_cache_timestamps[item_id] < self.image_cache_ttl): + return self.image_cache[item_id] try: result = subprocess.run( - ["cliphist", "decode", item_id], capture_output=True, check=True + ["cliphist", "decode", item_id], capture_output=True, check=True, timeout=3 ) pixbuf = self._create_pixbuf_from_bytes(result.stdout) if pixbuf: - self.image_cache[item_id] = pixbuf + with self.cache_lock: + self.image_cache[item_id] = pixbuf + self.image_cache_timestamps[item_id] = current_time return pixbuf except Exception as e: print(f"Error loading image preview: {e}", file=sys.stderr) return None + def _load_image_preview_async(self, item_id: str) -> Optional[GdkPixbuf.Pixbuf]: + """Load image preview (legacy method for compatibility).""" + return self._load_image_preview_cached(item_id) + def query(self, query_string: str) -> List[Result]: - """Search clipboard history using cliphist.""" + """Search clipboard history using cliphist with optimized performance.""" results = [] # Handle query string @@ -124,46 +185,82 @@ class ClipboardPlugin(PluginBase): query_string = "" # Show all items try: - # Load clipboard items - self.clipboard_items = self._load_clipboard_items() + # Load clipboard items from cache + clipboard_items = self._load_clipboard_items_cached() - # Filter items based on query + # Early exit if no items + if not clipboard_items: + return results + + # Filter items based on query with early termination filtered_items = [] - for item in self.clipboard_items: + query_lower = query_string.lower() if query_string else "" + + for item in clipboard_items: + # Stop if we have enough results + if len(filtered_items) >= self.max_results: + break + parts = item.split("\t", 1) content = parts[1] if len(parts) > 1 else item - if not query_string or query_string.lower() in content.lower(): + + # Fast filtering + if not query_lower or query_lower in content.lower(): filtered_items.append(item) - # Process all items without batching - for item in filtered_items: + # Process items with lazy image loading + for i, item in enumerate(filtered_items): + # Limit total results + if len(results) >= self.max_results: + break + parts = item.split("\t", 1) - item_id = parts[0] if len(parts) > 1 else "0" + item_id = parts[0] if len(parts) > 1 else str(i) content = parts[1] if len(parts) > 1 else item - # Handle image content + # Handle image content with lazy loading if self._is_image_data(content): - pixbuf = self._load_image_preview_async(item_id) - if pixbuf: + # Check if image is already cached + cached_pixbuf = None + with self.cache_lock: + cached_pixbuf = self.image_cache.get(item_id) + + if cached_pixbuf: + # Use cached image result = Result( title="Image from clipboard", subtitle="Click to copy image to clipboard", description="Image content", - icon=pixbuf, + icon=cached_pixbuf, relevance=1.0, plugin_name=self.name, action=lambda id=item_id: self._copy_to_clipboard(id), data={"bypass_max_results": True}, ) - results.append(result) - continue + else: + # Show placeholder and load image in background + result = Result( + title="Image from clipboard", + subtitle="Loading preview...", + description="Image content", + icon_name="image-x-generic", + relevance=1.0, + plugin_name=self.name, + action=lambda id=item_id: self._copy_to_clipboard(id), + data={"bypass_max_results": True}, + ) + # Load image in background (don't wait for it) + self.executor.submit(self._load_image_preview_cached, item_id) + + results.append(result) + continue # Handle text content display_text = self._get_text_preview(content) result = Result( title=display_text, subtitle="Text from clipboard", - description=content, + description=content if len(content) <= 100 else content[:97] + "...", icon_name="edit-paste", relevance=1.0, plugin_name=self.name, @@ -188,19 +285,22 @@ class ClipboardPlugin(PluginBase): return results def _copy_to_clipboard(self, entry_id: str): - """Copy entry to clipboard using cliphist.""" + """Copy entry to clipboard using cliphist with timeout.""" try: result = subprocess.run( - ["cliphist", "decode", entry_id], capture_output=True, check=True + ["cliphist", "decode", entry_id], capture_output=True, check=True, timeout=5 ) # Use wl-copy for Wayland or xclip for X11 try: - subprocess.run(["wl-copy"], input=result.stdout, check=True) + subprocess.run(["wl-copy"], input=result.stdout, check=True, timeout=3) except subprocess.SubprocessError: subprocess.run( ["xclip", "-selection", "clipboard"], input=result.stdout, check=True, + timeout=3 ) except subprocess.SubprocessError as e: print(f"Error copying to clipboard: {e}", file=sys.stderr) + except subprocess.TimeoutExpired as e: + print(f"Timeout copying to clipboard: {e}", file=sys.stderr) diff --git a/modules/launcher/plugins/otp.py b/modules/launcher/plugins/otp.py new file mode 100644 index 0000000..59476ac --- /dev/null +++ b/modules/launcher/plugins/otp.py @@ -0,0 +1,808 @@ +""" +OTP Manager plugin for the launcher. +Provides TOTP (Time-based One-Time Password) management. +""" + +import json +import subprocess +import time +import threading +from typing import List, Dict, Optional +from pathlib import Path +import base64 + +from fabric.utils import get_relative_path +from modules.launcher.plugin_base import PluginBase +from modules.launcher.result import Result +import utils.icons as icons + +try: + import pyotp + PYOTP_AVAILABLE = True +except ImportError: + PYOTP_AVAILABLE = False + + +class OTPPlugin(PluginBase): + """Plugin for managing TOTP (Time-based One-Time Password) codes.""" + + def __init__(self): + super().__init__() + self.display_name = "OTP Manager" + self.description = "Manage TOTP codes and 2FA authentication" + + self.secrets_file = Path(get_relative_path("../../../config/otp_accounts.json")) + self.secrets: Dict[str, Dict] = {} + self.last_update = 0 + + # Threading for auto-refresh + self.refresh_thread = None + self.stop_refresh = threading.Event() + + def initialize(self): + """Initialize the OTP plugin.""" + if not PYOTP_AVAILABLE: + raise RuntimeError("pyotp is required for OTP functionality. Install with: pip install pyotp") + + self.set_triggers(["otp", "2fa", "totp", "auth"]) + self._load_secrets() + self._ensure_config_file() + self._start_refresh_thread() + + def cleanup(self): + """Cleanup the OTP plugin.""" + if self.refresh_thread and self.refresh_thread.is_alive(): + self.stop_refresh.set() + self.refresh_thread.join(timeout=1) + + def _load_secrets(self): + """Load secrets from JSON file.""" + try: + if self.secrets_file.exists(): + with open(self.secrets_file, 'r', encoding='utf-8') as f: + self.secrets = json.load(f) + else: + self.secrets = {} + except Exception as e: + print(f"Error loading OTP secrets: {e}") + self.secrets = {} + + def _save_secrets(self): + """Save secrets to JSON file.""" + try: + self.secrets_file.parent.mkdir(parents=True, exist_ok=True) + with open(self.secrets_file, 'w', encoding='utf-8') as f: + json.dump(self.secrets, f, indent=2) + except Exception as e: + print(f"Error saving OTP secrets: {e}") + subprocess.run([ + "notify-send", "OTP Manager Error", + f"Failed to save config: {str(e)}", + "-i", "dialog-error" + ], check=False) + + def _ensure_config_file(self): + """Ensure the config file exists.""" + if not self.secrets_file.exists(): + self.secrets_file.parent.mkdir(parents=True, exist_ok=True) + with open(self.secrets_file, 'w', encoding='utf-8') as f: + json.dump({}, f, indent=2) + + def _start_refresh_thread(self): + """Start background thread for auto-refreshing tokens.""" + def refresh_loop(): + while not self.stop_refresh.wait(1): + current_time = time.time() + if current_time - self.last_update >= 1: + self.last_update = current_time + if self.secrets: + 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 time display in existing OTP result items.""" + try: + from gi.repository import GLib + import gc + + 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_otp_results = any( + result.data and result.data.get("type") == "totp" + for result in obj.results + if hasattr(result, 'data') and result.data + ) + + if has_otp_results: + self._update_existing_result_labels(obj.results_box) + return False + except Exception: + pass + return False + + GLib.idle_add(do_update) + except Exception: + pass + + def _update_existing_result_labels(self, results_box): + """Update subtitle labels in existing ResultItem widgets.""" + try: + time_display = self._get_time_remaining_with_blink() + 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") == "totp"): + self._update_result_item_content(child, time_display) + except Exception as e: + print(f"Error updating result labels: {e}") + + def _update_result_item_content(self, result_item, time_display): + """Update both the title (OTP code) and subtitle (time display) of a specific ResultItem.""" + try: + account_name = result_item.result.data.get("account", "") + if not account_name or account_name not in self.secrets: + return + + account_data = self.secrets[account_name] + secret = account_data.get("secret", "") + issuer = account_data.get("issuer", "") + display_name = f"{issuer} - {account_name}" if issuer else account_name + + current_totp_code = self._generate_totp(secret) + if not current_totp_code: + return + + old_code = result_item.result.data.get("code", "") + if current_totp_code != old_code: + result_item.result.data["code"] = current_totp_code + self._find_and_update_title_label(result_item, current_totp_code) + result_item.result.action = lambda code=current_totp_code: self._copy_to_clipboard(code) + + new_subtitle_markup = f"{display_name} • {time_display} remaining" + self._find_and_update_subtitle_label(result_item, new_subtitle_markup) + except Exception as e: + print(f"Error updating result item: {e}") + + def _find_and_update_title_label(self, result_item, new_title): + """Find the title label widget and update its text.""" + def find_title_label(widget): + if hasattr(widget, 'get_name') and widget.get_name() == "result-item-title": + return widget + if hasattr(widget, 'get_children'): + for child in widget.get_children(): + found = find_title_label(child) + if found: + return found + return None + + title_label = find_title_label(result_item) + if title_label and hasattr(title_label, 'set_label'): + title_label.set_label(new_title) + + def _find_and_update_subtitle_label(self, result_item, new_markup): + """Find the subtitle label widget and update its markup.""" + 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_markup'): + subtitle_label.set_markup(new_markup) + + def _copy_to_clipboard(self, text: str): + """Copy text to clipboard.""" + try: + try: + subprocess.run(["wl-copy"], input=text.encode(), check=True) + except subprocess.SubprocessError: + subprocess.run(["xclip", "-selection", "clipboard"], input=text.encode(), check=True) + except Exception as e: + print(f"Failed to copy to clipboard: {e}") + + def _trigger_refresh(self): + """Trigger launcher refresh to return to default OTP view.""" + 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, 'search_entry'): + # Clear the search entry and set it to just "otp " + launcher.search_entry.set_text("otp ") + # Position cursor at the end + launcher.search_entry.set_position(-1) + # Trigger the search to show default OTP view + if hasattr(launcher, '_perform_search'): + launcher._perform_search("otp ") + 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, 'search_entry') and hasattr(obj, '_perform_search'): + obj.search_entry.set_text("otp ") + obj.search_entry.set_position(-1) + obj._perform_search("otp ") + 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 _remove_account_and_refresh(self, account_name: str): + """Remove an account and trigger refresh to return to default OTP view.""" + try: + if account_name in self.secrets: + # Get account info for notification + account_data = self.secrets[account_name] + issuer = account_data.get("issuer", "") + display_name = f"{issuer} - {account_name}" if issuer else account_name + + # Remove the account + del self.secrets[account_name] + self._save_secrets() + + # Show notification + subprocess.run([ + "notify-send", "OTP Manager", + f"Removed '{display_name}' successfully", + "-i", "dialog-information" + ], check=False) + + # Trigger refresh to return to default OTP view + self._trigger_refresh() + else: + # Account not found + subprocess.run([ + "notify-send", "OTP Manager Error", + f"Account '{account_name}' not found", + "-i", "dialog-error" + ], check=False) + except Exception as e: + print(f"Error removing account {account_name}: {e}") + subprocess.run([ + "notify-send", "OTP Manager Error", + f"Failed to remove account: {str(e)}", + "-i", "dialog-error" + ], check=False) + + def _generate_totp(self, secret: str) -> Optional[str]: + """Generate TOTP code from secret.""" + try: + return pyotp.TOTP(secret).now() + except Exception as e: + print(f"Error generating TOTP: {e}") + return None + + def _get_time_remaining(self) -> int: + """Get seconds remaining until next token refresh.""" + return 30 - (int(time.time()) % 30) + + def _get_time_remaining_with_blink(self) -> str: + """Get time remaining with blinking effect.""" + time_remaining = self._get_time_remaining() + current_second = int(time.time()) + should_blink = (current_second % 2 == 0) + + if should_blink: + return f"{time_remaining}s" + else: + return f"{time_remaining}s" + + def query(self, query_string: str) -> List[Result]: + """Process OTP queries.""" + query = query_string.strip() + + if not query: + return self._list_otp_codes() + + query_lower = query.lower() + if query_lower.startswith("add "): + add_content = query[4:].strip() + + # Handle both formats: with ``` and without ``` + if "```" in add_content: + # Old format: add account```secret``` + parts = add_content.split("```", 1) + if len(parts) == 2: + account_name = parts[0].strip() + secret_or_uri = parts[1].strip() + return self._handle_direct_add(account_name, secret_or_uri) + elif " " in add_content: + # New format: add account secret + parts = add_content.split(" ", 1) + if len(parts) == 2: + account_name = parts[0].strip() + secret_or_uri = parts[1].strip() + return self._handle_direct_add(account_name, secret_or_uri) + + return self._handle_add_command(add_content) + elif query_lower == "remove" or query_lower.startswith("remove "): + # Handle both "remove" and "remove accountname" + if query_lower == "remove": + remove_content = "" + else: + remove_content = query[7:].strip() + return self._handle_remove_command(remove_content) + else: + return self._search_accounts(query) + + def _handle_direct_add(self, account_name: str, secret_or_uri: str) -> List[Result]: + """Handle direct addition of OTP account.""" + if not account_name or not secret_or_uri: + return [Result( + title="Invalid format", + subtitle="Usage: add or add ``````", + icon_markup=icons.question, + action=lambda: None, + relevance=1.0, + plugin_name=self.display_name, + data={"type": "help", "keep_launcher_open": True} + )] + + # Debug info + print(f"OTP Debug: Adding account '{account_name}' with secret/URI: '{secret_or_uri[:20]}...'") + + try: + if secret_or_uri.startswith("otpauth://"): + print("OTP Debug: Processing as otpauth URI") + return self._handle_otpauth_uri(account_name, secret_or_uri) + else: + print("OTP Debug: Processing as Base32 secret") + return self._handle_base32_secret(account_name, secret_or_uri) + except Exception as e: + print(f"OTP Debug: Error in _handle_direct_add: {e}") + return [Result( + title="Error adding account", + subtitle=f"Debug: {str(e)}", + icon_markup=icons.cancel, + action=lambda: None, + relevance=0.5, + plugin_name=self.display_name, + data={"type": "error", "keep_launcher_open": True} + )] + + def _list_otp_codes(self) -> List[Result]: + """List all OTP codes with current tokens.""" + results = [] + + if not self.secrets: + results.append(Result( + title="No OTP accounts configured", + subtitle="Use 'add ' to add your first account", + icon_markup=icons.key, + action=lambda: None, + relevance=1.0, + plugin_name=self.display_name, + data={"type": "empty", "keep_launcher_open": True} + )) + results.append(Result( + title="Available commands:", + subtitle="add | remove ", + icon_markup=icons.info, + action=lambda: None, + relevance=0.9, + plugin_name=self.display_name, + data={"type": "help", "keep_launcher_open": True} + )) + return results + + time_display = self._get_time_remaining_with_blink() + + for account_name, account_data in self.secrets.items(): + secret = account_data.get("secret", "") + issuer = account_data.get("issuer", "") + totp_code = self._generate_totp(secret) + + if totp_code: + display_name = f"{issuer} - {account_name}" if issuer else account_name + results.append(Result( + title=f"{totp_code}", + subtitle_markup=f"{display_name} • {time_display} remaining", + icon_markup=icons.key, + action=lambda code=totp_code: self._copy_to_clipboard(code), + relevance=1.0, + plugin_name=self.display_name, + data={"type": "totp", "account": account_name, "code": totp_code} + # Note: No keep_launcher_open flag - launcher will close after copying + )) + else: + results.append(Result( + title=f"Error: {account_name}", + subtitle="Invalid secret or configuration", + icon_markup=icons.cancel, + action=lambda: None, + relevance=0.5, + plugin_name=self.display_name, + data={"type": "error", "account": account_name, "keep_launcher_open": True} + )) + + return results + + def _search_accounts(self, query: str) -> List[Result]: + """Search accounts by name or issuer.""" + results = [] + query_lower = query.lower() + + for account_name, account_data in self.secrets.items(): + issuer = account_data.get("issuer", "").lower() + account_lower = account_name.lower() + + if query_lower in account_lower or query_lower in issuer: + secret = account_data.get("secret", "") + totp_code = self._generate_totp(secret) + + if totp_code: + display_name = f"{account_data.get('issuer', '')} - {account_name}" if account_data.get('issuer') else account_name + time_display = self._get_time_remaining_with_blink() + + results.append(Result( + title=f"{totp_code}", + subtitle_markup=f"{display_name} • {time_display} remaining", + icon_markup=icons.key, + action=lambda code=totp_code: self._copy_to_clipboard(code), + relevance=1.0, + plugin_name=self.display_name, + data={"type": "totp", "account": account_name, "code": totp_code} + # Note: No keep_launcher_open flag - launcher will close after copying + )) + + if not results: + results.append(Result( + title=f"No accounts found for '{query}'", + subtitle="Use 'add ' to create new account", + icon_markup=icons.magnifier, + action=lambda: None, + relevance=0.5, + plugin_name=self.display_name, + data={"type": "no_results", "keep_launcher_open": True} + )) + results.append(Result( + title="Available commands:", + subtitle="add | remove ", + icon_markup=icons.info, + action=lambda: None, + relevance=0.4, + plugin_name=self.display_name, + data={"type": "help", "keep_launcher_open": True} + )) + + return results + + def _handle_add_command(self, account_name: str) -> List[Result]: + """Handle manual addition of OTP secret.""" + if not account_name: + return [Result( + title="Enter account name", + subtitle="Usage: add ", + icon_markup=icons.question, + action=lambda: None, + relevance=1.0, + plugin_name=self.display_name, + data={"type": "help", "keep_launcher_open": True} + )] + + return [ + Result( + title=f"To add '{account_name}':", + subtitle=f"Type: add {account_name} ", + icon_markup=icons.bulb, + action=lambda: None, + relevance=1.0, + plugin_name=self.display_name, + data={"type": "instruction", "account": account_name, "keep_launcher_open": True} + ), + Result( + title="Base32 Secret Format:", + subtitle="Example: add gmail JBSWY3DPEHPK3PXP", + icon_markup=icons.info, + action=lambda: None, + relevance=0.9, + plugin_name=self.display_name, + data={"type": "help", "keep_launcher_open": True} + ), + Result( + title="otpauth URI Format:", + subtitle="Example: add github otpauth://totp/GitHub:user?secret=JBSWY3DPEHPK3PXP", + icon_markup=icons.info, + action=lambda: None, + relevance=0.9, + plugin_name=self.display_name, + data={"type": "help", "keep_launcher_open": True} + ), + Result( + title="Remove Account:", + subtitle="Example: remove gmail", + icon_markup=icons.trash, + action=lambda: None, + relevance=0.8, + plugin_name=self.display_name, + data={"type": "help", "keep_launcher_open": True} + ) + ] + + def _handle_remove_command(self, account_name: str) -> List[Result]: + """Handle removal of OTP account.""" + if not account_name: + # Show all available accounts for removal + results = [] + + if not self.secrets: + results.append(Result( + title="No OTP accounts to remove", + subtitle="Use 'add ' to add accounts first", + icon_markup=icons.info, + action=lambda: None, + relevance=1.0, + plugin_name=self.display_name, + data={"type": "empty", "keep_launcher_open": True} + )) + else: + results.append(Result( + title="Select account to remove:", + subtitle="Type: remove to remove an account", + icon_markup=icons.trash, + action=lambda: None, + relevance=1.0, + plugin_name=self.display_name, + data={"type": "help", "keep_launcher_open": True} + )) + + # Get time display for consistency with main OTP view + time_display = self._get_time_remaining_with_blink() + + # Show all accounts with their current OTP codes and remove actions + for acc_name, account_data in self.secrets.items(): + secret = account_data.get("secret", "") + issuer = account_data.get("issuer", "") + display_name = f"{issuer} - {acc_name}" if issuer else acc_name + + # Generate current TOTP code + totp_code = self._generate_totp(secret) + + if totp_code: + results.append(Result( + title=f"{totp_code}", + subtitle_markup=f"Remove {display_name} • {time_display} remaining", + icon_markup=icons.trash, + action=lambda account=acc_name: self._remove_account_and_refresh(account), + relevance=0.9, + plugin_name=self.display_name, + data={"type": "remove_option", "account": acc_name, "code": totp_code, "keep_launcher_open": True} + )) + else: + results.append(Result( + title=f"Error: {acc_name}", + subtitle=f"Remove {display_name} (Invalid secret)", + icon_markup=icons.trash, + action=lambda account=acc_name: self._remove_account_and_refresh(account), + relevance=0.8, + plugin_name=self.display_name, + data={"type": "remove_option", "account": acc_name, "keep_launcher_open": True} + )) + + return results + + # Check if account exists + if account_name not in self.secrets: + return [Result( + title=f"Account '{account_name}' not found", + subtitle="Use 'remove' to see all available accounts", + icon_markup=icons.cancel, + action=lambda: None, + relevance=0.5, + plugin_name=self.display_name, + data={"type": "error", "keep_launcher_open": True} + )] + + # Get account info for confirmation + account_data = self.secrets[account_name] + issuer = account_data.get("issuer", "") + display_name = f"{issuer} - {account_name}" if issuer else account_name + + # Remove the account + try: + del self.secrets[account_name] + self._save_secrets() + + return [Result( + title=f"✓ Removed '{display_name}'", + subtitle="OTP account removed successfully", + icon_markup=icons.check, + action=lambda: self._trigger_refresh(), + relevance=1.0, + plugin_name=self.display_name, + data={"type": "success", "keep_launcher_open": True} + )] + except Exception as e: + return [Result( + title="Error removing account", + subtitle=f"Failed to remove account: {str(e)}", + icon_markup=icons.cancel, + action=lambda: None, + relevance=0.5, + plugin_name=self.display_name, + data={"type": "error", "keep_launcher_open": True} + )] + + + + def _handle_base32_secret(self, account_name: str, secret: str) -> List[Result]: + """Handle raw Base32 secret.""" + try: + # Clean up the secret - remove spaces, dashes, and convert to uppercase + clean_secret = secret.replace(" ", "").replace("-", "").replace("_", "").upper() + + # Remove any non-base32 characters + import re + clean_secret = re.sub(r'[^A-Z2-7]', '', clean_secret) + + # Add padding if needed (Base32 requires padding to multiple of 8) + while len(clean_secret) % 8 != 0: + clean_secret += '=' + + # Validate Base32 format + try: + base64.b32decode(clean_secret) + except Exception as e: + return [Result( + title="Invalid Base32 secret", + subtitle=f"Error: {str(e)}. Please check the secret format", + icon_markup=icons.cancel, + action=lambda: None, + relevance=0.5, + plugin_name=self.display_name, + data={"type": "error", "keep_launcher_open": True} + )] + + # Test if the secret can generate a valid TOTP + try: + test_totp = pyotp.TOTP(clean_secret) + test_code = test_totp.now() + if not test_code or len(test_code) != 6: + raise ValueError("Generated invalid TOTP code") + except Exception as e: + return [Result( + title="Invalid secret for TOTP", + subtitle=f"Cannot generate TOTP: {str(e)}", + icon_markup=icons.cancel, + action=lambda: None, + relevance=0.5, + plugin_name=self.display_name, + data={"type": "error", "keep_launcher_open": True} + )] + + self.secrets[account_name] = { + "secret": clean_secret, + "issuer": "", + "algorithm": "SHA1", + "digits": 6, + "period": 30 + } + self._save_secrets() + + return [Result( + title=f"✓ Added '{account_name}'", + subtitle=f"OTP account added successfully (secret: {clean_secret[:4]}...)", + icon_markup=icons.check, + action=lambda: self._trigger_refresh(), + relevance=1.0, + plugin_name=self.display_name, + data={"type": "success", "keep_launcher_open": True} + )] + except Exception as e: + return [Result( + title="Error adding account", + subtitle=f"Unexpected error: {str(e)}", + icon_markup=icons.cancel, + action=lambda: None, + relevance=0.5, + plugin_name=self.display_name, + data={"type": "error", "keep_launcher_open": True} + )] + + def _handle_otpauth_uri(self, account_name: str, uri: str) -> List[Result]: + """Handle otpauth:// URI.""" + try: + from urllib.parse import urlparse, parse_qs + + parsed = urlparse(uri) + if parsed.scheme != "otpauth" or parsed.netloc != "totp": + return [Result( + title="Invalid otpauth URI", + subtitle="Only otpauth://totp/ URIs are supported", + icon_markup=icons.cancel, + action=lambda: None, + relevance=0.5, + plugin_name=self.display_name, + data={"type": "error", "keep_launcher_open": True} + )] + + if not account_name: + account_path = parsed.path.lstrip("/") + if ":" in account_path: + issuer, extracted_name = account_path.split(":", 1) + account_name = extracted_name + else: + account_name = account_path + + params = parse_qs(parsed.query) + secret = params.get("secret", [""])[0] + issuer = params.get("issuer", [""])[0] + algorithm = params.get("algorithm", ["SHA1"])[0] + digits = int(params.get("digits", ["6"])[0]) + period = int(params.get("period", ["30"])[0]) + + if not secret: + return [Result( + title="No secret in URI", + subtitle="otpauth URI must contain a secret parameter", + icon_markup=icons.cancel, + action=lambda: None, + relevance=0.5, + plugin_name=self.display_name, + data={"type": "error", "keep_launcher_open": True} + )] + + self.secrets[account_name] = { + "secret": secret, + "issuer": issuer, + "algorithm": algorithm, + "digits": digits, + "period": period + } + self._save_secrets() + + display_name = f"{issuer} - {account_name}" if issuer else account_name + return [Result( + title=f"✓ Added '{display_name}'", + subtitle="OTP account added from URI", + icon_markup=icons.check, + action=lambda: self._trigger_refresh(), + relevance=1.0, + plugin_name=self.display_name, + data={"type": "success", "keep_launcher_open": True} + )] + except Exception as e: + return [Result( + title="Error parsing otpauth URI", + subtitle=str(e), + icon_markup=icons.cancel, + action=lambda: None, + relevance=0.5, + plugin_name=self.display_name, + data={"type": "error", "keep_launcher_open": True} + )] \ No newline at end of file diff --git a/modules/launcher/plugins/power.py b/modules/launcher/plugins/power.py index 17e04c2..29ae53b 100644 --- a/modules/launcher/plugins/power.py +++ b/modules/launcher/plugins/power.py @@ -58,24 +58,36 @@ class PowerPlugin(PluginBase): def query(self, query_string: str) -> List[Result]: """Search power commands based on query.""" - if not query_string.strip(): - return [] - query = query_string.lower().strip() results = [] - for cmd, info in self.commands.items(): - if query in cmd.lower() or query in info["description"].lower(): + # If no query, show all power commands + if not query: + for cmd, info in self.commands.items(): result = Result( title=cmd.capitalize(), subtitle=info["description"], icon_markup=info["icon"], action=info["action"], - relevance=1.0 if query == cmd else 0.7, + relevance=1.0, plugin_name=self.display_name, data={"command": cmd}, ) results.append(result) + else: + # Filter commands based on query + for cmd, info in self.commands.items(): + if query in cmd.lower() or query in info["description"].lower(): + result = Result( + title=cmd.capitalize(), + subtitle=info["description"], + icon_markup=info["icon"], + action=info["action"], + relevance=1.0 if query == cmd else 0.7, + plugin_name=self.display_name, + data={"command": cmd}, + ) + results.append(result) return results diff --git a/modules/launcher/plugins/scripts.py b/modules/launcher/plugins/scripts.py new file mode 100644 index 0000000..4c9cc59 --- /dev/null +++ b/modules/launcher/plugins/scripts.py @@ -0,0 +1,187 @@ +""" +Scripts plugin for the launcher. +Provides access to various system scripts and utilities. +""" + +import os +import subprocess +from typing import List + +from fabric.utils import exec_shell_command_async +import utils.icons as icons +from modules.launcher.plugin_base import PluginBase +from modules.launcher.result import Result + + +class ScriptsPlugin(PluginBase): + """ + Plugin for executing system scripts and utilities. + """ + + def __init__(self): + super().__init__() + self.display_name = "Scripts" + self.description = "System scripts and utilities" + + def initialize(self): + """Initialize the scripts plugin.""" + # Use individual trigger keywords for each specific action + self.set_triggers([ + # Color picker variants + "color hex", + "color rgb", + "color hsv", + + # Night light + "night" + + ]) + + def cleanup(self): + """Cleanup the scripts plugin.""" + pass + + + def _run_colorpicker(self, format_type: str): + """Run color picker with specified format.""" + script_path = self._get_script_path("hyprpicker.sh") + exec_shell_command_async(f"bash '{script_path}' -{format_type}") + + + + def _get_script_path(self, script_name: str) -> str: + """Get the full path to a script from the scripts directory.""" + # Get the current working directory (should be Modus root) + current_dir = os.getcwd() + + # Try scripts directory relative to current working directory + script_path = os.path.join(current_dir, "scripts", script_name) + if os.path.exists(script_path): + return script_path + + # Try relative to this file's location + plugin_dir = os.path.dirname(os.path.abspath(__file__)) + modus_root = os.path.dirname(os.path.dirname(os.path.dirname(plugin_dir))) + script_path = os.path.join(modus_root, "scripts", script_name) + if os.path.exists(script_path): + return script_path + + # Fallback to relative path + return f"scripts/{script_name}" + + def _get_trigger_actions(self) -> dict: + """Get the map of trigger patterns to actions.""" + + return { + # Color picker triggers + "color hex": { + "title": "Color Picker (HEX)", + "subtitle": "Pick a color and copy HEX value to clipboard", + "icon": icons.colorpicker, + "action": lambda: self._run_colorpicker("hex"), + }, + "color rgb": { + "title": "Color Picker (RGB)", + "subtitle": "Pick a color and copy RGB value to clipboard", + "icon": icons.colorpicker, + "action": lambda: self._run_colorpicker("rgb"), + }, + "color hsv": { + "title": "Color Picker (HSV)", + "subtitle": "Pick a color and copy HSV value to clipboard", + "icon": icons.colorpicker, + "action": lambda: self._run_colorpicker("hsv"), + }, + + # Night light triggers + "night": { + "title": "Toggle Night Light", + "subtitle": "Toggle blue light filter for eye comfort", + "icon": icons.night, + "action": lambda: self._run_system_command("hyprctl keyword decoration:screen_shader ~/.config/hypr/shaders/blue_light_filter.glsl"), + } + } + + def query(self, query_string: str) -> List[Result]: + """Search for script actions based on query.""" + query = query_string.strip().lower() + results = [] + + # Get trigger actions + trigger_actions = self._get_trigger_actions() + + # If no query, return empty results (don't show all actions) + if not query: + return [] + + # Check if query exactly matches one of our triggers + # If so, show only that specific result + for trigger_key in trigger_actions: + if query == trigger_key.lower(): + action_data = trigger_actions[trigger_key] + return [Result( + title=action_data["title"], + subtitle=action_data["subtitle"], + icon_markup=action_data["icon"], + action=action_data["action"], + relevance=1.0, + plugin_name=self.display_name, + data={"trigger": trigger_key} + )] + + # Search through trigger actions for partial matches + for trigger, action_data in trigger_actions.items(): + # Check if query matches trigger or is contained in title + if (query in trigger.lower() or + query in action_data["title"].lower() or + any(word in trigger.lower() for word in query.split())): + + # Calculate relevance based on match quality + if query == trigger.lower(): + relevance = 1.0 + elif trigger.lower().startswith(query): + relevance = 0.9 + elif query in trigger.lower(): + relevance = 0.8 + else: + relevance = 0.7 + + results.append(Result( + title=action_data["title"], + subtitle=action_data["subtitle"], + icon_markup=action_data["icon"], + action=action_data["action"], + relevance=relevance, + plugin_name=self.display_name, + data={"trigger": trigger} + )) + + return sorted(results, key=lambda x: x.relevance, reverse=True) + + def query_triggered(self, _query_string: str, trigger: str) -> List[Result]: + """ + Process a triggered query (when plugin is activated by a specific trigger). + Show only the specific result for the triggered keyword. + """ + # Get trigger actions + trigger_actions = self._get_trigger_actions() + + # Find the exact trigger match and return only that result + trigger_lower = trigger.lower().strip() + + if trigger_lower in trigger_actions: + action_data = trigger_actions[trigger_lower] + return [Result( + title=action_data["title"], + subtitle=action_data["subtitle"], + icon_markup=action_data["icon"], + action=action_data["action"], + relevance=1.0, + plugin_name=self.display_name, + data={"trigger": trigger_lower} + )] + + # If no exact match found, return empty list + return [] + + diff --git a/modules/launcher/plugins/wallpaper.py b/modules/launcher/plugins/wallpaper.py index 2c3bb63..ad6b78c 100644 --- a/modules/launcher/plugins/wallpaper.py +++ b/modules/launcher/plugins/wallpaper.py @@ -9,14 +9,16 @@ import hashlib import colorsys import re import json -from typing import List +from typing import List, Dict, Optional from PIL import Image -from gi.repository import GdkPixbuf +from gi.repository import GdkPixbuf, GLib from modules.launcher.plugin_base import PluginBase from modules.launcher.result import Result from fabric.utils.helpers import exec_shell_command_async import config.data as data import utils.icons as icons +import threading +import time class WallpaperPlugin(PluginBase): @@ -29,6 +31,8 @@ class WallpaperPlugin(PluginBase): self.display_name = "Wallpaper" self.wallpapers = [] self.cache_dir = f"{data.CACHE_DIR}/thumbs" + self.thumbnail_cache: Dict[str, Optional[GdkPixbuf.Pixbuf]] = {} + self.thumbnail_loading = set() # Track which thumbnails are being loaded self.schemes = { "scheme-tonal-spot": "Tonal Spot", "scheme-content": "Content", @@ -45,6 +49,8 @@ class WallpaperPlugin(PluginBase): self.set_triggers(["wall", "wall "]) self._load_wallpapers() os.makedirs(self.cache_dir, exist_ok=True) + # Start background thumbnail creation + self._start_background_thumbnail_creation() def cleanup(self): """Cleanup the wallpaper plugin.""" @@ -127,15 +133,96 @@ class WallpaperPlugin(PluginBase): if not os.path.exists(cache_path): try: with Image.open(full_path) as img: - # Use faster thumbnail creation - img.thumbnail((48, 48), Image.Resampling.LANCZOS) - img.save(cache_path, "PNG") + # Use faster thumbnail creation with smaller size for better performance + img.thumbnail((32, 32), Image.Resampling.LANCZOS) + img.save(cache_path, "PNG", optimize=True) except Exception as e: print(f"Error creating thumbnail for {filename}: {e}") return None return cache_path + def _start_background_thumbnail_creation(self): + """Start background thread to create thumbnails for all wallpapers.""" + def create_thumbnails(): + for wallpaper in self.wallpapers: + if wallpaper not in self.thumbnail_loading: + self.thumbnail_loading.add(wallpaper) + try: + cache_path = self._create_thumbnail(wallpaper) + if cache_path and os.path.exists(cache_path): + # Load thumbnail into memory cache + try: + pixbuf = GdkPixbuf.Pixbuf.new_from_file_at_scale( + cache_path, 32, 32, True + ) + self.thumbnail_cache[wallpaper] = pixbuf + except Exception as e: + print(f"Error loading thumbnail for {wallpaper}: {e}") + self.thumbnail_cache[wallpaper] = None + else: + self.thumbnail_cache[wallpaper] = None + except Exception as e: + print(f"Error processing thumbnail for {wallpaper}: {e}") + self.thumbnail_cache[wallpaper] = None + finally: + self.thumbnail_loading.discard(wallpaper) + + # Small delay to prevent overwhelming the system + time.sleep(0.01) + + # Start background thread + thread = threading.Thread(target=create_thumbnails, daemon=True) + thread.start() + + def _get_thumbnail_fast(self, filename: str) -> Optional[GdkPixbuf.Pixbuf]: + """Get thumbnail quickly from cache or return None if not ready.""" + # Return cached thumbnail if available + if filename in self.thumbnail_cache: + return self.thumbnail_cache[filename] + + # Check if thumbnail file exists and load it immediately + cache_path = self._get_cache_path(filename) + if os.path.exists(cache_path): + try: + pixbuf = GdkPixbuf.Pixbuf.new_from_file_at_scale( + cache_path, 32, 32, True + ) + self.thumbnail_cache[filename] = pixbuf + return pixbuf + except Exception as e: + print(f"Error loading thumbnail for {filename}: {e}") + self.thumbnail_cache[filename] = None + return None + + # If not in cache and file doesn't exist, trigger background creation + if filename not in self.thumbnail_loading: + self.thumbnail_loading.add(filename) + def create_async(): + try: + cache_path = self._create_thumbnail(filename) + if cache_path and os.path.exists(cache_path): + try: + pixbuf = GdkPixbuf.Pixbuf.new_from_file_at_scale( + cache_path, 32, 32, True + ) + self.thumbnail_cache[filename] = pixbuf + except Exception as e: + print(f"Error loading thumbnail for {filename}: {e}") + self.thumbnail_cache[filename] = None + else: + self.thumbnail_cache[filename] = None + except Exception as e: + print(f"Error creating thumbnail for {filename}: {e}") + self.thumbnail_cache[filename] = None + finally: + self.thumbnail_loading.discard(filename) + + thread = threading.Thread(target=create_async, daemon=True) + thread.start() + + return None + def _set_wallpaper(self, filename: str, scheme: str = None): """Set wallpaper with optional matugen integration.""" full_path = os.path.join(data.WALLPAPERS_DIR, filename) @@ -749,22 +836,16 @@ class WallpaperPlugin(PluginBase): relevance = 0.8 matching_wallpapers.append((wallpaper, relevance)) - # Sort by relevance - no limit since we use bypass_max_results + # Sort by relevance and limit results for better performance matching_wallpapers.sort(key=lambda x: x[1], reverse=True) - for wallpaper, relevance in matching_wallpapers: - # Only create thumbnails for displayed results - cache_path = self._get_cache_path(wallpaper) - icon = None + # Limit to first 50 results for performance (can be increased if needed) + max_results = 50 if query else 20 # Show fewer when no query to load faster + matching_wallpapers = matching_wallpapers[:max_results] - # Try to load existing thumbnail first - if os.path.exists(cache_path): - try: - icon = GdkPixbuf.Pixbuf.new_from_file_at_scale( - cache_path, 32, 32, True - ) - except Exception as e: - print(f"Error loading existing thumbnail: {e}") + for wallpaper, relevance in matching_wallpapers: + # Use fast thumbnail loading + icon = self._get_thumbnail_fast(wallpaper) results.append( Result( diff --git a/modules/launcher/result.py b/modules/launcher/result.py index 79f7e08..26db630 100644 --- a/modules/launcher/result.py +++ b/modules/launcher/result.py @@ -17,6 +17,7 @@ class Result: # Display information title: str subtitle: str = "" + subtitle_markup: Optional[str] = None # Pango markup for subtitle description: str = "" icon: Optional[GdkPixbuf.Pixbuf] = None icon_name: Optional[str] = None diff --git a/modules/launcher/result_item.py b/modules/launcher/result_item.py index 149c416..815f85e 100644 --- a/modules/launcher/result_item.py +++ b/modules/launcher/result_item.py @@ -95,14 +95,25 @@ class ResultItem(EventBox): text_box.add(title_label) # Subtitle (if present) - if self.result.subtitle: - subtitle_label = Label( - label=self.result.subtitle, - name="result-item-subtitle", - h_align="start", - v_align="center", - ellipsize="end", - ) + if self.result.subtitle or self.result.subtitle_markup: + if self.result.subtitle_markup: + # Use markup for subtitle (supports Pango markup) + subtitle_label = Label( + markup=self.result.subtitle_markup, + name="result-item-subtitle", + h_align="start", + v_align="center", + ellipsize="end", + ) + else: + # Use plain text for subtitle + subtitle_label = Label( + label=self.result.subtitle, + name="result-item-subtitle", + h_align="start", + v_align="center", + ellipsize="end", + ) text_box.add(subtitle_label) # Plugin name (small text) diff --git a/utils/icons.py b/utils/icons.py index 7b0e8a8..13a9b15 100644 --- a/utils/icons.py +++ b/utils/icons.py @@ -183,6 +183,14 @@ calendar: str = "" connect: str = "" disconnect: str = "ﴖ" +# otp +check: str = "" +bulb: str = "" +key: str = "" +magnifier: str = "" +plus: str = "" +info: str = "" + # Dice dice_1: str = "" dice_2: str = ""