diff --git a/src/window/launcher/plugins/__init__.py b/src/window/launcher/plugins/__init__.py
deleted file mode 100644
index 666274b..0000000
--- a/src/window/launcher/plugins/__init__.py
+++ /dev/null
@@ -1,3 +0,0 @@
-"""
-Built-in plugins for the launcher.
-"""
diff --git a/src/window/launcher/plugins/applications.py b/src/window/launcher/plugins/applications.py
deleted file mode 100644
index 56a0ab9..0000000
--- a/src/window/launcher/plugins/applications.py
+++ /dev/null
@@ -1,192 +0,0 @@
-from fabric.utils import logger
-import json
-from typing import List
-
-from fabric.utils import (
- DesktopApp,
- get_desktop_applications,
- get_relative_path,
- re,
- exec_shell_command_async,
-)
-
-from utils.roam import modus_service
-from window.launcher.plugin_base import PluginBase
-from window.launcher.result import Result
-
-
-class ApplicationsPlugin(PluginBase):
- def __init__(self):
- super().__init__()
- self.display_name = "Applications"
- self.description = "Search and launch desktop applications"
-
- def initialize(self):
- pass
-
- def cleanup(self):
- pass
-
- def _pin_application(self, app):
- """Pin an application to the dock."""
- config_path = get_relative_path("../../../../config/dock.json")
- try:
- with open(config_path, "r") as file:
- pinned_apps = json.load(file)
- except (FileNotFoundError, json.JSONDecodeError):
- pinned_apps = []
-
- # Handle legacy format (dict with "pinned_apps" key) and convert to new format (simple list)
- if isinstance(pinned_apps, dict):
- pinned_apps = pinned_apps.get("pinned_apps", [])
-
- # Check if app is already pinned (by name/app_id)
- app_id = app.name # Use app.name as the identifier
- if app_id not in pinned_apps:
- pinned_apps.append(app_id)
-
- # Save the updated list
- with open(config_path, "w") as file:
- json.dump(pinned_apps, file, indent=4)
-
- # Notify dock about the change via modus_service
- if modus_service:
- try:
- dock_apps_json = json.dumps(pinned_apps)
- modus_service.dock_apps = dock_apps_json
- except Exception as e:
- logger.error(f"Failed to notify dock about pinned app change: {e}")
-
- def query(self, query_string: str) -> List[Result]:
- """Search applications based on query."""
- if not query_string.strip():
- return self._get_all_applications()
-
- try:
- applications = get_desktop_applications(include_hidden=False)
- except Exception as e:
- logger.error(f"Failed to load applications: {e}")
- applications = []
-
- query = query_string.lower().strip()
- results = []
-
- for app in applications:
- relevance = self._calculate_relevance(app, query)
- if relevance > 0:
- description = app.description or app.generic_name or ""
- if len(description) > 80:
- description = description[:70] + "..."
-
- result = Result(
- title=app.display_name or app.name,
- subtitle=description,
- icon=app.get_icon_pixbuf(size=48),
- action=lambda a=app: self._launch_application(a),
- relevance=relevance,
- plugin_name=self.display_name,
- data={
- "app": app,
- "pin_action": lambda a=app: self._pin_application(a),
- },
- )
- results.append(result)
-
- return results
-
- def _calculate_relevance(self, app, query: str) -> float:
- """Calculate relevance score for an application."""
- if not query:
- return 0.0
-
- # Get searchable text
- name = (app.name or "").lower()
- display_name = (app.display_name or "").lower()
- description = (app.description or "").lower()
- generic_name = (app.generic_name or "").lower()
- executable = (app.executable or "").lower()
-
- # Exact matches get highest score
- if query == name or query == display_name:
- return 1.0
-
- # Starts with matches get high score
- if (
- name.startswith(query)
- or display_name.startswith(query)
- or generic_name.startswith(query)
- ):
- return 0.9
-
- # Contains matches get medium score
- if (
- query in name
- or query in display_name
- or query in description
- or query in generic_name
- ):
- return 0.7
-
- # Fuzzy matching for partial matches
- if self._fuzzy_match(query, name) or self._fuzzy_match(query, display_name):
- return 0.5
-
- # Executable name matching
- if executable and query in executable:
- return 0.4
-
- return 0.0
-
- def _fuzzy_match(self, query: str, text: str) -> bool:
- """Simple fuzzy matching algorithm."""
- if not query or not text:
- return False
-
- # Create regex pattern for fuzzy matching
- pattern = ".*".join(re.escape(char) for char in query)
- return bool(re.search(pattern, text, re.IGNORECASE))
-
- def _launch_application(self, app: DesktopApp):
-
- # Remove ALL % codes (e.g., %u, %U, %f, %F, %i, %c, etc.)
- cleaned_command = re.sub(r"%\w+", "", app.command_line).strip()
-
- # Final command with hyprctl dispatch
- final_command = (
- f"hyprctl dispatch 'hl.dsp.exec_cmd([[uwsm app -- {cleaned_command}]])'"
- )
- exec_shell_command_async(final_command)
-
- # app.launch()
-
- def _get_all_applications(self) -> List[Result]:
- """Get a list of all available applications."""
- try:
- applications = get_desktop_applications(include_hidden=False)
- except Exception as e:
- logger.error(f"Failed to load applications: {e}")
- return []
-
- results = []
-
- for app in applications:
- # Truncate description
- description = app.description or app.generic_name or ""
- if len(description) > 80:
- description = description[:70] + "..."
-
- result = Result(
- title=app.display_name or app.name,
- subtitle=description,
- icon=app.get_icon_pixbuf(size=48),
- action=lambda a=app: self._launch_application(a),
- relevance=0.5, # Default relevance for all apps
- plugin_name=self.display_name,
- data={
- "app": app,
- "pin_action": lambda a=app: self._pin_application(a),
- },
- )
- results.append(result)
-
- return results
diff --git a/src/window/launcher/plugins/caffeine.py b/src/window/launcher/plugins/caffeine.py
deleted file mode 100644
index f62adb7..0000000
--- a/src/window/launcher/plugins/caffeine.py
+++ /dev/null
@@ -1,205 +0,0 @@
-from fabric.utils import logger
-import subprocess
-from threading import Timer
-from typing import List
-
-from fabric.utils import exec_shell_command_async, get_relative_path
-
-import shared.data as data
-from window.launcher.plugin_base import PluginBase
-from window.launcher.result import Result
-
-
-class CaffeinePlugin(PluginBase):
- """
- Plugin for preventing system idle using the inhibit script.
- """
-
- def __init__(self):
- super().__init__()
- self.display_name = "Caffeine"
- self.description = "Prevent system from going idle"
- self.inhibit_script = get_relative_path("../../../utils/inhibit.py")
-
- # Predefined durations
- self.durations = {
- "30m": "30 minutes",
- "1h": "1 hour",
- "2h": "2 hours",
- "4h": "4 hours",
- "8h": "8 hours",
- "on": "On",
- "off": "Off",
- }
-
- def initialize(self):
- """Initialize the caffeine plugin."""
- self.set_triggers(["caffeine"])
-
- def cleanup(self):
- """Cleanup the caffeine plugin."""
- pass
-
- def _create_inhibit_action(self, duration: str):
- """Create an inhibit action that properly captures the duration."""
-
- def action():
- try:
- # Run the script in the background without waiting
- subprocess.Popen(
- ["python3", self.inhibit_script, duration],
- stdout=subprocess.DEVNULL,
- 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:
- logger.error(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:
- return True
- if query.isdigit():
- return True
- if query.endswith(("h", "m", "s")) and query[:-1].replace(".", "").isdigit():
- return True
- return False
-
- def _get_default_action(self, query: str):
- """Get the default action for a direct duration input."""
- if self._is_valid_duration(query):
- return self._create_inhibit_action(query)
- return None
-
- def query(self, query_string: str) -> List[Result]:
- """Search caffeine durations based on query."""
- # For empty queries, show all available durations
- if not query_string.strip():
- query_string = "" # Will match all durations in the loop below
-
- query = query_string.lower().strip()
- results = []
-
- # Handle direct search entry (e.g., "caffeine 30m" or just "30m")
- if query.startswith("caffeine "):
- query = query[9:].strip() # Remove "caffeine " prefix
- # If query becomes empty after removing prefix, show all durations
- if not query:
- query = "" # Will match all durations in the loop below
- elif query == "caffeine":
- # Handle just "caffeine" without space - show all durations
- query = ""
- elif not query:
- # Handle empty query - show all durations
- pass
- elif (
- query in self.durations
- or query.isdigit()
- or (
- query.endswith(("h", "m", "s"))
- and query[:-1].replace(".", "").isdigit()
- )
- ):
- # If query is a valid duration without prefix, use it directly
- pass
- else:
- return []
-
- # Add custom duration option
- if query.isdigit() or (
- query.endswith(("h", "m", "s")) and query[:-1].replace(".", "").isdigit()
- ):
- result = Result(
- title=f"Custom: {query}",
- subtitle="Set custom duration",
- icon_name="caffeine",
- action=self._create_inhibit_action(query),
- relevance=1.0,
- plugin_name=self.display_name,
- data={"duration": query},
- )
- results.append(result)
-
- # Add predefined durations
- for duration, description in self.durations.items():
- # If query is empty, show all durations; otherwise filter by query
- if not query or query in duration or query in description.lower():
- # Special handling for "off" - it should stop inhibition
- if duration.lower() == "off":
- subtitle = "Stop idle inhibition"
- else:
- subtitle = f"Prevent idle for {description.lower()}"
-
- result = Result(
- title=description,
- subtitle=subtitle,
- icon_name="caffeine",
- action=self._create_inhibit_action(duration),
- relevance=0.9 if query == duration else 0.7,
- plugin_name=self.display_name,
- data={"duration": duration},
- )
- results.append(result)
-
- # Set default action for direct duration input
- if results:
- results[0].default_action = self._get_default_action(query)
-
- return results
diff --git a/src/window/launcher/plugins/calculator.py b/src/window/launcher/plugins/calculator.py
deleted file mode 100644
index ce4b175..0000000
--- a/src/window/launcher/plugins/calculator.py
+++ /dev/null
@@ -1,168 +0,0 @@
-from fabric.utils import logger
-import math
-import re
-import subprocess
-import time
-from typing import List
-
-from utils.conversion import Conversion
-from window.launcher.plugin_base import PluginBase
-from window.launcher.result import Result
-
-
-class CalculatorPlugin(PluginBase):
- """
- Plugin for calculating mathematical expressions and converting units.
- """
-
- def __init__(self):
- super().__init__()
- self.display_name = "Calculator"
- self.description = "Evaluate mathematical expressions and convert units"
-
- # Safe functions for evaluation
- self.safe_functions = {
- "abs": abs,
- "round": round,
- "min": min,
- "max": max,
- "sum": sum,
- "pow": pow,
- "sqrt": math.sqrt,
- "sin": math.sin,
- "cos": math.cos,
- "tan": math.tan,
- "asin": math.asin,
- "acos": math.acos,
- "atan": math.atan,
- "log": math.log,
- "log10": math.log10,
- "exp": math.exp,
- "pi": math.pi,
- "e": math.e,
- }
-
- # Initialize conversion utility
- self.converter = Conversion()
-
- # Pre-compiled regex patterns
- self.expression_pattern = re.compile(r"[\d+\-*/^()=]")
- self.number_pattern = re.compile(r"\d")
- self.conversion_pattern = re.compile(
- r"(\d+(?:\.\d+)?)\s*([^0-9\s]+)\s*(?:to|in|=)\s*([^0-9\s]+)"
- )
-
- # Cache for conversion results
- self._conversion_cache = {}
- self._last_cache_cleanup = time.time()
- self._cache_cleanup_interval = 300 # 5 minutes
-
- def initialize(self):
- """Initialize the files plugin."""
- self.set_triggers(["="])
-
- def _cleanup_cache(self):
- """Clean up old cache entries."""
- current_time = time.time()
- if current_time - self._last_cache_cleanup > self._cache_cleanup_interval:
- self._conversion_cache.clear()
- self._last_cache_cleanup = current_time
-
- def query(self, query: str) -> List[Result]:
- """Process a query and return results."""
- if not query:
- return []
-
- # Clean up cache periodically
- self._cleanup_cache()
-
- # Check if it's a conversion query
- conversion_match = self.conversion_pattern.match(query)
- if conversion_match:
- try:
- value, from_unit, to_unit = conversion_match.groups()
- value = float(value)
-
- # Check cache first
- cache_key = f"{value}_{from_unit}_{to_unit}"
- if cache_key in self._conversion_cache:
- result = self._conversion_cache[cache_key]
- subtitle = f"{value} {from_unit} = {result:.6g} {to_unit}"
- else:
- # Use the conversion utility
- from_unit_clean = self.converter.clean_type(from_unit)
- to_unit_clean = self.converter.clean_type(to_unit)
- result = self.converter.convert(
- value, from_unit_clean, to_unit_clean
- )
- # Cache the result
- self._conversion_cache[cache_key] = result
- subtitle = f"{value} {from_unit} = {result:.6g} {to_unit}"
-
- return [
- Result(
- title=f"{result:.6g} {to_unit}",
- subtitle=subtitle,
- icon_name="calculator-symbolic",
- action=lambda r=f"{result:.6g}": self._copy_to_clipboard(r),
- relevance=1.0,
- plugin_name=self.display_name,
- data={"from": (value, from_unit), "to": (result, to_unit)},
- )
- ]
- except ValueError as e:
- return [
- Result(
- title="Invalid conversion",
- subtitle=str(e),
- icon_name="calculator-symbolic",
- relevance=0.0,
- plugin_name=self.display_name,
- )
- ]
-
- # Check if it's a math expression
- if self.expression_pattern.search(query):
- try:
- # Evaluate the expression
- result = eval(query, {"__builtins__": {}}, self.safe_functions)
- if isinstance(result, (int, float)):
- return [
- Result(
- title=f"{result:.6g}",
- subtitle=f"{query} = {result:.6g}",
- icon_name="calculator-symbolic",
- action=lambda r=f"{result:.6g}": self._copy_to_clipboard(r),
- relevance=1.0,
- plugin_name=self.display_name,
- )
- ]
- except Exception as e:
- logger.error(f"An error occurred: {e}")
-
- return []
-
- def _format_cache_age(self, age_seconds: float) -> str:
- """Format cache age for display."""
- if age_seconds < 60:
- return f"{int(age_seconds)}s ago"
- elif age_seconds < 3600:
- return f"{int(age_seconds // 60)}m ago"
- else:
- return f"{int(age_seconds // 3600)}h ago"
-
- def _copy_to_clipboard(self, text: str):
- """Copy text to clipboard using cliphist."""
- try:
- # First copy to clipboard
- subprocess.run(["wl-copy"], input=text.encode(), check=True)
- # Then store in cliphist
- subprocess.run(["cliphist", "store"], input=text.encode(), check=True)
- except subprocess.CalledProcessError:
- # If cliphist fails, at least we have the text in clipboard
- pass
-
- def cleanup(self):
- """Clean up resources."""
- self._conversion_cache.clear()
- self.converter.cleanup()
diff --git a/src/window/launcher/plugins/clipboard.py b/src/window/launcher/plugins/clipboard.py
deleted file mode 100644
index 9393ea4..0000000
--- a/src/window/launcher/plugins/clipboard.py
+++ /dev/null
@@ -1,435 +0,0 @@
-from fabric.utils import logger
-import subprocess
-import tempfile
-import threading
-from concurrent.futures import ThreadPoolExecutor
-from typing import Dict, List, Optional
-
-from fabric.utils import GdkPixbuf, GLib, os, time
-
-from window.launcher.plugin_base import PluginBase
-from window.launcher.result import Result
-
-
-class ClipboardPlugin(PluginBase):
- def __init__(self):
- super().__init__()
- self.name = "clipboard"
- self.display_name = "Clipboard History"
- self.description = "Search and manage clipboard history using cliphist"
-
- # Performance settings - show more history like example_cliphist.py
- self.max_results = 50
- # Cache clipboard items for 5 seconds (more responsive)
- self.cache_ttl = 5
-
- # Initialize cache and temp directory
- self.tmp_dir = tempfile.mkdtemp(prefix="cliphist-")
- # Cache images forever like example_cliphist.py
- self.image_cache: Dict[str, GdkPixbuf.Pixbuf] = {}
- self.clipboard_items_cache: List[str] = []
- self.cache_timestamp = 0
-
- # 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):
- """Initialize the plugin."""
- self.set_triggers(["clip"])
- try:
- 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)
-
- # Clear caches
- with self.cache_lock:
- self.image_cache.clear()
- self.clipboard_items_cache.clear()
- except Exception as e:
- logger.error(f"Error cleaning up temporary files: {e}")
-
- def invalidate_cache(self):
- """Force invalidation of the clipboard cache."""
- with self.cache_lock:
- self.clipboard_items_cache.clear()
- self.cache_timestamp = 0
- # Keep image cache forever like example_cliphist.py
-
- def _force_launcher_refresh(self):
- """Force the launcher to refresh its results."""
- try:
-
- 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")
- and hasattr(launcher, "_perform_search")
- ):
- # Get current search text to preserve the query
- current_text = launcher.search_entry.get_text()
- # Trigger the search to refresh results
- launcher._perform_search(current_text)
- 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"
- ):
- current_text = obj.search_entry.get_text()
- obj._perform_search(current_text)
- return False
-
- except Exception as e:
- logger.error(f"Error forcing launcher refresh: {e}")
-
- return False # Don't repeat
-
- # Use immediate refresh
- GLib.timeout_add(10, trigger_refresh)
-
- except Exception as e:
- logger.error(f"Could not trigger refresh: {e}")
-
- def _load_clipboard_items_cached(self) -> List[str]:
- """Load clipboard items from cliphist with caching and change detection."""
- current_time = time.time()
-
- # Always load fresh data to check for changes
- try:
- result = subprocess.run(
- ["cliphist", "list"], capture_output=True, check=True, timeout=5
- )
- stdout_str = result.stdout.decode("utf-8", errors="replace")
- if stdout_str.strip():
- lines = stdout_str.strip().split("\n")
- items = [
- 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:
- """Create a GdkPixbuf from image bytes with size limit."""
- try:
- loader = GdkPixbuf.PixbufLoader()
- loader.write(image_data)
- loader.close()
- pixbuf = loader.get_pixbuf()
-
- # Scale image if needed
- width, height = pixbuf.get_width(), pixbuf.get_height()
- if width > height:
- new_width = max_size
- new_height = int(height * (max_size / width))
- else:
- new_height = max_size
- new_width = int(width * (max_size / height))
-
- return pixbuf.scale_simple(
- new_width, new_height, GdkPixbuf.InterpType.BILINEAR
- )
- except GLib.Error:
- return None
-
- def _is_image_data(self, content: str) -> bool:
- """Determine if clipboard content is likely an image (like example_cliphist.py)."""
- return (
- content.startswith("data:image/")
- or content.startswith("\x89PNG")
- or content.startswith("GIF8")
- or content.startswith("\xff\xd8\xff")
- or "binary" in content.lower()
- and any(
- ext in content.lower() for ext in ["jpg", "jpeg", "png", "bmp", "gif"]
- )
- )
-
- def _get_text_preview(self, content: str) -> str:
- """Get a text preview of the content."""
- if len(content) > 50:
- return content[:37] + "..."
- return content
-
- def _load_image_preview_cached(self, item_id: str) -> Optional[GdkPixbuf.Pixbuf]:
- """Load image preview with forever caching (like example_cliphist.py)."""
- # Check cache first - cache forever like example_cliphist.py
- with self.cache_lock:
- if item_id in self.image_cache:
- return self.image_cache[item_id]
-
- try:
- result = subprocess.run(
- ["cliphist", "decode", item_id],
- capture_output=True,
- check=True,
- timeout=3,
- )
- pixbuf = self._create_pixbuf_from_bytes(result.stdout)
- if pixbuf:
- with self.cache_lock:
- self.image_cache[item_id] = pixbuf
-
- return pixbuf
- except Exception as e:
- logger.error(f"Error loading image preview: {e}")
- 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 with optimized performance."""
- results = []
-
- # Handle query string
- if query_string.lower() == "clip":
- query_string = "" # Show all items
-
- try:
- # Load clipboard items from cache
- clipboard_items = self._load_clipboard_items_cached()
-
- # Early exit if no items
- if not clipboard_items:
- return results
-
- # Filter items based on query - search through ALL items first
- filtered_items = []
- query_lower = query_string.lower() if query_string else ""
-
- for item in clipboard_items:
- parts = item.split("\t", 1)
- content = parts[1] if len(parts) > 1 else item
- content_lower = content.lower()
-
- # Fast filtering - search through all items, don't limit here
- if not query_lower or query_lower in content_lower:
- # Calculate relevance score for better sorting
- relevance = 1.0
- if query_lower:
- if content_lower.startswith(query_lower):
- relevance = 1.0 # Exact start match
- elif query_lower in content_lower:
- # Position-based scoring: earlier matches get higher scores
- position = content_lower.find(query_lower)
- relevance = max(
- 0.5, 1.0 - (position / len(content_lower)) * 0.4
- )
-
- filtered_items.append((item, relevance))
-
- # Sort by relevance (highest first) and then limit results
- filtered_items.sort(key=lambda x: x[1], reverse=True)
- filtered_items = filtered_items[: self.max_results]
-
- # Process items with lazy image loading
- for i, (item, relevance) in enumerate(filtered_items):
- parts = item.split("\t", 1)
- item_id = parts[0] if len(parts) > 1 else str(i)
- content = parts[1] if len(parts) > 1 else item
-
- # Handle image content like example_cliphist.py
- if self._is_image_data(content):
- # Check if image is already cached (forever cache like example_cliphist.py)
- cached_pixbuf = None
- with self.cache_lock:
- cached_pixbuf = self.image_cache.get(item_id)
-
- if cached_pixbuf:
- # Use cached image immediately
- result = Result(
- title="Image from clipboard",
- subtitle="Click to copy image to clipboard",
- description="Image content",
- icon=cached_pixbuf,
- relevance=relevance,
- plugin_name=self.name,
- action=lambda id=item_id: self._copy_to_clipboard(id),
- data={"bypass_max_results": True},
- )
- else:
- # Try to load image immediately like example_cliphist.py
- try:
- immediate_pixbuf = self._load_image_preview_cached(item_id)
- if immediate_pixbuf:
- # Successfully loaded immediately
- result = Result(
- title="Image from clipboard",
- subtitle="Click to copy image to clipboard",
- description="Image content",
- icon=immediate_pixbuf,
- relevance=relevance,
- plugin_name=self.name,
- action=lambda id=item_id: self._copy_to_clipboard(
- id
- ),
- data={"bypass_max_results": True},
- )
- else:
- # Show placeholder if loading failed
- result = Result(
- title="Image from clipboard",
- subtitle="Click to copy image to clipboard",
- description="Image content",
- icon_name="image-x-generic",
- relevance=relevance,
- plugin_name=self.name,
- action=lambda id=item_id: self._copy_to_clipboard(
- id
- ),
- data={"bypass_max_results": True},
- )
- except Exception:
- # Fallback to placeholder
- result = Result(
- title="Image from clipboard",
- subtitle="Click to copy image to clipboard",
- description="Image content",
- icon_name="image-x-generic",
- relevance=relevance,
- plugin_name=self.name,
- action=lambda id=item_id: self._copy_to_clipboard(id),
- data={"bypass_max_results": True},
- )
-
- 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 if len(content) <= 100 else content[:97] + "..."
- ),
- icon_name="edit-paste",
- relevance=relevance,
- plugin_name=self.name,
- action=lambda id=item_id: self._copy_to_clipboard(id),
- data={"bypass_max_results": True},
- )
- results.append(result)
-
- except Exception as e:
- # Handle errors gracefully
- results.append(
- Result(
- title="Error accessing clipboard history",
- subtitle=str(e),
- icon_name="dialog-error",
- relevance=0.0,
- plugin_name=self.name,
- data={"bypass_max_results": True},
- )
- )
-
- return results
-
- def _copy_to_clipboard(self, entry_id: str):
- """Copy entry to clipboard using cliphist with timeout."""
- try:
- result = subprocess.run(
- ["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, timeout=3)
- except subprocess.SubprocessError:
- subprocess.run(
- ["xclip", "-selection", "clipboard"],
- input=result.stdout,
- check=True,
- timeout=3,
- )
-
- # Don't invalidate image cache - keep images cached forever like example_cliphist.py
- # Only invalidate clipboard items cache
- with self.cache_lock:
- self.clipboard_items_cache.clear()
- self.cache_timestamp = 0
-
- except subprocess.SubprocessError as e:
- logger.error(f"Error copying to clipboard: {e}")
- except subprocess.TimeoutExpired as e:
- logger.error(f"Timeout copying to clipboard: {e}")
diff --git a/src/window/launcher/plugins/colorpicker.py b/src/window/launcher/plugins/colorpicker.py
deleted file mode 100644
index 424ad9c..0000000
--- a/src/window/launcher/plugins/colorpicker.py
+++ /dev/null
@@ -1,146 +0,0 @@
-import os
-import re
-import subprocess
-from typing import List
-
-from fabric.utils import exec_shell_command, exec_shell_command_async, logger
-from shared.data import APP_NAME
-from window.launcher.plugin_base import PluginBase
-from window.launcher.result import Result
-
-
-class ColorPickerPlugin(PluginBase):
- """
- Plugin for picking colors from the screen using hyprpicker.
- """
-
- def __init__(self):
- super().__init__()
- self.display_name = "Color Picker"
- self.description = "Pick colors from the screen (HEX, RGB, HSV)"
- self.tmp_icon_path = "/tmp/color.png"
-
- self.options = [
- {"name": "Pick HEX", "arg": "hex", "icon": "color-select-symbolic"},
- {"name": "Pick RGB", "arg": "rgb", "icon": "color-select-symbolic"},
- {"name": "Pick HSV", "arg": "hsv", "icon": "color-select-symbolic"},
- ]
-
- def initialize(self):
- """Initialize the color picker plugin."""
- self.set_triggers(["color", "cp"])
-
- def cleanup(self):
- """Cleanup resources."""
- if os.path.exists(self.tmp_icon_path):
- try:
- os.remove(self.tmp_icon_path)
- except OSError:
- pass
-
- def query(self, query_string: str) -> List[Result]:
- """Process a search query and return results."""
- q = query_string.strip().lower()
-
- results = []
-
- # Filter options based on query
- filtered = (
- self.options
- if not q
- else [o for o in self.options if q in o["name"].lower() or q == o["arg"]]
- )
-
- for option in filtered:
- mode = option["arg"]
- results.append(
- Result(
- title=f"Color Picker: {option['name']}",
- subtitle=f"Pick and copy {mode.upper()} color to clipboard",
- icon_name=option["icon"],
- action=lambda m=mode: self._execute_pick(m),
- relevance=1.0,
- plugin_name=self.display_name,
- )
- )
-
- return results
-
- def _execute_pick(self, mode: str):
- """Run hyprpicker, process output, copy to clipboard and notify."""
- try:
- # Run hyprpicker
- # -n: do not print a newline
- # -f: format
- raw = exec_shell_command(f"hyprpicker -n -f {mode}")
- text = raw if isinstance(raw, str) else ""
-
- color = self._sanitize_color_output(mode, text)
- if not color:
- logger.warning(
- f"[ColorPicker] Failed to parse color from output: {text}"
- )
- return
-
- # Copy to clipboard
- self._copy_to_clipboard(color)
-
- # Generate preview and notify
- self._send_notification(mode, color)
-
- except Exception as e:
- logger.error(f"[ColorPicker] Error during pick: {e}")
-
- def _sanitize_color_output(self, mode: str, text: str) -> str:
- """Extract color value from hyprpicker output."""
- joined = " ".join(ln.strip() for ln in text.splitlines() if ln.strip())
-
- if mode == "hex":
- matches = re.findall(r"#?[0-9A-Fa-f]{6,8}", joined)
- if not matches:
- return ""
- val = matches[-1]
- return val if val.startswith("#") else f"#{val}"
-
- if mode in ("rgb", "hsv"):
- # Matches "r, g, b" or "h, s, v"
- matches = re.findall(
- r"\d+(?:\.\d+)?\s*,\s*\d+(?:\.\d+)?\s*,\s*\d+(?:\.\d+)?", joined
- )
- return matches[-1] if matches else ""
-
- return ""
-
- def _copy_to_clipboard(self, text: str):
- """Copy text to clipboard using wl-copy and cliphist."""
- try:
- subprocess.run(["wl-copy"], input=text.encode(), check=True)
- subprocess.run(["cliphist", "store"], input=text.encode(), check=True)
- except subprocess.CalledProcessError:
- pass
-
- def _send_notification(self, mode: str, color: str):
- """Create a color preview icon and send a notification."""
- # CSS-like color specification for ImageMagick
- color_spec = (
- color
- if mode == "hex"
- else (f"rgb({color})" if mode == "rgb" else f"hsv({color})")
- )
-
- # Cleanup existing tmp icon
- if os.path.exists(self.tmp_icon_path):
- try:
- os.remove(self.tmp_icon_path)
- except OSError:
- pass
-
- # Use magick to create a 64x64 color block
- # Then send notification and cleanup
- cmd = (
- f"magick -size 64x64 xc:'{color_spec}' {self.tmp_icon_path} >/dev/null 2>&1; "
- f"notify-send 'Color Picked' '{mode.upper()}: {color}' -i {self.tmp_icon_path} -a '{APP_NAME}' -e; "
- f"rm -f {self.tmp_icon_path}"
- )
-
- exec_shell_command_async(["bash", "-lc", cmd])
diff --git a/src/window/launcher/plugins/emoji.py b/src/window/launcher/plugins/emoji.py
deleted file mode 100644
index 307f9c3..0000000
--- a/src/window/launcher/plugins/emoji.py
+++ /dev/null
@@ -1,291 +0,0 @@
-import json
-import subprocess
-from collections import OrderedDict
-from typing import Dict, List
-
-from fabric.utils import os, time, GLib, logger
-
-import shared.data as data
-from window.launcher.plugin_base import PluginBase
-from window.launcher.result import Result
-
-
-class EmojiPlugin(PluginBase):
- """
- Plugin for searching and copying emojis.
- """
-
- def __init__(self):
- super().__init__()
- self.name = "emoji"
- self.display_name = "Emoji"
- self.description = "Search and copy emojis"
- self.emoji_data = {}
-
- # Language detection (default to en)
- self.lang = os.getenv("LANG", "en").split("_")[0].split(".")[0]
- if not self.lang:
- self.lang = "en"
-
- # Paths
- self.state_dir = os.path.join(GLib.get_user_state_dir(), data.APP_NAME)
- self.emoji_path = os.path.join(self.state_dir, f"emojis_{self.lang}.json")
-
- # Use cache directory for recent emojis
- self.recent_emoji_path = os.path.join(data.CACHE_DIR, "recent_emoji.json")
- self.recent_emojis = OrderedDict()
- self.max_recent_emojis = 20
- self._is_downloading = False
-
- def initialize(self):
- """Initialize the emoji plugin."""
- self.set_triggers(["em"])
- self._load_recent_emojis()
-
- # Check if local data exists, if not trigger download
- if not os.path.exists(self.emoji_path):
- self._download_emoji_data()
- else:
- self._load_emoji_data()
-
- def cleanup(self):
- """Cleanup the emoji plugin."""
- pass
-
- def _download_emoji_data(self, force=False):
- """Download emoji annotations from CLDR repository."""
- if self._is_downloading and not force:
- return
-
- os.makedirs(self.state_dir, exist_ok=True)
- url = (
- "https://raw.githubusercontent.com/unicode-org/cldr-json/main/"
- f"cldr-json/cldr-annotations-full/annotations/{self.lang}/annotations.json"
- )
-
- self._is_downloading = True
-
- def on_done(*args):
- self._is_downloading = False
- if os.path.exists(self.emoji_path):
- self._load_emoji_data()
- logger.info(f"[Emoji] Data updated for language: {self.lang}")
- else:
- logger.error(f"[Emoji] Download failed for URL: {url}")
-
- logger.info(f"[Emoji] Starting download: {url}")
- from fabric.utils import exec_shell_command_async
-
- exec_shell_command_async(f"curl -L '{url}' -o '{self.emoji_path}'", on_done)
-
- def _load_emoji_data(self):
- """Load emoji data from JSON file and parse CLDR structure."""
- try:
- if os.path.exists(self.emoji_path):
- with open(self.emoji_path, "r", encoding="utf-8") as f:
- raw_data = json.load(f)
- self.emoji_data = self._parse_cldr_json(raw_data)
- else:
- logger.warning(f"[Emoji] Local data missing: {self.emoji_path}")
- except Exception as e:
- logger.error(f"[Emoji] Error loading emoji data: {e}")
-
- def _parse_cldr_json(self, raw_data: Dict) -> Dict:
- """Parse CLDR annotations JSON into a flatter format for searching."""
- try:
- # Structure matches individual file: annotations -> annotations
- annotations_container = raw_data.get("annotations", {})
- raw_annotations = annotations_container.get("annotations", {})
-
- if not raw_annotations:
- # Fallback to the nested structure just in case it's a full bundle
- main = raw_data.get("main", {})
- lang_data = main.get(self.lang, main.get("en", {}))
- annotations_container = lang_data.get("annotations", {})
- raw_annotations = annotations_container.get("annotations", {})
-
- parsed = {}
- for emoji, info in raw_annotations.items():
- # Extract TTL (name)
- tts = info.get("tts", [""])
- name = tts[0] if isinstance(tts, list) else tts
-
- # Extract keywords
- keywords = info.get("default", [])
- if isinstance(keywords, str):
- keywords = [keywords]
-
- parsed[emoji] = {
- "name": name,
- "keywords": keywords,
- }
- return parsed
- except Exception as e:
- logger.error(f"[Emoji] CLDR parsing error: {e}")
- return {}
-
- def _load_recent_emojis(self):
- """Load recently used emojis from JSON file."""
- try:
- if os.path.exists(self.recent_emoji_path):
- with open(self.recent_emoji_path, "r", encoding="utf-8") as f:
- recent_data = json.load(f)
- # Convert to OrderedDict to maintain order
- self.recent_emojis = OrderedDict(recent_data)
- else:
- # Create empty recent emojis file
- self.recent_emojis = OrderedDict()
- self._save_recent_emojis()
- except Exception as e:
- logger.error(f"Error loading recent emoji data: {e}")
- self.recent_emojis = OrderedDict()
-
- def _save_recent_emojis(self):
- """Save recently used emojis to JSON file."""
- try:
- # Ensure the cache directory exists
- os.makedirs(data.CACHE_DIR, exist_ok=True)
-
- with open(self.recent_emoji_path, "w", encoding="utf-8") as f:
- json.dump(dict(self.recent_emojis), f, ensure_ascii=False, indent=2)
- except Exception as e:
- logger.error(f"Error saving recent emoji data: {e}")
-
- def _add_to_recent(self, emoji: str):
- """Add an emoji to the recent list."""
- # Remove if already exists (to move it to front)
- if emoji in self.recent_emojis:
- del self.recent_emojis[emoji]
-
- # Add to front with current timestamp
- self.recent_emojis[emoji] = time.time()
-
- # Keep only the most recent emojis
- while len(self.recent_emojis) > self.max_recent_emojis:
- # Remove the oldest item
- self.recent_emojis.popitem(last=False)
-
- # Save to file
- self._save_recent_emojis()
-
- def _copy_to_clipboard(self, emoji: str):
- """Copy emoji to clipboard and track usage."""
- try:
- # Try Wayland first
- try:
- subprocess.run(["wl-copy"], input=emoji.encode(), check=True)
- except subprocess.SubprocessError:
- # Fall back to X11
- subprocess.run(
- ["xclip", "-selection", "clipboard"],
- input=emoji.encode(),
- check=True,
- )
-
- # Track this emoji as recently used
- self._add_to_recent(emoji)
-
- except Exception as e:
- logger.error(f"Failed to copy to clipboard: {e}")
-
- def query(self, query_string: str) -> List[Result]:
- """Search emojis based on query."""
- results = []
- query = query_string.lower().strip()
-
- # Check if we need to download/reload data
- if not self.emoji_data and not self._is_downloading:
- if not os.path.exists(self.emoji_path):
- self._download_emoji_data()
- else:
- self._load_emoji_data()
-
- # Handle update command
- if query == "updatejson":
- self._download_emoji_data(force=True)
- return [
- Result(
- title="Updating Emoji Database...",
- subtitle=f"Downloading latest annotations for {self.lang}",
- icon_name="view-refresh-symbolic",
- relevance=1.0,
- plugin_name=self.display_name,
- )
- ]
-
- # If data is not yet loaded and not downloading, try loading it
- if not self.emoji_data and not self._is_downloading:
- self._load_emoji_data()
-
- # Show status if downloading
- if self._is_downloading:
- results.append(
- Result(
- title="Downloading Emoji Data...",
- subtitle="Results will appear once finished",
- icon_name="view-refresh-symbolic",
- relevance=0.1,
- plugin_name=self.display_name,
- )
- )
-
- # If no query, show recently used emojis
- if not query:
- if self.recent_emojis:
- for emoji in reversed(list(self.recent_emojis.keys())):
- if emoji in self.emoji_data:
- info = self.emoji_data[emoji]
- results.append(self._create_emoji_result(emoji, info, 1.0))
- return results
-
- # Search by name, keywords, or the emoji itself
- for emoji, info in self.emoji_data.items():
- relevance = 0
- name = info.get("name", "").lower()
- keywords = info.get("keywords", [])
-
- # Exact match with emoji
- if query == emoji:
- relevance = 1.0
- # Exact match with name
- elif query == name:
- relevance = 0.95
- # Starts with name
- elif name.startswith(query):
- relevance = 0.9
- # Contains name
- elif query in name:
- relevance = 0.8
- # In keywords
- elif any(query == k.lower() for k in keywords):
- relevance = 0.7
- elif any(query in k.lower() for k in keywords):
- relevance = 0.6
-
- if relevance > 0:
- results.append(self._create_emoji_result(emoji, info, relevance))
-
- # Sort by relevance and limit
- results.sort(key=lambda x: x.relevance, reverse=True)
- return results[:20]
-
- def _create_emoji_result(self, emoji: str, info: Dict, relevance: float) -> Result:
- """Create a Result object for an emoji."""
- name = info.get("name", "").capitalize()
- keywords = ", ".join(info.get("keywords", [])[:3])
-
- is_recent = emoji in self.recent_emojis
- subtitle = f"{keywords}" + (" • Recent" if is_recent else "")
-
- # Use larger font size for emojis to make them visible
- icon_markup = f"{GLib.markup_escape_text(emoji)}"
-
- return Result(
- title=name,
- subtitle=subtitle,
- icon_markup=icon_markup,
- action=lambda e=emoji: self._copy_to_clipboard(e),
- relevance=relevance,
- plugin_name=self.display_name,
- data={"emoji": emoji, "name": name, "recent": is_recent},
- )
diff --git a/src/window/launcher/plugins/power.py b/src/window/launcher/plugins/power.py
deleted file mode 100644
index 7c05d5d..0000000
--- a/src/window/launcher/plugins/power.py
+++ /dev/null
@@ -1,102 +0,0 @@
-from typing import List
-
-from fabric.utils import exec_shell_command_async
-
-from window.launcher.plugin_base import PluginBase
-from window.launcher.result import Result
-
-
-class PowerPlugin(PluginBase):
- """
- Plugin for system power management operations.
- """
-
- def __init__(self):
- super().__init__()
- self.display_name = "Power"
- self.description = "System power management and control"
- self.commands = {
- "shutdown": {
- "description": "Shutdown the system",
- "icon": "system-shutdown-symbolic",
- "action": self.shutdown,
- },
- "restart": {
- "description": "Restart the system",
- "icon": "system-reboot-symbolic",
- "action": self.restart,
- },
- "lock": {
- "description": "Lock the screen",
- "icon": "system-lock-screen-symbolic",
- "action": self.lock,
- },
- "suspend": {
- "description": "Suspend the system",
- "icon": "system-suspend-symbolic",
- "action": self.suspend,
- },
- "logout": {
- "description": "Logout from current session",
- "icon": "system-log-out-symbolic",
- "action": self.logout,
- },
- }
-
- def initialize(self):
- """Initialize the power plugin."""
- self.set_triggers(["power"])
-
- def cleanup(self):
- """Cleanup the power plugin."""
- pass
-
- def query(self, query_string: str) -> List[Result]:
- """Search power commands based on query."""
- query = query_string.lower().strip()
- results = []
-
- # 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_name=info["icon"],
- action=info["action"],
- 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_name=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
-
- def shutdown(self, *args) -> None:
- exec_shell_command_async("systemctl poweroff")
-
- def restart(self, *args) -> None:
- exec_shell_command_async("systemctl reboot")
-
- def lock(self, *args) -> None:
- exec_shell_command_async("loginctl lock-session")
-
- def suspend(self, *args) -> None:
- exec_shell_command_async("systemctl suspend")
-
- def logout(self, *args) -> None:
- exec_shell_command_async("hyprctl dispatch 'hl.dsp.exit()'")
diff --git a/src/window/launcher/plugins/screencapture.py b/src/window/launcher/plugins/screencapture.py
deleted file mode 100644
index d746813..0000000
--- a/src/window/launcher/plugins/screencapture.py
+++ /dev/null
@@ -1,442 +0,0 @@
-from typing import List
-from services.screencapture import screen_capture_service
-from window.launcher.plugin_base import PluginBase
-from window.launcher.result import Result
-
-
-class ScreencapturePlugin(PluginBase):
- """
- Plugin for taking screenshots and screen recordings using screen-capture.sh script.
- """
-
- def __init__(self):
- super().__init__()
- self.display_name = "Screencapture"
- self.description = "Take screenshots and screen recordings"
-
- def initialize(self):
- """Initialize the screencapture plugin."""
- self.set_triggers(["sc"])
-
- def cleanup(self):
- """Cleanup the screencapture plugin."""
- pass
-
- def get_commands(self):
- """Return available commands for this plugin."""
- return {
- # Screenshot commands
- "screenshot": "Take a screenshot of the main display",
- "screenshot-region": "Take a screenshot of selected region",
- "screenshot-both": "Take a screenshot of both displays",
- "screenshot-active": "Take a screenshot of the active display",
- # Recording commands (with audio)
- "record": "Start recording main display with audio",
- "record-region": "Start recording selected region with audio",
- "record-active": "Start recording active display with audio",
- # Recording overrides
- "record-noaudio": "Start recording main display without audio",
- "record-hq": "Start high-quality recording (eDP-1)",
- "record-gif": "Start GIF recording (eDP-1)",
- # Control commands
- "stop": "Stop current recording",
- # Conversion commands
- "convert-webm": "Convert latest MKV recording to WebM format",
- "convert-iphone": "Convert latest recordings for iPhone",
- "convert-youtube": "Convert latest recordings for YouTube",
- "convert-gif": "Convert latest recordings to GIF",
- # File conversion
- "convert-file": "Convert specific file (Usage: convert-file [type] [path])",
- }
-
- def _activate_default_plugins(self):
- """No action needed for script cleanup."""
- pass
-
- def _is_recording(self):
- """Check if recording is currently active."""
- return screen_capture_service.is_recording
-
- def _get_command_result(self, command: str) -> Result:
- """Get a Result object for a specific command."""
- # Map aliases manually if needed
- if command == "ss":
- command = "screenshot"
- if command == "rec":
- command = "record"
- if command == "conv":
- command = "convert"
-
- command_info = {
- # Screenshot commands
- "screenshot": (
- "Take Screenshot (eDP-1)",
- "Capture the main display",
- "camera-photo-symbolic",
- lambda: screen_capture_service.screenshot("eDP-1"),
- ),
- "screenshot-region": (
- "Take Region Screenshot",
- "Capture a selected region",
- "camera-photo-symbolic",
- lambda: screen_capture_service.screenshot("region"),
- ),
- "screenshot-both": (
- "Take Screenshot (Both Displays)",
- "Capture both displays combined",
- "video-joined-displays-symbolic",
- lambda: screen_capture_service.screenshot("both"),
- ),
- "screenshot-active": (
- "Take Screenshot (Active)",
- "Capture the active display",
- "camera-photo-symbolic",
- lambda: screen_capture_service.screenshot("active"),
- ),
- # Recording commands (with audio)
- "record": (
- "Start Recording (eDP-1)",
- "Record the main display with audio",
- "media-record-symbolic",
- lambda: screen_capture_service.record("eDP-1"),
- ),
- "record-region": (
- "Start Region Recording",
- "Record a selected region with audio",
- "media-record-symbolic",
- lambda: screen_capture_service.record("selection"),
- ),
- "record-active": (
- "Start Recording (Active)",
- "Record the active display with audio",
- "media-record-symbolic",
- lambda: screen_capture_service.record("active"),
- ),
- # Recording overrides
- "record-noaudio": (
- "Start Recording No Audio (eDP-1)",
- "Record without audio",
- "media-record-symbolic",
- lambda: screen_capture_service.record("eDP-1", no_audio=True),
- ),
- "record-hq": (
- "Start HQ Recording (eDP-1)",
- "High-quality recording for YouTube",
- "media-record-symbolic",
- lambda: screen_capture_service.record("eDP-1", mode="hq"),
- ),
- "record-gif": (
- "Start GIF Recording (eDP-1)",
- "Record as optimized GIF",
- "media-record-symbolic",
- lambda: screen_capture_service.record("eDP-1", mode="gif"),
- ),
- # Control commands
- "stop": (
- "Stop Recording",
- "Stop the current screen recording",
- "media-playback-stop-symbolic",
- lambda: screen_capture_service.stop_recording(),
- ),
- # Conversion commands
- "convert-webm": (
- "Convert Latest to WebM",
- "Convert latest MKV recording to WebM format",
- "video-x-generic-symbolic",
- lambda: screen_capture_service.convert("webm"),
- ),
- "convert-iphone": (
- "Convert Latest for iPhone",
- "Convert latest MKV recording for iPhone",
- "video-x-generic-symbolic",
- lambda: screen_capture_service.convert("iphone"),
- ),
- "convert-youtube": (
- "Convert Latest for YouTube",
- "Convert latest recording for YouTube",
- "video-x-generic-symbolic",
- lambda: screen_capture_service.convert("youtube"),
- ),
- "convert-gif": (
- "Convert Latest to GIF",
- "Convert latest recording to GIF",
- "image-x-generic-symbolic",
- lambda: screen_capture_service.convert("gif"),
- ),
- }
-
- if command in command_info:
- title, subtitle, icon, action = command_info[command]
- if action is not None: # Regular command
- return Result(
- title=title,
- subtitle=subtitle,
- icon_name=icon,
- action=action,
- relevance=1.0,
- plugin_name=self.display_name,
- )
- else: # File-based command, show instruction
- return Result(
- title=title,
- subtitle=subtitle,
- icon_name=icon,
- action=lambda: None, # No action for instruction
- relevance=1.0,
- plugin_name=self.display_name,
- )
-
- return None
-
- def query(self, query_string: str) -> List[Result]:
- """Search for screencapture actions based on query."""
- # Import here to avoid circular imports
-
- # Clean the query string
- query = query_string.strip().lower()
-
- results = []
-
- # Parse query for file-based commands
- query_parts = query.split()
- if len(query_parts) >= 3 and query_parts[0] == "convert-file":
- format_type = query_parts[1]
- file_param = " ".join(query_parts[2:])
- if format_type in ["webm", "iphone", "youtube", "gif"]:
- return [
- Result(
- title=f"Convert {file_param} to {format_type.upper()}",
- subtitle=f"Convert specified file to {format_type} format",
- icon_name=(
- "video-x-generic-symbolic"
- if format_type != "gif"
- else "image-x-generic-symbolic"
- ),
- action=lambda fp=file_param, ft=format_type: (
- screen_capture_service.convert(ft, fp)
- ),
- relevance=1.0,
- plugin_name=self.display_name,
- )
- ]
-
- # Check if query matches a command and return it as a result
- command_result = self._get_command_result(query)
- if command_result:
- return [command_result]
-
- # Check recording status
- is_recording = self._is_recording()
-
- # If recording is active, show stop button first with highest relevance
- if is_recording:
- results.append(
- Result(
- title="Stop Recording",
- subtitle="Stop the current screen recording",
- icon_name="media-playback-stop-symbolic",
- action=lambda: screen_capture_service.stop_recording(),
- relevance=2.0, # Highest relevance to appear at top
- plugin_name=self.display_name,
- )
- )
-
- # Screenshot actions
- results.extend(
- [
- Result(
- title="Take Screenshot",
- subtitle="Capture the entire screen (eDP-1)",
- icon_name="camera-photo-symbolic",
- action=lambda: screen_capture_service.screenshot("eDP-1"),
- relevance=1.0,
- plugin_name=self.display_name,
- ),
- Result(
- title="Take Region Screenshot",
- subtitle="Capture a selected region",
- icon_name="camera-photo-symbolic",
- action=lambda: screen_capture_service.screenshot("region"),
- relevance=0.9,
- plugin_name=self.display_name,
- ),
- Result(
- title="Take Screenshot (Both Displays)",
- subtitle="Capture both displays combined",
- icon_name="video-joined-displays-symbolic",
- action=lambda: screen_capture_service.screenshot("both"),
- relevance=0.8,
- plugin_name=self.display_name,
- ),
- Result(
- title="Take Screenshot (HDMI-A-1)",
- subtitle="Capture HDMI display",
- icon_name="video-display-symbolic",
- action=lambda: screen_capture_service.screenshot("HDMI-A-1"),
- relevance=0.7,
- plugin_name=self.display_name,
- ),
- ]
- )
-
- # Standard recording actions
- results.extend(
- [
- Result(
- title="Start Recording (eDP-1)",
- subtitle="Record the main display with audio",
- icon_name="media-record-symbolic",
- action=lambda: screen_capture_service.record("eDP-1"),
- relevance=0.7,
- plugin_name=self.display_name,
- ),
- Result(
- title="Start Region Recording",
- subtitle="Record a selected region",
- icon_name="media-record-symbolic",
- action=lambda: screen_capture_service.record("selection"),
- relevance=0.6,
- plugin_name=self.display_name,
- ),
- Result(
- title="Start Recording (HDMI-A-1)",
- subtitle="Record HDMI display with audio",
- icon_name="media-record-symbolic",
- action=lambda: screen_capture_service.record("HDMI-A-1"),
- relevance=0.5,
- plugin_name=self.display_name,
- ),
- ]
- )
-
- # No-audio recording actions
- results.extend(
- [
- Result(
- title="Start Recording No Audio (eDP-1)",
- subtitle="Record the main display without audio",
- icon_name="media-record-symbolic",
- action=lambda: screen_capture_service.record(
- "eDP-1", no_audio=True
- ),
- relevance=0.65,
- plugin_name=self.display_name,
- ),
- Result(
- title="Start Region Recording No Audio",
- subtitle="Record a selected region without audio",
- icon_name="media-record-symbolic",
- action=lambda: screen_capture_service.record(
- "selection", no_audio=True
- ),
- relevance=0.55,
- plugin_name=self.display_name,
- ),
- Result(
- title="Start Recording No Audio (HDMI-A-1)",
- subtitle="Record HDMI display without audio",
- icon_name="media-record-symbolic",
- action=lambda: screen_capture_service.record(
- "HDMI-A-1", no_audio=True
- ),
- relevance=0.45,
- plugin_name=self.display_name,
- ),
- ]
- )
-
- # High-quality recording actions
- results.extend(
- [
- Result(
- title="Start HQ Recording (eDP-1)",
- subtitle="High-quality recording for YouTube",
- icon_name="media-record-symbolic",
- action=lambda: screen_capture_service.record("eDP-1", mode="hq"),
- relevance=0.4,
- plugin_name=self.display_name,
- ),
- Result(
- title="Start HQ Region Recording",
- subtitle="High-quality region recording",
- icon_name="media-record-symbolic",
- action=lambda: screen_capture_service.record(
- "selection", mode="hq"
- ),
- relevance=0.3,
- plugin_name=self.display_name,
- ),
- Result(
- title="Start HQ Recording (HDMI-A-1)",
- subtitle="High-quality HDMI recording",
- icon_name="media-record-symbolic",
- action=lambda: screen_capture_service.record("HDMI-A-1", mode="hq"),
- relevance=0.2,
- plugin_name=self.display_name,
- ),
- ]
- )
-
- # GIF recording actions
- results.extend(
- [
- Result(
- title="Start GIF Recording (eDP-1)",
- subtitle="Record as optimized GIF",
- icon_name="media-record-symbolic",
- action=lambda: screen_capture_service.record("eDP-1", mode="gif"),
- relevance=0.1,
- plugin_name=self.display_name,
- ),
- Result(
- title="Start GIF Region Recording",
- subtitle="Record selected region as GIF",
- icon_name="media-record-symbolic",
- action=lambda: screen_capture_service.record(
- "selection", mode="gif"
- ),
- relevance=0.05,
- plugin_name=self.display_name,
- ),
- ]
- )
-
- # Conversion actions
- results.extend(
- [
- Result(
- title="Convert Latest to WebM",
- subtitle="Convert latest MKV recording to WebM format",
- icon_name="video-x-generic-symbolic",
- action=lambda: screen_capture_service.convert("webm"),
- relevance=0.01,
- plugin_name=self.display_name,
- ),
- Result(
- title="Convert Latest for iPhone",
- subtitle="Convert latest MKV recording for iPhone compatibility",
- icon_name="video-x-generic-symbolic",
- action=lambda: screen_capture_service.convert("iphone"),
- relevance=0.01,
- plugin_name=self.display_name,
- ),
- Result(
- title="Convert Latest for YouTube",
- subtitle="Convert latest recording for YouTube upload",
- icon_name="video-x-generic-symbolic",
- action=lambda: screen_capture_service.convert("youtube"),
- relevance=0.01,
- plugin_name=self.display_name,
- ),
- Result(
- title="Convert Latest to GIF",
- subtitle="Convert latest recording to GIF format",
- icon_name="image-x-generic-symbolic",
- action=lambda: screen_capture_service.convert("gif"),
- relevance=0.01,
- plugin_name=self.display_name,
- ),
- ]
- )
-
- return results
diff --git a/src/window/launcher/plugins/wallpaper.py b/src/window/launcher/plugins/wallpaper.py
deleted file mode 100644
index abe9ad9..0000000
--- a/src/window/launcher/plugins/wallpaper.py
+++ /dev/null
@@ -1,717 +0,0 @@
-import hashlib
-import threading
-
-import tomlkit
-from typing import Dict, List, Optional
-
-from fabric.utils import (
- GdkPixbuf,
- GLib,
- exec_shell_command_async,
- logger,
- os,
- random,
- time,
-)
-
-import shared.data as data
-from window.launcher.plugin_base import PluginBase
-from window.launcher.result import Result
-
-
-class WallpaperPlugin(PluginBase):
- """
- Plugin for wallpaper management with search, random selection, and matugen integration.
- """
-
- def __init__(self):
- super().__init__()
- 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",
- "scheme-expressive": "Expressive",
- "scheme-fidelity": "Fidelity",
- "scheme-fruit-salad": "Fruit Salad",
- "scheme-monochrome": "Monochrome",
- "scheme-neutral": "Neutral",
- "scheme-rainbow": "Rainbow",
- }
-
- def initialize(self):
- """Initialize the wallpaper plugin."""
- self.set_triggers(["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."""
- pass
-
- def _load_wallpapers(self):
- """Load available wallpapers from the wallpapers directory."""
- try:
- if os.path.exists(data.WALLPAPERS_DIR):
- self.wallpapers = sorted(
- [f for f in os.listdir(data.WALLPAPERS_DIR) if self._is_image(f)]
- )
- else:
- logger.error(f"Wallpapers directory not found: {data.WALLPAPERS_DIR}")
- except Exception as e:
- logger.error(f"Error loading wallpapers: {e}")
-
- def _is_image(self, filename: str) -> bool:
- """Check if file is a supported image format."""
- return filename.lower().endswith(
- (".png", ".jpg", ".jpeg", ".bmp", ".gif", ".webp")
- )
-
- def _get_matugen_state(self) -> bool:
- """Get current matugen state from config.toml."""
- self.matugen_enabled = True
- try:
- with open(data.CONFIG_FILE, "r") as f:
- config = tomlkit.load(f)
- self.matugen_enabled = config.get("matugen_enabled", True)
- except FileNotFoundError:
- pass
- except Exception as e:
- logger.error(f"Error reading config file: {e}")
-
- return self.matugen_enabled
-
- def _set_matugen_state(self, enabled: bool):
- """Set matugen state and save to config.toml."""
- self.matugen_enabled = enabled
- try:
- config = {}
- if os.path.exists(data.CONFIG_FILE):
- with open(data.CONFIG_FILE, "r") as f:
- config = tomlkit.load(f)
-
- config["matugen_enabled"] = enabled
-
- with open(data.CONFIG_FILE, "w") as f:
- tomlkit.dump(config, f)
-
- except Exception as e:
- logger.error(f"Error writing matugen state to config: {e}")
-
- # Clear the search query after toggling matugen
- self._clear_launcher_query()
-
- # # Send notification
- # status = "enabled" if enabled else "disabled"
- # exec_shell_command_async(
- # f"notify-send '🎨 Matugen' 'Dynamic colors {status}' -a '{
- # data.APP_NAME_CAP
- # }' -e"
- # )
-
- def _clear_launcher_query(self):
- """Clear the launcher search query and reset to trigger."""
- 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"):
-
- def clear_and_refresh():
- # Clear the search entry to just the trigger
- launcher.search_entry.set_text("wall ")
- # Position cursor at the end
- launcher.search_entry.set_position(-1)
- # Trigger the search to show default wallpaper view
- if hasattr(launcher, "_perform_search"):
- launcher._perform_search("wall ")
- return False
-
- # Use a small delay to ensure the action completes first
- GLib.timeout_add(50, clear_and_refresh)
- return
-
- # 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"):
-
- def clear_and_refresh():
- obj.search_entry.set_text("wall ")
- obj.search_entry.set_position(-1)
- obj._perform_search("wall ")
- return False
-
- GLib.timeout_add(50, clear_and_refresh)
- return
-
- except Exception as e:
- logger.error(f"Could not clear launcher query: {e}")
-
- def _get_cache_path(self, filename: str) -> str:
- """Get cache path for wallpaper thumbnail."""
- file_hash = hashlib.md5(filename.encode("utf-8")).hexdigest()
- return os.path.join(self.cache_dir, f"{file_hash}.png")
-
- def _create_thumbnail(self, filename: str) -> str:
- """Create thumbnail for wallpaper if it doesn't exist."""
- full_path = os.path.join(data.WALLPAPERS_DIR, filename)
- cache_path = self._get_cache_path(filename)
-
- if not os.path.exists(cache_path):
- try:
- pixbuf = GdkPixbuf.Pixbuf.new_from_file_at_scale(
- full_path, 32, 32, True
- )
- pixbuf.savev(cache_path, "png", [], [])
- except Exception as e:
- logger.error(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:
- logger.error(
- f"Error loading thumbnail for {wallpaper}: {e}"
- )
- self.thumbnail_cache[wallpaper] = None
- else:
- self.thumbnail_cache[wallpaper] = None
- except Exception as e:
- logger.error(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:
- logger.error(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:
- logger.error(f"Error loading thumbnail for {filename}: {e}")
- self.thumbnail_cache[filename] = None
- else:
- self.thumbnail_cache[filename] = None
- except Exception as e:
- logger.error(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 and apply matugen color scheme if enabled."""
- full_path = os.path.join(data.WALLPAPERS_DIR, filename)
- current_wall = os.path.expanduser("~/.current.wall")
-
- if scheme is None:
- scheme = self._get_current_scheme()
-
- # Update current wallpaper symlink
- if os.path.isfile(current_wall) or os.path.islink(current_wall):
- os.remove(current_wall)
- os.symlink(full_path, current_wall)
-
- # Always set the wallpaper image
- exec_shell_command_async(
- f'awww img "{
- full_path
- }" -t outer --transition-duration 1.5 --transition-step 255 --transition-fps 60 -f Nearest'
- )
-
- # If Matugen is enabled, also apply the color scheme
- matugen_enabled = self._get_matugen_state()
- if matugen_enabled:
- exec_shell_command_async(
- f'matugen image "{full_path}" -t {scheme} --source-color-index 0'
- )
-
- def _set_random_wallpaper(self):
- """Set a random wallpaper."""
- if not self.wallpapers:
- return
-
- filename = random.choice(self.wallpapers)
- self._set_wallpaper(filename)
- # Show notification for immediate feedback
- exec_shell_command_async(
- f"notify-send '🎲 Random Wallpaper' 'Applied: {filename}' -a '{
- data.APP_NAME_CAP
- }' -e"
- )
- return filename
-
- def _get_current_scheme(self) -> str:
- """Get current color scheme from config (default to tonal-spot)."""
- try:
- with open(data.CONFIG_FILE, "r") as f:
- config = tomlkit.load(f)
- return config.get("current_scheme", "scheme-tonal-spot")
- except FileNotFoundError:
- return "scheme-tonal-spot"
- except Exception as e:
- logger.error(f"Error reading current scheme from config: {e}")
- return "scheme-tonal-spot"
-
- def _set_current_scheme(self, scheme: str):
- """Set current color scheme and apply it."""
- scheme_name = self.schemes.get(scheme, scheme)
- matugen_enabled = self._get_matugen_state()
-
- try:
- config = tomlkit.document()
- if os.path.exists(data.CONFIG_FILE):
- with open(data.CONFIG_FILE, "r") as f:
- config = tomlkit.load(f)
-
- config["current_scheme"] = scheme
-
- with open(data.CONFIG_FILE, "w") as f:
- tomlkit.dump(config, f)
-
- except Exception as e:
- logger.error(f"Error saving current scheme to config: {e}")
- return
-
- # Apply the scheme to current wallpaper if matugen is enabled
- if matugen_enabled:
- current_wall = os.path.expanduser("~/.current.wall")
- if os.path.exists(current_wall) and os.path.islink(current_wall):
- # Get the current wallpaper path
- wallpaper_path = os.readlink(current_wall)
- if os.path.exists(wallpaper_path):
- # Apply the new scheme to current wallpaper
- exec_shell_command_async(
- f'matugen image "{wallpaper_path}" -t {scheme} --source-color-index 0'
- )
-
- # Send notification
- exec_shell_command_async(
- f"notify-send '🎨 Color Scheme' 'Applied {
- scheme_name
- } scheme' -a '{data.APP_NAME_CAP}' -e"
- )
- else:
- # No current wallpaper, just show scheme change notification
- exec_shell_command_async(
- f"notify-send '🎨 Color Scheme' 'Set to {
- scheme_name
- } (will apply to next wallpaper)' -a '{data.APP_NAME_CAP}' -e"
- )
- else:
- # No current wallpaper, just show scheme change notification
- exec_shell_command_async(
- f"notify-send '🎨 Color Scheme' 'Set to {
- scheme_name
- } (will apply to next wallpaper)' -a '{data.APP_NAME_CAP}' -e"
- )
- else:
- # Matugen is disabled, just save the setting
- exec_shell_command_async(
- f"notify-send '🎨 Color Scheme' 'Set to {
- scheme_name
- } (matugen disabled)' -a '{data.APP_NAME_CAP}' -e"
- )
-
- def _get_status_indicators(self) -> tuple:
- """Get current status indicators for display."""
- current_scheme = self._get_current_scheme()
- matugen_enabled = self._get_matugen_state()
- scheme_name = self.schemes.get(current_scheme, current_scheme)
-
- indicators = []
- if not matugen_enabled:
- indicators.append("Matugen Off")
-
- indicator_text = " • " + " • ".join(indicators) if indicators else ""
- status_text = f"Matugen: {
- 'Enabled' if matugen_enabled else 'Disabled'
- } • Scheme: {scheme_name}"
-
- return indicator_text, status_text, current_scheme, matugen_enabled
-
- def query(self, query_string: str) -> List[Result]:
- """Search wallpapers and provide management options."""
- results = []
- query = query_string.lower().strip()
-
- # Get status indicators for consistent display
- indicator_text, status_text, current_scheme, matugen_enabled = (
- self._get_status_indicators()
- )
-
- # Special commands
- if query.strip() == "random":
- # Show result for random wallpaper (execute on Enter)
- results.append(
- Result(
- title=f"Random Wallpaper{indicator_text}",
- subtitle=f"Set a random wallpaper • {status_text}",
- icon_name="media-playlist-shuffle-symbolic",
- action=lambda: self._set_random_wallpaper(),
- relevance=1.0,
- plugin_name=self.display_name,
- data={
- "action": "random",
- "bypass_max_results": True,
- "keep_launcher_open": True,
- },
- )
- )
- elif query.startswith("random") and query.strip() != "random":
- # Show suggestion for random wallpaper (partial match)
- results.append(
- Result(
- title=f"Random Wallpaper{indicator_text}",
- subtitle=f"Set a random wallpaper • {status_text}",
- icon_name="media-playlist-shuffle-symbolic",
- action=lambda: self._set_random_wallpaper(),
- relevance=0.9,
- plugin_name=self.display_name,
- data={
- "action": "random_suggestion",
- "bypass_max_results": True,
- "keep_launcher_open": True,
- },
- )
- )
-
- # Color scheme commands
- if "scheme" in query:
- current_scheme = self._get_current_scheme()
- matugen_enabled = self._get_matugen_state()
-
- # Show all available schemes
- for scheme_id, scheme_name in self.schemes.items():
- # Check if this scheme matches the query (for filtering)
- if query.strip() == "scheme":
- # Show all schemes when just "scheme" is typed
- scheme_matches = True
- else:
- # Extract search terms from query (remove "scheme" keyword)
- search_terms = query.replace("scheme", "").strip().split()
- scheme_matches = False
-
- # Check if any search term matches the scheme name or ID
- for term in search_terms:
- if (
- term.lower() in scheme_name.lower()
- or term.lower() in scheme_id.lower()
- ):
- scheme_matches = True
- break
-
- if scheme_matches:
- # Create indicators
- indicators = []
- if scheme_id == current_scheme:
- indicators.append("Current")
- if not matugen_enabled:
- indicators.append("Matugen Off")
-
- indicator_text = (
- " • " + " • ".join(indicators) if indicators else ""
- )
-
- results.append(
- Result(
- title=f"{scheme_name}{indicator_text}",
- subtitle=f"Set color scheme to {scheme_name}"
- + (
- " (requires matugen enabled)"
- if not matugen_enabled
- else ""
- ),
- icon_name="color-management-symbolic",
- action=lambda s=scheme_id: self._set_current_scheme(s),
- relevance=1.0 if scheme_id == current_scheme else 0.8,
- plugin_name=self.display_name,
- data={
- "action": "scheme_select",
- "scheme": scheme_id,
- "bypass_max_results": True,
- "keep_launcher_open": True,
- },
- )
- )
-
- # Matugen controls
- if "matugen" in query:
- current_state = self._get_matugen_state()
-
- # Check for exact command matches
- if query.strip() == "matugen on" or query.strip() == "matugen enable":
- # Show result for enabling matugen (execute on Enter)
- results.append(
- Result(
- title=f"Enable Matugen{indicator_text}",
- subtitle=f"Turn on dynamic color generation • {status_text}",
- icon_name="color-management-symbolic",
- action=lambda: self._set_matugen_state(True),
- relevance=0.9,
- plugin_name=self.display_name,
- data={
- "action": "matugen_on",
- "bypass_max_results": True,
- "keep_launcher_open": True,
- },
- )
- )
- elif query.strip() == "matugen off" or query.strip() == "matugen disable":
- # Show result for disabling matugen (execute on Enter)
- results.append(
- Result(
- title=f"Disable Matugen{indicator_text}",
- subtitle=f"Turn off dynamic color generation • {status_text}",
- icon_name="color-management-symbolic",
- action=lambda: self._set_matugen_state(False),
- relevance=0.9,
- plugin_name=self.display_name,
- data={
- "action": "matugen_off",
- "bypass_max_results": True,
- "keep_launcher_open": True,
- },
- )
- )
- elif query.strip() == "matugen toggle":
- # Show result for toggling matugen (execute on Enter)
- new_state = not current_state
- results.append(
- Result(
- title=f"Toggle Matugen to {
- 'Enabled' if new_state else 'Disabled'
- }{indicator_text}",
- subtitle=f"Switch matugen to {
- 'enabled' if new_state else 'disabled'
- } • {status_text}",
- icon_name="color-management-symbolic",
- action=lambda: self._set_matugen_state(new_state),
- relevance=0.9,
- plugin_name=self.display_name,
- data={
- "action": "matugen_toggle",
- "bypass_max_results": True,
- "keep_launcher_open": True,
- },
- )
- )
- elif ("on" in query or "enable" in query) and not query.strip().endswith(
- ("on", "enable")
- ):
- # Show suggestion for enabling (partial match)
- results.append(
- Result(
- title=f"Enable Matugen{indicator_text}",
- subtitle=f"Turn on dynamic color generation • {status_text}",
- icon_name="color-management-symbolic",
- action=lambda: self._set_matugen_state(True),
- relevance=0.8,
- plugin_name=self.display_name,
- data={
- "action": "matugen_on_suggestion",
- "bypass_max_results": True,
- "keep_launcher_open": True,
- },
- )
- )
- elif ("off" in query or "disable" in query) and not query.strip().endswith(
- ("off", "disable")
- ):
- # Show suggestion for disabling (partial match)
- results.append(
- Result(
- title=f"Disable Matugen{indicator_text}",
- subtitle=f"Turn off dynamic color generation • {status_text}",
- icon_name="color-management-symbolic",
- action=lambda: self._set_matugen_state(False),
- relevance=0.8,
- plugin_name=self.display_name,
- data={
- "action": "matugen_off_suggestion",
- "bypass_max_results": True,
- "keep_launcher_open": True,
- },
- )
- )
- elif "toggle" in query and not query.strip().endswith("toggle"):
- # Show suggestion for toggling (partial match)
- results.append(
- Result(
- title=f"Toggle Matugen{indicator_text}",
- subtitle=f"Switch matugen state • {status_text}",
- icon_name="color-management-symbolic",
- action=lambda: self._set_matugen_state(not current_state),
- relevance=0.8,
- plugin_name=self.display_name,
- data={
- "action": "matugen_toggle_suggestion",
- "bypass_max_results": True,
- "keep_launcher_open": True,
- },
- )
- )
- else:
- # Show current state with scheme info
- results.append(
- Result(
- title=f"Matugen: {'Enabled' if current_state else 'Disabled'}{
- indicator_text
- }",
- subtitle=f"Dynamic colors • {status_text}",
- icon_name="color-management-symbolic",
- action=lambda: None,
- relevance=0.8,
- plugin_name=self.display_name,
- data={"action": "matugen_status", "bypass_max_results": True},
- )
- )
-
- # Status command
- if query == "status" or query == "info":
- results.append(
- Result(
- title=f"Wallpaper System Status{indicator_text}",
- subtitle=status_text,
- icon_name="color-management-symbolic",
- action=lambda: None,
- relevance=1.0,
- plugin_name=self.display_name,
- data={"action": "status", "bypass_max_results": True},
- )
- )
-
- # Show quick actions
- results.append(
- Result(
- title=f"Random Wallpaper{indicator_text}",
- subtitle=f"Set a random wallpaper • {status_text}",
- icon_name="media-playlist-shuffle-symbolic",
- action=lambda: self._set_random_wallpaper(),
- relevance=0.9,
- plugin_name=self.display_name,
- data={
- "action": "random_quick",
- "bypass_max_results": True,
- "keep_launcher_open": True,
- },
- )
- )
-
- # Search wallpapers by filename - Show ALL wallpapers like example_wallpapers.py
- if not query or (
- query
- and "matugen" not in query
- and "random" not in query
- and "scheme" not in query
- and "status" not in query
- and "info" not in query
- ):
- matching_wallpapers = []
- for wallpaper in self.wallpapers:
- if not query or query in wallpaper.lower():
- # Calculate relevance
- relevance = 1.0 if query == wallpaper.lower() else 0.7
- if query and query in wallpaper.lower():
- relevance = 0.8
- matching_wallpapers.append((wallpaper, relevance))
-
- # Sort by relevance and show ALL wallpapers (like example_wallpapers.py)
- matching_wallpapers.sort(key=lambda x: x[1], reverse=True)
-
- # Show ALL wallpapers instead of limiting (following example_wallpapers.py pattern)
- for wallpaper, relevance in matching_wallpapers:
- # Use fast thumbnail loading
- icon = self._get_thumbnail_fast(wallpaper)
-
- results.append(
- Result(
- title=f"{wallpaper}{indicator_text if not query else ''}",
- subtitle=f"Set as wallpaper{
- ' • ' + status_text if not query else ''
- }",
- icon=icon,
- icon_name="image-x-generic-symbolic" if not icon else None,
- action=lambda w=wallpaper: self._set_wallpaper(w),
- relevance=relevance,
- plugin_name=self.display_name,
- data={
- "wallpaper": wallpaper,
- "action": "set",
- "bypass_max_results": True,
- "keep_launcher_open": True,
- },
- )
- )
-
- # Sort by relevance - no limit since we use bypass_max_results
- results.sort(key=lambda x: x.relevance, reverse=True)
- return results
diff --git a/src/window/launcher/plugins/websearch.py b/src/window/launcher/plugins/websearch.py
deleted file mode 100644
index 66a3f3c..0000000
--- a/src/window/launcher/plugins/websearch.py
+++ /dev/null
@@ -1,260 +0,0 @@
-from fabric.utils import logger
-import subprocess
-import urllib.parse
-from typing import List
-
-from window.launcher.plugin_base import PluginBase
-from window.launcher.result import Result
-
-
-class WebSearchPlugin(PluginBase):
- """
- Web search plugin that supports multiple search engines.
- """
-
- def __init__(self):
- super().__init__()
- self.display_name = "Web Search"
- self.description = "Search the web using various search engines"
- self.current_trigger = "" # Track the current active trigger
-
- # Define search engines with their URLs and icons
- self.search_engines = {
- "google": {
- "name": "Google",
- "url": "https://www.google.com/search?q={}",
- "icon": "google-symbolic",
- "description": "Search with Google",
- },
- "duckduckgo": {
- "name": "DuckDuckGo",
- "url": "https://duckduckgo.com/?q={}",
- "icon": "preferences-desktop-search",
- "description": "Search with DuckDuckGo (privacy-focused)",
- },
- "youtube": {
- "name": "YouTube",
- "url": "https://www.youtube.com/results?search_query={}",
- "icon": "youtube-symbolic",
- "description": "Search videos on YouTube",
- },
- "github": {
- "name": "GitHub",
- "url": "https://github.com/search?q={}",
- "icon": "github-desktop-symbolic",
- "description": "Search repositories on GitHub",
- },
- "stackoverflow": {
- "name": "Stack Overflow",
- "url": "https://stackoverflow.com/search?q={}",
- "icon": "stack",
- "description": "Search programming questions on Stack Overflow",
- },
- "wikipedia": {
- "name": "Wikipedia",
- "url": "https://en.wikipedia.org/wiki/Special:Search?search={}",
- "icon": "search-menus-symbolic",
- "description": "Search articles on Wikipedia",
- },
- "reddit": {
- "name": "Reddit",
- "url": "https://www.reddit.com/search/?q={}",
- "icon": "reddit-symbolic",
- "description": "Search discussions on Reddit",
- },
- "linkedin": {
- "name": "LinkedIn",
- "url": "https://www.linkedin.com/search/results/all/?keywords={}",
- "icon": "link",
- "description": "Search professionals and jobs on LinkedIn",
- },
- }
-
- # Default search engine
- self.default_engine = "google"
-
- def initialize(self):
- """Initialize the web search plugin."""
- # Set up triggers for web search - only main triggers, not individual search engines
- triggers = ["?"]
-
- # Don't add individual search engine triggers to avoid cluttering trigger keywords
- # Search engines can still be used within web search context (e.g., "web google cats")
-
- self.set_triggers(triggers)
-
- def get_active_trigger(self, query_string: str) -> str:
- """Override to track which trigger was activated."""
- trigger = super().get_active_trigger(query_string)
- if trigger:
- self.current_trigger = trigger.strip()
- return trigger
-
- def cleanup(self):
- """Cleanup the web search plugin."""
- pass
-
- def query(self, query_string: str) -> List[Result]:
- """Process web search queries."""
- results = []
-
- if not query_string.strip():
- # Show available search engines when no query
- return self._get_search_engine_list()
-
- # Check if the query is a URL
- if self._is_url(query_string):
- results.append(self._create_url_result(query_string))
- return results
-
- # Parse query to check if it starts with a search engine name
- engine_name, search_query = self._parse_engine_query(query_string)
-
- if engine_name and engine_name in self.search_engines:
- # Specific search engine specified in query (e.g., "google cats")
- if search_query:
- # Search with specific engine
- results.append(self._create_search_result(engine_name, search_query))
- else:
- # Show specific engine info
- results.append(self._create_engine_info_result(engine_name))
- else:
- # General search - offer multiple engines
- search_query = query_string.strip()
-
- # Add default search engine first
- results.append(
- self._create_search_result(self.default_engine, search_query)
- )
-
- # Add other popular search engines
- popular_engines = ["duckduckgo", "youtube", "github"]
- for engine in popular_engines:
- if engine != self.default_engine:
- results.append(self._create_search_result(engine, search_query))
-
- return results
-
- def _parse_engine_query(self, query: str) -> tuple[str, str]:
- """Parse query to extract search engine and search terms."""
- query = query.strip().lower()
-
- for engine in self.search_engines.keys():
- if query.startswith(f"{engine} "):
- return engine, query[len(engine) :].strip()
- elif query == engine:
- return engine, ""
-
- return "", query
-
- def _is_url(self, text: str) -> bool:
- """Check if the text is a URL."""
- text = text.strip().lower()
- return text.startswith(("http://", "https://", "www.")) or (
- "." in text and " " not in text and len(text) > 3
- )
-
- def _create_url_result(self, url: str) -> Result:
- """Create a result for opening a URL directly."""
- # Add protocol if missing
- if not url.startswith(("http://", "https://")):
- if url.startswith("www."):
- url = "https://" + url
- else:
- url = "https://" + url
-
- return Result(
- title=f"Open {url}",
- subtitle="Open this URL in your default browser",
- icon_name="link",
- action=lambda u=url: self._open_url(u),
- relevance=1.0,
- plugin_name=self.display_name,
- data={"type": "url", "url": url},
- )
-
- def _open_url(self, url: str):
- """Open a URL directly in the default browser."""
- try:
- subprocess.Popen(
- ["xdg-open", url], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL
- )
- except Exception as e:
- logger.error(f"Failed to open URL: {e}")
-
- def _get_search_engine_list(self) -> List[Result]:
- """Get list of available search engines."""
- results = []
-
- for engine_id, engine_info in self.search_engines.items():
- result = Result(
- title=engine_info["name"],
- subtitle=engine_info["description"],
- icon_name=engine_info["icon"],
- action=lambda e=engine_id: self._show_engine_help(e),
- relevance=1.0 if engine_id == self.default_engine else 0.8,
- plugin_name=self.display_name,
- data={"type": "engine_info", "engine": engine_id},
- )
- results.append(result)
-
- return results
-
- def _create_search_result(self, engine_id: str, query: str) -> Result:
- """Create a search result for a specific engine and query."""
- engine_info = self.search_engines[engine_id]
-
- return Result(
- title=f"Search '{query}' on {engine_info['name']}",
- subtitle=f"{engine_info['description']} - {query}",
- icon_name=engine_info["icon"],
- action=lambda e=engine_id, q=query: self._perform_search(e, q),
- relevance=1.0 if engine_id == self.default_engine else 0.9,
- plugin_name=self.display_name,
- data={"type": "search", "engine": engine_id, "query": query},
- )
-
- def _create_engine_info_result(self, engine_id: str) -> Result:
- """Create an info result for a specific search engine."""
- engine_info = self.search_engines[engine_id]
-
- return Result(
- title=f"{engine_info['name']} Search",
- subtitle=f"{engine_info['description']} - Type your search query",
- icon_name=engine_info["icon"],
- action=lambda: None, # No action for info result
- relevance=1.0,
- plugin_name=self.display_name,
- data={"type": "engine_ready", "engine": engine_id},
- )
-
- def _perform_search(self, engine_id: str, query: str):
- """Perform a web search using the specified engine."""
- if not query.strip():
- return
-
- engine_info = self.search_engines.get(engine_id)
- if not engine_info:
- return
-
- # URL encode the search query
- encoded_query = urllib.parse.quote_plus(query)
- search_url = engine_info["url"].format(encoded_query)
-
- try:
- # Open the search URL in the default browser
- subprocess.Popen(
- ["xdg-open", search_url],
- stdout=subprocess.DEVNULL,
- stderr=subprocess.DEVNULL,
- )
- except Exception as e:
- logger.error(f"Failed to open search URL: {e}")
-
- def _show_engine_help(self, engine_id: str):
- """Show help for a specific search engine."""
- engine_info = self.search_engines.get(engine_id)
- if engine_info:
- print(
- f"Search engine: {engine_info['name']} - {engine_info['description']}"
- )