chore(src/window/launcher/plugins): remove built-in plugins
This commit is contained in:
11 files changed
-2961
No files matched your search
@@ -1,3 +0,0 @@
|
||||
"""
|
||||
Built-in plugins for the launcher.
|
||||
"""
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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()
|
||||
@@ -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 "<meta http-equiv" not in line
|
||||
]
|
||||
else:
|
||||
items = []
|
||||
|
||||
# Check if data has changed
|
||||
with self.cache_lock:
|
||||
data_changed = (
|
||||
not self.clipboard_items_cache
|
||||
or len(items) != len(self.clipboard_items_cache)
|
||||
or items != self.clipboard_items_cache
|
||||
)
|
||||
|
||||
# If cache is still valid and data hasn't changed, return cached data
|
||||
if (
|
||||
not data_changed
|
||||
and self.clipboard_items_cache
|
||||
and current_time - self.cache_timestamp < self.cache_ttl
|
||||
):
|
||||
return self.clipboard_items_cache.copy()
|
||||
|
||||
# Update cache with fresh data
|
||||
self.clipboard_items_cache = items
|
||||
self.cache_timestamp = current_time
|
||||
|
||||
# Keep image cache forever like example_cliphist.py - don't clear on data changes
|
||||
|
||||
return items
|
||||
except subprocess.CalledProcessError as e:
|
||||
logger.error(f"Error loading clipboard history: {e}")
|
||||
return []
|
||||
except Exception as e:
|
||||
logger.error(f"Unexpected error: {e}")
|
||||
return []
|
||||
|
||||
def _load_clipboard_items(self) -> 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}")
|
||||
@@ -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])
|
||||
@@ -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"<span size='xx-large'>{GLib.markup_escape_text(emoji)}</span>"
|
||||
|
||||
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},
|
||||
)
|
||||
@@ -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()'")
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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']}"
|
||||
)
|
||||
Reference in new issue
Block a user