diff --git a/docs/plugins.md b/docs/plugins.md new file mode 100644 index 0000000..bd6f0f1 --- /dev/null +++ b/docs/plugins.md @@ -0,0 +1,533 @@ +# Plugin Development Guide + +Modus spotlight supports third-party plugins. Drop a Python file in `config/plugins/` and it gets auto-discovered on next startup. + +## Quick Start + +Copy the hello example to your plugins directory: + +```bash +cp -r examples/plugins/hello.py config/plugins/hello.py +``` + +Open spotlight and type `hi` or `hello`. The plugin shows a greeting result. + +## How It Works + +## How It Works + +1. On startup, Modus scans `config/plugins/` for `.py` files and package directories. +2. Each file is imported. The `PLUGIN` module-level variable is extracted (must be a `SpotlightPlugin` subclass). +3. Plugins are instantiated, initialized, and registered. +4. User types query -> spotlight routes to matching plugin (by keyword or global search) -> `search()` is called -> results are rendered. + +## Plugin Class Attributes + +| Attribute | Type | Required | Description | +|-----------------|------------|----------|-------------| +| `id` | `str` | Yes | Unique identifier (e.g., `"weather"`). | +| `name` | `str` | Yes | Human-readable name shown in UI. | +| `icon` | `str` | No | GTK icon name for the header. | +| `keywords` | `list[str]`| No | Trigger prefixes (e.g., `["wall", "wr"]`). User types `wall ` to activate. | +| `searchable` | `bool` | No | If `True`, plugin appears in global search (no prefix needed). Default `False`. | +| `priority` | `int` | No | Lower = searched first in global results. Default `100`. | +| `refresh_interval` | `int` | No | Auto-refresh timer in ms (0 = disabled). Fires `on_timer_tick()`. | + +## Methods + +### Required + +```python +def search(self, query: str, token: Any) -> list[SearchResult]: + """Return results matching query. + + token: Cancellation token. Check token.is_cancelled before expensive work. + Must return list of SearchResult. Never create GTK widgets. + """ +``` + +### Optional + +| Method | When Called | Use Case | +|--------|-------------|----------| +| `initialize()` | After instantiation | Load data, start services | +| `cleanup()` | On unload/disable | Free resources | +| `reload()` | On refresh | Default: cleanup + initialize | +| `on_activate(update_fn)` | Keyword prefix matched | Set up state for keyword mode | +| `on_deactivate()` | Keyword prefix cleared | Tear down keyword state | +| `on_submit(query)` | Enter pressed in keyword mode | Handle final input | +| `detect(text)` | Global search, each keystroke | Return `True` to exclusively handle input | +| `handle_external(command, args)` | External command (e.g., keybind) | Handle non-search invocations | +| `release_memory()` | Spotlight hides | Clear heavy cached data | + +## SearchResult Fields + +```python +SearchResult( + id="unique_id", # Unique within your plugin + title="Display Title", # Main text + subtitle="Description", # Secondary text + icon_name="", # GTK icon name (optional) + score=80.0, # 0-100, higher = better match + render_type="default", # Controls rendering (see below) + action=lambda: ..., # Called when user selects this result + plugin_name="", # Auto-filled by spotlight + plugin_id="", # Auto-filled by spotlight + metadata={}, # Arbitrary data for renderers +) +``` + +### Render Types + +| `render_type` | Widget Style | Use Case | +|-------------------|-----------------------------|-----------------------| +| `"default"` | Icon + title + subtitle | Generic results | +| `"app"` | Desktop app icon + name | Application spotlight | +| `"calc"` | Expression + result | Calculator | +| `"emoji"` | Emoji char + name + keywords | Emoji picker | +| `"power"` | Icon + name + description | Power menu | +| `"wallpaper"` | Thumbnail image | Wallpaper picker | +| `"clipboard_text"`| Text preview | Clipboard entries | +| `"clipboard_image"` | Image preview | Clipboard images | + +Use `"default"` unless you need a specific layout. Custom render types can be added to `main.py`'s `_render_result()`. + +## Keyword Plugins vs Global Search + +**Keyword-only** (`searchable = False`): +- User types `gg ` -> routes exclusively to your plugin +- `search()` receives the text after the prefix (e.g., `"linux news"`) +- Good for: YouTube search, Google search, wallpaper browser + +**Global search** (`searchable = True`): +- Plugin appears in results alongside apps, calculator, emoji +- `search()` receives the full query +- Good for: Calculator, emoji, power menu + +**Both** (`searchable = True` + `keywords`): +- Activates via keyword prefix AND appears in global results +- Good for: Wallpapers, clipboard, weather + +## Cancellation + +The `token` parameter has an `is_cancelled` flag. Check it before expensive operations: + +```python +def search(self, query: str, token: Any) -> list[SearchResult]: + results = [] + for item in self._large_dataset: + if token.is_cancelled: + return [] # Abort immediately + if self._matches(item, query): + results.append(item) + return results +``` + +The spotlight cancels previous searches when the user types new input. Without cancellation checks, your plugin wastes CPU on stale queries. + +## Examples + +### Hello World (single-file, no deps) + +File: `config/plugins/hello.py` + +```python +from typing import Any + +from window.spotlight.api import SpotlightPlugin, SearchResult + + +class HelloPlugin(SpotlightPlugin): + id = "hello" + name = "Hello" + icon = "face-smile-symbolic" + keywords = ["hi", "hello"] + searchable = True + priority = 100 + + def search(self, query: str, token: Any) -> list[SearchResult]: + if not query.strip(): + return [] + return [ + SearchResult( + id="hello_world", + title="Hello, World!", + subtitle="A friendly greeting", + icon_name="face-smile-symbolic", + score=100.0, + render_type="default", + action=lambda: print("Hello!"), + ) + ] + +PLUGIN = HelloPlugin +``` + +Test: `fabric-cli exec modus1 'deep_reload hello'` + +### Google / YouTube / Link Opener (keyword-only, shell-safe) + +File: `config/plugins/query.py` (built-in, but you can write your own) + +```python +import urllib.parse +from typing import Any + +from utils.functions import spawn_detached +from window.spotlight.api import SpotlightPlugin, SearchResult + + +class GooglePlugin(SpotlightPlugin): + id = "google" + name = "Google" + icon = "system-search-symbolic" + keywords = ["gg"] + searchable = False + + def search(self, query: str, token: Any) -> list[SearchResult]: + return [] + + def on_submit(self, query: str) -> None: + q = query.strip() + if q: + url = f"https://www.google.com/search?q={urllib.parse.quote(q)}" + spawn_detached(["xdg-open", url]) # no shell injection risk + + +PLUGIN = GooglePlugin +``` + +Usage: type `gg linux news` then Enter. Opens browser. + +### Weather Plugin (package plugin, third-party deps, I/O cancellation) + +Full example on disk at `config/plugins/weather/`. Demonstrates: +- Package plugin structure +- Auto-installed dependencies via `requirements.txt` + `uv` +- I/O with cancellation checks +- Result caching + +```python +"""Weather plugin for Modus spotlight. + +Demonstrates: + - Package plugin structure + - Third-party deps via requirements.txt (uv venv auto-install) + - Keyword activation + global search + - Deep reload during development + - Cancellation checks for I/O + - External command support +""" + +import os +from typing import Any + +from window.spotlight.api import SpotlightPlugin, SearchResult + +try: + import requests +except ImportError: + requests = None + + +class WeatherPlugin(SpotlightPlugin): + id = "weather" + name = "Weather" + icon = "weather-clear-symbolic" + keywords = ["wth", "weather"] + searchable = True + priority = 75 + + def __init__(self, context): + super().__init__(context) + self._cache: dict | None = None + + def initialize(self) -> None: + pass + + def cleanup(self) -> None: + self._cache = None + + def search(self, query: str, token: Any) -> list[SearchResult]: + q = query.strip().lower() + + # Strip keyword prefix ("wth london" -> "london") + for prefix in ("wth ", "weather "): + if q.startswith(prefix): + q = q[len(prefix):].strip() + break + if q in ("wth", "weather"): + q = "" + + if not q: + return [SearchResult(...hint...)] + + data = self._fetch_weather(q, token) + if token.is_cancelled or not data: + return [] + + temp = data.get("main", {}).get("temp", "?") + desc = data.get("weather", [{}])[0].get("description", "?") + city = data.get("name", q) + + return [ + SearchResult( + id=f"weather_{q}", + title=f"{city}: {temp}°C", + subtitle=desc.capitalize(), + icon_name="weather-clear-symbolic", + score=100.0, + render_type="default", + ) + ] + + def _fetch_weather(self, city: str, token: Any) -> dict | None: + """Fetch weather from OpenWeatherMap with caching.""" + if self._cache: + return self._cache + + api_key = os.environ.get("OPENWEATHER_API_KEY") + if not api_key or requests is None: + return None + + resp = requests.get( + "https://api.openweathermap.org/data/2.5/weather", + params={"q": city, "appid": api_key, "units": "metric"}, + timeout=10, + ) + if token.is_cancelled: + return None + if resp.status_code == 200: + self._cache = resp.json() + return self._cache + return None + + def handle_external(self, command: str, args: str) -> None: + print(f"External: {command} {args}") + + +PLUGIN = WeatherPlugin +``` + +File layout: + +``` +examples/plugins/weather/ + __init__.py # plugin code + requirements.txt # requests>=2.28 +``` + +Copy to plugins dir and try: +```bash +cp -r examples/plugins/weather config/plugins/weather +fabric-cli exec modus1 'deep_reload weather' +``` + +Now type `wth london` in spotlight. + +On first load, `PluginLoader` creates `config/plugins/.venvs/weather/` via `uv venv`, +installs `requests` via `uv pip install`, then imports your plugin. + +Usage: type `wth london` or just search `london` in global mode. + +Set env var in your Hyprland config: +``` +env = OPENWEATHER_API_KEY, your_key_here +``` + +### Color Picker (global search, clipboard action) + +```python +from typing import Any + +from utils.functions import run_command +from window.spotlight.api import SpotlightPlugin, SearchResult + + +class ColorPlugin(SpotlightPlugin): + id = "colors" + name = "Colors" + icon = "color-select-symbolic" + keywords = [] + searchable = True + priority = 80 + + COLORS = [ + ("Red", "#FF0000"), + ("Green", "#00FF00"), + ("Blue", "#0000FF"), + ] + + def search(self, query: str, token: Any) -> list[SearchResult]: + q = query.lower().strip() + if not q: + return [] + results = [] + for name, hex_code in self.COLORS: + if q in name.lower(): + results.append( + SearchResult( + id=f"color_{name.lower()}", + title=name, + subtitle=hex_code, + icon_name="color-select-symbolic", + score=90.0, + render_type="default", + action=lambda h=hex_code: self._copy_color(h), + ) + ) + return results + + def _copy_color(self, hex_code: str) -> None: + run_command(["wl-copy"], input=hex_code.encode(), text=False) + + +PLUGIN = ColorPlugin +``` + +## Package Plugins + +For complex plugins, use a directory with `plugin.py` or `__init__.py`: + +``` +config/plugins/ + myplugin/ + __init__.py # or plugin.py — must export PLUGIN = MyPluginClass + utils.py + data.json + requirements.txt # optional third-party deps +``` + +## Third-Party Dependencies + +If your plugin needs external Python packages, add `requirements.txt` next to your plugin file or in your package directory. Modus auto-installs them into an isolated virtualenv using `uv`. + +### Single-file plugin with dependencies + +``` +config/plugins/ + weather.py + weather/requirements.txt # companion directory +``` + +or + +``` +config/plugins/ + weather.py + requirements.txt # next to .py file +``` + +### Package plugin with dependencies + +``` +config/plugins/ + myplugin/ + __init__.py + utils.py + requirements.txt +``` + +### What happens at load time + +1. Modus checks for `requirements.txt` next to your `.py` file or in your package directory. +2. If found, it creates a venv at `config/plugins/.venvs//` using `uv venv`. +3. It runs `uv pip install -r requirements.txt --python /bin/python`. +4. The venv's `site-packages` is added to `sys.path` before your plugin is imported. +5. If `requirements.txt` changes later, deps are re-installed automatically. +6. **Requirement**: `uv` must be installed (`pip install uv` or `cargo install uv`). + +Only one `requirements.txt` per plugin is consulted. The venv persists across restarts. + +## Development Workflow + +### Edit → Reload → Test loop + +The fastest way to develop a plugin: + +1. Edit your plugin file +2. Run `fabric-cli exec modus1 'deep_reload your_plugin_id'` +3. Open spotlight and test + +No Modus restart needed. + +Example session: + +```bash +# Copy the weather example +cp -r examples/plugins/weather config/plugins/weather + +# --- First iteration --- +vim config/plugins/weather/__init__.py +fabric-cli exec modus1 'deep_reload weather' + +# --- Watch logs --- +journalctl -f --user -t modus + +# --- Second iteration --- +vim config/plugins/weather/__init__.py +fabric-cli exec modus1 'deep_reload weather' + +# --- Test output from terminal --- +fabric-cli exec modus1 'wth london' +``` + +### Deep reload under the hood + +``` +fabric-cli exec modus1 'deep_reload weather' +``` + +1. `Spotlight.handle_external("deep_reload", "weather")` catches the built-in command +2. `PluginManager.deep_reload("weather")` runs: + - `entry.instance.cleanup()` — free old resources + - `PluginLoader.unload_module("modus_plugin_user_weather")` — removes stale module from `sys.modules` + - `PluginLoader._load_package_plugin(path, "user")` — re-imports from disk + - `entry.plugin_class = fresh_class` — swaps the class reference + - `entry.instance = fresh_class(context)` — new instance + - `entry.instance.initialize()` — fresh setup + +### Check from terminal without UI + +```bash +fabric-cli exec modus1 'your_plugin_id some_query' +``` + +This routes to the plugin via keyword match or direct id lookup. + +### Errors & logging + +If a plugin fails to load or crashes during search, check the Modus log: + +```bash +journalctl --user -f -t modus +``` + +On reload failure, the old instance stays active. The log will show what went wrong. + +## External Commands (Keybinds) + +Plugins can be invoked from Hyprland keybinds via `handle_external`: + +```python +def handle_external(self, command: str, args: str) -> None: + if command == "myplugin": + self._do_stuff(args) +``` + +Hyprland config: +```lua +["SUPER + M"] = "fabric-cli exec modus1 'myplugin some_argument'" +``` + +## Tips + +- **Cache heavy data** in `initialize()`, not in `search()`. The search method runs on every keystroke. +- **Use `fuzzy_score`** from `utils.functions` for fuzzy matching against user input. +- **Keep `search()` fast**. If you do I/O, check `token.is_cancelled` frequently between operations. +- **Use `metadata`** to pass data to renderers. The spotlight owns all GTK widget creation — your plugin never touches widgets. +- **Use `run_command` from `utils.functions`** for subprocess calls — handles timeouts and errors. Never use shell strings (`exec_shell_command(f"...")`) with user input — that's a shell injection risk. +- **Use `spawn_detached` from `utils.functions`** to launch background processes without blocking. +- **Use `thread()` from `utils.functions`** for background work from `initialize()` or `handle_external()`. +- **Set `OPENWEATHER_API_KEY`** in your Hyprland env for the weather example. +- **Test from terminal**: `fabric-cli exec modus1 'deep_reload your_plugin_id'` diff --git a/examples/plugins/hello.py b/examples/plugins/hello.py new file mode 100644 index 0000000..3998e8d --- /dev/null +++ b/examples/plugins/hello.py @@ -0,0 +1,46 @@ +"""Example: simplest possible single-file plugin. + +Usage: + cp examples/plugins/hello.py config/plugins/hello.py + + Then open spotlight and type "hi" or "hello". + Or deep-reload from terminal: + fabric-cli exec modus1 'deep_reload hello' +""" + +from typing import Any + +from window.spotlight.api import SearchResult, SpotlightPlugin + + +class HelloPlugin(SpotlightPlugin): + id = "hello" + name = "Hello" + icon = "face-smile-symbolic" + keywords = ["hi", "hello"] + searchable = True + priority = 100 + + def initialize(self) -> None: + pass + + def cleanup(self) -> None: + pass + + def search(self, query: str, token: Any) -> list[SearchResult]: + if not query.strip(): + return [] + return [ + SearchResult( + id="hello_world", + title="Hello, World!", + subtitle="A friendly greeting", + icon_name="face-smile-symbolic", + score=100.0, + render_type="default", + action=lambda: print("Hello!"), + ) + ] + + +PLUGIN = HelloPlugin diff --git a/examples/plugins/weather/__init__.py b/examples/plugins/weather/__init__.py new file mode 100644 index 0000000..ca2e72a --- /dev/null +++ b/examples/plugins/weather/__init__.py @@ -0,0 +1,128 @@ +"""Example: package plugin with third-party dependencies. + +Demonstrates: + - Package plugin structure + - Third-party deps via requirements.txt (uv venv auto-install) + - Keyword activation + global search + - Deep reload during development + - Cancellation checks for I/O + - External command support + +Usage: + cp -r examples/plugins/weather config/plugins/weather + + Set your API key: + env = OPENWEATHER_API_KEY, your_key_here + + Open spotlight and type "wth london" or just "london". + After editing, deep-reload: + fabric-cli exec modus1 'deep_reload weather' +""" + +import os +from typing import Any + +from window.spotlight.api import SearchResult, SpotlightPlugin + +try: + import requests +except ImportError: + requests = None # type: ignore[assignment] + + +class WeatherPlugin(SpotlightPlugin): + id = "weather" + name = "Weather" + icon = "weather-clear-symbolic" + keywords = ["wth", "weather"] + searchable = True + priority = 75 + + def __init__(self, context): + super().__init__(context) + self._cache: dict | None = None + + def initialize(self) -> None: + pass + + def cleanup(self) -> None: + self._cache = None + + def release_memory(self) -> None: + self._cache = None + + def search(self, query: str, token: Any) -> list[SearchResult]: + q = query.strip().lower() + + for prefix in ("wth ", "weather "): + if q.startswith(prefix): + q = q[len(prefix) :].strip() + break + if q in ("wth", "weather"): + q = "" + + if not q: + return [ + SearchResult( + id="weather_hint", + title="Weather", + subtitle="Type 'wth ' to get current weather", + icon_name="weather-clear-symbolic", + score=100.0, + render_type="default", + ) + ] + + data = self._fetch_weather(q, token) + if token.is_cancelled or not data: + return [] + + temp = data.get("main", {}).get("temp", "?") + desc = data.get("weather", [{}])[0].get("description", "?") + city = data.get("name", q) + + return [ + SearchResult( + id=f"weather_{q}", + title=f"{city}: {temp}°C", + subtitle=desc.capitalize(), + icon_name="weather-clear-symbolic", + score=100.0, + render_type="default", + ) + ] + + def _fetch_weather(self, city: str, token: Any) -> dict | None: + if self._cache: + return self._cache + + api_key = os.environ.get("OPENWEATHER_API_KEY") + if not api_key or requests is None: + return None + + try: + resp = requests.get( + "https://api.openweathermap.org/data/2.5/weather", + params={"q": city, "appid": api_key, "units": "metric"}, + timeout=10, + ) + if token.is_cancelled: + return None + if resp.status_code == 200: + self._cache = resp.json() + return self._cache + except Exception: + pass + return None + + def handle_external(self, command: str, args: str) -> None: + print(f"[Weather] External: {command} {args}") + + def on_activate(self, update_fn=None) -> None: + print("[Weather] Keyword mode activated") + + def on_deactivate(self) -> None: + print("[Weather] Keyword mode deactivated") + + +PLUGIN = WeatherPlugin diff --git a/examples/plugins/weather/requirements.txt b/examples/plugins/weather/requirements.txt new file mode 100644 index 0000000..b18d513 --- /dev/null +++ b/examples/plugins/weather/requirements.txt @@ -0,0 +1 @@ +requests>=2.28 diff --git a/src/services/wallpaper.py b/src/services/wallpaper.py new file mode 100644 index 0000000..381a999 --- /dev/null +++ b/src/services/wallpaper.py @@ -0,0 +1,116 @@ +import hashlib +import mimetypes + +from fabric.core import Service, Signal +from fabric.utils import GdkPixbuf as gdk_pixbuf +from fabric.utils import exec_shell_command_async, os + +from shared.data import ( + WALLPAPER_PATH, + WALLPAPER_THUMBS_PATH, + WALLPAPERS_THUMBNAILS_SIZE, +) + +for path in (WALLPAPER_PATH, WALLPAPER_THUMBS_PATH): + os.makedirs(path, exist_ok=True) + + +def get_thumbnail_filename(image_path: str) -> str: + key = hashlib.sha256(image_path.encode()).hexdigest() + size = WALLPAPERS_THUMBNAILS_SIZE + return f"{key}_{size}x{size}.webp" + + +def create_thumbnail( + image_path: str, thumbnail_path: str, size: int = WALLPAPERS_THUMBNAILS_SIZE +) -> bool: + """Create WebP thumbnail with proper memory cleanup to prevent leaks.""" + pixbuf = None + try: + # Load image at scaled size to minimize memory usage + pixbuf = gdk_pixbuf.Pixbuf.new_from_file_at_scale(image_path, size, size, True) + # Save as WebP with quality 85 for better compression (60-80% smaller than PNG) + result = pixbuf.savev(thumbnail_path, "webp", ["quality"], ["85"]) + return bool(result) + except Exception: + return False + finally: + # Explicitly clear pixbuf reference to free memory + if pixbuf is not None: + del pixbuf + + +def generate_colors_from_wallpaper(image_path: str) -> bool: + cmd = f'matugen image "{image_path}"' + return bool(exec_shell_command_async(cmd)) + + +class WallpaperService(Service): + @Signal + def wallpaper_ready(self, image_path: str, thumbnail_path: str) -> None: ... + + @Signal + def wallpaper_set(self, image_path: str) -> None: ... + + @Signal + def colors_generated(self, image_path: str) -> None: ... + + def __init__(self, **kwargs): + super().__init__(**kwargs) + self._processed_count = 0 + self._file_cache: list[str] | None = None + + def get_all_wallpapers(self, use_cache: bool = True) -> list[str]: + if use_cache and self._file_cache is not None: + return self._file_cache + + all_files = [] + for file in os.listdir(WALLPAPER_PATH): + file_full = WALLPAPER_PATH + "/" + file + mt = mimetypes.guess_type(file_full)[0] + if mt and mt.startswith("image/"): + all_files.append(file_full) + + self._file_cache = all_files + return all_files + + def invalidate_cache(self) -> None: + self._file_cache = None + + def get_thumbnail_path(self, image_path: str) -> str: + thumbnail_filename = get_thumbnail_filename(image_path) + return WALLPAPER_THUMBS_PATH + "/" + thumbnail_filename + + def has_thumbnail(self, image_path: str) -> bool: + thumbnail_path = self.get_thumbnail_path(image_path) + return os.path.isfile(thumbnail_path) + + def set_wallpaper(self, image_path: str): + exec_shell_command_async( + f"awww img {image_path} --transition-type none --transition-duration 0 --transition-fps 60" + ) + self.wallpaper_set(image_path) + + target_path = os.path.abspath(image_path) + home_dir = os.path.expanduser("~") + link_path = os.path.join(home_dir, ".current.wall") + + if os.path.islink(link_path) or os.path.exists(link_path): + os.remove(link_path) + + os.symlink(target_path, link_path) + + exec_shell_command_async(f'matugen image "{image_path}" --source-color-index 0') + self.colors_generated(image_path) + + def get_wallpaper_info(self, image_path: str) -> dict: + """Get information about a wallpaper""" + stat = os.stat(image_path) + return { + "path": image_path, + "filename": os.path.basename(image_path), + "size": stat.st_size, + "modified": stat.st_mtime, + "has_thumbnail": self.has_thumbnail(image_path), + "thumbnail_path": self.get_thumbnail_path(image_path), + } diff --git a/src/styles/window/spotlight.css b/src/styles/window/spotlight.css new file mode 100644 index 0000000..540e757 --- /dev/null +++ b/src/styles/window/spotlight.css @@ -0,0 +1,175 @@ +@keyframes app-slot-shine { + from { + background-color: @fg-a12; + } + to { + background-color: alpha(@surface-bright, 0.5); + } +} + +@keyframes header-shine { + from { + background-color: @fg-a12; + } + to { + background-color: alpha(@surface-bright, 0.3); + } +} + +#spotlight { + margin: 8px; + padding: 5px; + border-radius: apply(radius-lg); + background-color: @bg-a80; + border: 1px solid @outline-a40; +} + +.spotlight-scroll-clip { + border-radius: apply(radius-md); +} + +.app-search-entry { + color: @foreground; +} + +.app-search-header { + padding: 8px; + border: solid 1px transparent; + border-radius: apply(radius-md); + background-color: @fg-a12; + transition: background-color apply(transition-medium) ease-in-out; +} + +.app-search-header.shine { + animation: header-shine 1s ease-in-out infinite alternate; +} + +.app-slot { + padding: 8px; + border: solid 1px transparent; + border-radius: apply(radius-md); + background-color: @fg-a12; + transition: border apply(transition-fast) ease-in-out, + background-color apply(transition-medium) ease-in-out; +} + +.app-slot:hover { + border-color: @outline-a40; + background-color: @fg-a18; +} + +.app-slot:focus, +.app-slot.selected { + border-color: @primary-a40; + background-color: @primary-a20; +} + +.app-slot.shine { + animation: app-slot-shine 0.3s ease-in-out; +} + +.app-title { + color: @foreground; + font-size: apply(font-base); + font-weight: 600; +} + +.app-description { + color: @text-dim; + font-size: apply(font-sm); +} + +.calc-expression { + color: @text-dim; + font-size: apply(font-base); +} + +.calc-result-text { + color: @foreground; + font-size: apply(font-lg); + font-weight: bold; +} + +.power-menu-name { + color: @foreground; + font-size: apply(font-lg); + font-weight: 600; +} + +.power-menu-description { + color: @text-muted; + font-size: apply(font-sm); +} + +.app-slot.wallpaper>box { + min-height: 4rem; + transition: all 0.2s cubic-bezier(0.2, 1, 0.8, 1); + background-size: cover; +} + +.app-slot.wallpaper:focus > box, +.app-slot.wallpaper.selected > box { + min-height: 12rem; +} + +.emoji-slot { + min-height: 3rem; +} + +.emoji-char { + font-size: 24px; + min-width: 2rem; +} + +.emoji-name { + color: var(--foreground); + font-size: 14px; + font-weight: normal; +} + +.emoji-no-results { + min-height: 3rem; +} + +.emoji-no-results-text { + color: darker(var(--foreground)); + font-size: 14px; + font-style: italic; +} + +.clipboard-no-results { + min-height: 3rem; +} + +.clipboard-no-results-text { + color: darker(var(--foreground)); + font-size: 14px; + font-style: italic; +} + +.clipboard-image { + min-height: 4rem; + transition: all 0.2s cubic-bezier(0.2, 1, 0.8, 1); + background-size: cover; +} + +.clipboard-image:focus, +.clipboard-image.selected { + min-height: 12rem; +} + +.clipboard-text { + min-height: 3rem; +} + +.clipboard-icon { + font-size: 20px; + min-width: 2rem; +} + +.clipboard-content { + color: var(--foreground); + font-size: 14px; + font-weight: normal; + margin-left: 1rem; +} diff --git a/src/window/spotlight/api/__init__.py b/src/window/spotlight/api/__init__.py new file mode 100644 index 0000000..5ed3e29 --- /dev/null +++ b/src/window/spotlight/api/__init__.py @@ -0,0 +1,5 @@ +from window.spotlight.api.context import PluginContext +from window.spotlight.api.plugin import SpotlightPlugin +from window.spotlight.api.result import SearchResult + +__all__ = ["PluginContext", "SearchResult", "SpotlightPlugin"] diff --git a/src/window/spotlight/api/context.py b/src/window/spotlight/api/context.py new file mode 100644 index 0000000..63fca74 --- /dev/null +++ b/src/window/spotlight/api/context.py @@ -0,0 +1,20 @@ +from typing import Any + + +class PluginContext: + """Dependency injection container passed to every plugin. + + Plugins access services through this instead of importing globals. + """ + + def __init__(self, services: dict[str, Any] | None = None): + self._services: dict[str, Any] = services or {} + + def get_service(self, name: str) -> Any | None: + return self._services.get(name) + + def register_service(self, name: str, service: Any) -> None: + self._services[name] = service + + def has_service(self, name: str) -> bool: + return name in self._services diff --git a/src/window/spotlight/api/plugin.py b/src/window/spotlight/api/plugin.py new file mode 100644 index 0000000..c571733 --- /dev/null +++ b/src/window/spotlight/api/plugin.py @@ -0,0 +1,94 @@ +from abc import ABC, abstractmethod +from typing import Any + +from window.spotlight.api.context import PluginContext +from window.spotlight.api.result import SearchResult + + +class SpotlightPlugin(ABC): + """Base class for all spotlight plugins. + + Plugins provide data via search(). Never create GTK widgets. + + Attributes: + id: Unique identifier (e.g., "calculator"). + name: Human-readable name (e.g., "Calculator"). + icon: Default GTK icon name. + keywords: Trigger words that route queries exclusively. + searchable: Whether plugin participates in global search. + priority: Lower = searched first. + refresh_interval: Auto-refresh interval in ms (0 = disabled). + """ + + id: str = "" + name: str = "" + icon: str = "" + keywords: list[str] | None = None + keyword_icons: dict[str, str] | None = None + searchable: bool = False + priority: int = 100 + refresh_interval: int = 0 + + def __init_subclass__(cls, **kwargs): + super().__init_subclass__(**kwargs) + if cls.keywords is None: + cls.keywords = [] + if cls.keyword_icons is None: + cls.keyword_icons = {} + + def __init__(self, context: PluginContext): + self.context = context + + def initialize(self) -> None: + """Called once after instantiation. Override for setup.""" + + def cleanup(self) -> None: + """Called on unload/disable. Override to release resources.""" + + def reload(self) -> None: + """Called when plugin needs to refresh its state. + Default: cleanup + re-initialize.""" + self.cleanup() + self.initialize() + + @abstractmethod + def search(self, query: str, token: Any) -> list[SearchResult]: + """Return results matching query. + + token: Cancellation token. Check token.is_cancelled before expensive work. + Must return list of SearchResult. Never create GTK widgets. + """ + ... + + def on_activate(self, update_fn=None) -> None: + """Called when keyword routes to this plugin. + + update_fn: callable(result_id, subtitle_text) for in-place label updates. + """ + + def on_deactivate(self) -> None: + """Called when keyword routes away from this plugin.""" + + def on_submit(self, query: str) -> None: + """Called when user presses Enter while this plugin is active.""" + + def detect(self, text: str) -> bool: + """Return True if this plugin should exclusively handle the input.""" + return False + + def handle_external(self, command: str, args: str) -> None: + """Handle external commands sent to the plugin.""" + + def release_memory(self) -> None: + """Release heavy cached data. Called when spotlight hides. + + Plugins should clear large caches here. They will be + reloaded lazily on next search. + """ + + def on_timer_tick(self) -> list[SearchResult] | None: + """Called every second when the plugin is active. + + Return updated results to re-render, or None to skip. + Used for live data like countdown timers. + """ diff --git a/src/window/spotlight/api/result.py b/src/window/spotlight/api/result.py new file mode 100644 index 0000000..6e2bd8e --- /dev/null +++ b/src/window/spotlight/api/result.py @@ -0,0 +1,31 @@ +from dataclasses import dataclass, field +from typing import Any, Callable + + +@dataclass +class SearchResult: + """Data carrier for a single search result. + + Plugins return these. Never create GTK widgets. + The spotlight owns all rendering. + """ + + id: str + title: str + subtitle: str = "" + icon_name: str = "" + icon_data: bytes | None = None + score: float = 0.0 + action: Callable[[], Any] | None = None + render_type: str = "default" + plugin_name: str = "" + plugin_id: str = "" + metadata: dict[str, Any] = field(default_factory=dict) + + def __post_init__(self): + self.score = max(0.0, min(100.0, self.score)) + + def activate(self): + if self.action: + return self.action() + return None diff --git a/src/window/spotlight/core/__init__.py b/src/window/spotlight/core/__init__.py new file mode 100644 index 0000000..b15d0d1 --- /dev/null +++ b/src/window/spotlight/core/__init__.py @@ -0,0 +1,14 @@ +from window.spotlight.core.loader import DiscoveredPlugin, PluginLoader +from window.spotlight.core.manager import PluginManager +from window.spotlight.core.registry import PluginEntry, PluginRegistry +from window.spotlight.core.search import SearchPipeline, SearchToken + +__all__ = [ + "DiscoveredPlugin", + "PluginEntry", + "PluginLoader", + "PluginManager", + "PluginRegistry", + "SearchPipeline", + "SearchToken", +] diff --git a/src/window/spotlight/core/loader.py b/src/window/spotlight/core/loader.py new file mode 100644 index 0000000..718f5ce --- /dev/null +++ b/src/window/spotlight/core/loader.py @@ -0,0 +1,399 @@ +import importlib +import importlib.util +import inspect +import subprocess +import sys +from dataclasses import dataclass +from pathlib import Path +from typing import Type + +from fabric.utils import logger + +from window.spotlight.api.plugin import SpotlightPlugin + +VENV_DIR = ".venvs" + + +@dataclass +class DiscoveredPlugin: + """A plugin class discovered on disk, not yet instantiated.""" + + plugin_class: Type[SpotlightPlugin] + source: str + path: str + module_name: str = "" + + +class PluginLoader: + """Discovers and imports plugins from filesystem paths. + + Handles: + - Single-file plugins (weather.py) + - Package plugins (weather/plugin.py or weather/__init__.py) + - Built-in plugins from the plugins/ directory + - Third-party plugins from config/data/system paths + - Per-plugin virtualenvs for isolated dependencies + """ + + USER_PATHS = [Path(__file__).resolve().parents[4] / "config" / "plugins"] + + def __init__(self, builtin_dir: str | Path): + self._builtin_dir = Path(builtin_dir) + self._loaded_modules: set[str] = set() + + def discover(self) -> list[DiscoveredPlugin]: + """Scan all paths and return discovered plugin classes.""" + found: list[DiscoveredPlugin] = [] + seen_ids: set[str] = set() + + # Built-in plugins first + for entry in self._discover_directory(self._builtin_dir, "builtin"): + if entry.plugin_class.id not in seen_ids: + seen_ids.add(entry.plugin_class.id) + found.append(entry) + + # Third-party paths + source_labels = ["user", "system"] + for path, source in zip(self.USER_PATHS, source_labels, strict=False): + if path.is_dir(): + for entry in self._discover_directory(path, source): + pid = entry.plugin_class.id + if pid and pid not in seen_ids: + seen_ids.add(pid) + found.append(entry) + elif pid in seen_ids: + logger.warning( + f"[PluginLoader] Duplicate plugin id '{pid}' " + f"from {entry.path}, skipping" + ) + + return found + + def _discover_directory( + self, directory: Path, source: str + ) -> list[DiscoveredPlugin]: + """Scan a directory for plugins.""" + results: list[DiscoveredPlugin] = [] + if not directory.is_dir(): + return results + + for item in sorted(directory.iterdir()): + try: + discovered = self._discover_item(item, source) + if discovered: + results.append(discovered) + except Exception as e: + logger.error(f"[PluginLoader] Error discovering {item.name}: {e}") + + return results + + def _discover_item(self, item: Path, source: str) -> DiscoveredPlugin | None: + """Discover a single plugin from a file or directory.""" + if item.is_file() and item.suffix == ".py": + if item.name.startswith("_"): + return None + return self._load_file_plugin(item, source) + + if item.is_dir() and not item.name.startswith("_"): + return self._load_package_plugin(item, source) + + return None + + def _load_file_plugin(self, filepath: Path, source: str) -> DiscoveredPlugin | None: + """Import a single .py file and extract PLUGIN.""" + # Check for requirements.txt next to the .py file + req_file = filepath.parent / "requirements.txt" + if req_file.exists(): + self._ensure_venv(filepath.parent, source) + + # Also check for a companion directory with requirements + # e.g. weather.py + weather/requirements.txt + companion_dir = filepath.parent / filepath.stem + if companion_dir.is_dir(): + companion_req = companion_dir / "requirements.txt" + if companion_req.exists(): + self._ensure_venv(companion_dir, source) + + module_name = f"modus_plugin_{source}_{filepath.stem}" + plugin_class = self._import_and_extract(filepath, module_name) + if plugin_class: + return DiscoveredPlugin( + plugin_class=plugin_class, + source=source, + path=str(filepath), + ) + return None + + def _load_package_plugin( + self, pkg_dir: Path, source: str + ) -> DiscoveredPlugin | None: + """Import a package plugin (plugin.py or __init__.py). + + Sets up the package hierarchy in sys.modules so that relative + imports (from .module import ...) work correctly. + """ + # Check for requirements.txt in package directory + req_file = pkg_dir / "requirements.txt" + if req_file.exists(): + self._ensure_venv(pkg_dir, source) + + parent_pkg = pkg_dir.parent.name + pkg_name = pkg_dir.name + full_pkg_name = f"{parent_pkg}.{pkg_name}" + mod_name = full_pkg_name + + # Ensure parent directory is on sys.path for submodule resolution + parent_dir = str(pkg_dir.parent) + if parent_dir not in sys.path: + sys.path.insert(0, parent_dir) + + # Set up package hierarchy so relative imports work + self._ensure_package_modules(pkg_dir.parent, parent_pkg, pkg_name) + + candidates = [ + (pkg_dir / "plugin.py", None), + (pkg_dir / "__init__.py", [str(pkg_dir)]), + ] + + for candidate, search_locs in candidates: + if candidate.exists(): + plugin_class = self._import_and_extract( + candidate, full_pkg_name, search_locs + ) + if plugin_class: + return DiscoveredPlugin( + plugin_class=plugin_class, + source=source, + path=str(candidate), + module_name=mod_name, + ) + + return None + + def _ensure_package_modules( + self, parent_dir: Path, parent_pkg: str, pkg_name: str + ) -> None: + """Create stub package modules in sys.modules for relative imports.""" + import types + + parent_mod_name = parent_pkg + if parent_mod_name not in sys.modules: + parent_mod = types.ModuleType(parent_mod_name) + parent_mod.__path__ = [str(parent_dir)] + parent_mod.__package__ = parent_mod_name + sys.modules[parent_mod_name] = parent_mod + + pkg_mod_name = f"{parent_pkg}.{pkg_name}" + if pkg_mod_name not in sys.modules: + pkg_dir = parent_dir / pkg_name + pkg_mod = types.ModuleType(pkg_mod_name) + pkg_mod.__path__ = [str(pkg_dir)] + pkg_mod.__package__ = pkg_mod_name + sys.modules[pkg_mod_name] = pkg_mod + + # ------------------------------------------------------------------ + # Plugin virtualenv management + # ------------------------------------------------------------------ + + def _ensure_venv(self, plugin_dir: Path, source: str) -> None: + """Create or update a venv for a plugin if requirements.txt exists. + + The venv lives at /.venvs// + Dependencies are installed into it, and its site-packages + are added to sys.path before the plugin is imported. + """ + req_file = plugin_dir / "requirements.txt" + if not req_file.exists(): + return + + venv_base = plugin_dir.parent / VENV_DIR + venv_base.mkdir(parents=True, exist_ok=True) + venv_dir = venv_base / plugin_dir.name + + if not venv_dir.is_dir(): + logger.info(f"[PluginLoader] Creating venv for {plugin_dir.name}") + if not self._create_venv(venv_dir): + return + + # Check if we need to install/update deps + marker_file = venv_dir / ".deps_installed" + needs_install = not marker_file.exists() or self._reqs_changed( + req_file, marker_file + ) + if needs_install: + logger.info(f"[PluginLoader] Installing deps for {plugin_dir.name}") + if self._install_deps(venv_dir, req_file): + marker_file.write_text(req_file.read_text()) + else: + logger.error( + f"[PluginLoader] Dep install failed for {plugin_dir.name}, " + "will retry next time" + ) + + # Add venv site-packages to sys.path + site_packages = self._find_site_packages(venv_dir) + if site_packages and str(site_packages) not in sys.path: + sys.path.insert(0, str(site_packages)) + + def _create_venv(self, venv_dir: Path) -> bool: + """Create a virtualenv using uv.""" + try: + result = subprocess.run( + ["uv", "venv", str(venv_dir)], + capture_output=True, + text=True, + timeout=30, + ) + if result.returncode != 0: + logger.error(f"[PluginLoader] uv venv failed: {result.stderr[:300]}") + return result.returncode == 0 + except FileNotFoundError: + logger.error("[PluginLoader] uv not found — install it first") + return False + except Exception as e: + logger.error(f"[PluginLoader] Failed to create venv: {e}") + return False + + def _install_deps(self, venv_dir: Path, req_file: Path) -> bool: + """Install dependencies from requirements.txt using uv.""" + venv_python = venv_dir / "bin" / "python" + try: + result = subprocess.run( + [ + "uv", + "pip", + "install", + "-r", + str(req_file), + "--python", + str(venv_python), + ], + capture_output=True, + text=True, + timeout=120, + ) + if result.returncode != 0: + logger.error( + f"[PluginLoader] uv pip install failed: {result.stderr[:500]}" + ) + return False + return True + except FileNotFoundError: + logger.error("[PluginLoader] uv not found — install it first") + return False + except Exception as e: + logger.error(f"[PluginLoader] Failed to install deps: {e}") + return False + + def _find_site_packages(self, venv_dir: Path) -> Path | None: + """Find site-packages directory in a venv.""" + # Look in lib/pythonX.Y/site-packages + lib_dir = venv_dir / "lib" + if lib_dir.is_dir(): + for item in lib_dir.iterdir(): + if item.name.startswith("python"): + site_pkgs = item / "site-packages" + if site_pkgs.is_dir(): + return site_pkgs + return None + + def _reqs_changed(self, req_file: Path, marker_file: Path) -> bool: + """Check if requirements.txt changed since last install.""" + try: + return req_file.read_text() != marker_file.read_text() + except Exception: + return True + + # ------------------------------------------------------------------ + # Module import + # ------------------------------------------------------------------ + + def _import_and_extract( + self, + filepath: Path, + module_name: str, + submodule_search_locations: list[str] | None = None, + ) -> Type[SpotlightPlugin] | None: + """Import a module and extract the PLUGIN attribute. + + For package __init__.py files, pass submodule_search_locations + so that relative imports within the package work. + """ + try: + spec = importlib.util.spec_from_file_location( + module_name, + filepath, + submodule_search_locations=submodule_search_locations, + ) + if not spec or not spec.loader: + return None + + module = importlib.util.module_from_spec(spec) + sys.modules[module_name] = module + self._loaded_modules.add(module_name) + spec.loader.exec_module(module) + + plugin = self._extract_plugin_class(module, module_name) + if not plugin: + self._unload_module(module_name) + return plugin + + except Exception as e: + logger.error(f"[PluginLoader] Failed to import {filepath.name}: {e}") + return None + + def _extract_plugin_class( + self, module, module_name: str + ) -> Type[SpotlightPlugin] | None: + """Extract plugin class from a module. + + Priority: + 1. MODULE.PLUGIN attribute (explicit, preferred) + 2. First SpotlightPlugin subclass found in module (fallback) + """ + # Preferred: explicit PLUGIN variable + plugin_attr = getattr(module, "PLUGIN", None) + if plugin_attr is not None: + if ( + inspect.isclass(plugin_attr) + and issubclass(plugin_attr, SpotlightPlugin) + and plugin_attr is not SpotlightPlugin + ): + return plugin_attr + logger.warning( + f"[PluginLoader] {module_name}.PLUGIN is not a " + f"SpotlightPlugin subclass, ignoring" + ) + return None + + # Fallback: scan for subclass + for attr_name in dir(module): + attr_value = getattr(module, attr_name) + if ( + inspect.isclass(attr_value) + and issubclass(attr_value, SpotlightPlugin) + and attr_value is not SpotlightPlugin + and inspect.getmodule(attr_value) is module + ): + logger.info( + f"[PluginLoader] Found {attr_name} via subclass scan " + f"in {module_name} (consider adding PLUGIN = {attr_name})" + ) + return attr_value + + logger.warning(f"[PluginLoader] No SpotlightPlugin found in {module_name}") + return None + + def unload_module(self, module_name: str) -> None: + """Remove a plugin module from sys.modules.""" + if module_name in self._loaded_modules: + self._loaded_modules.discard(module_name) + for key in list(sys.modules.keys()): + if key == module_name or key.startswith(module_name + "."): + sys.modules.pop(key, None) + + def unload_all(self) -> None: + """Remove all loaded plugin modules from sys.modules.""" + for name in list(self._loaded_modules): + self.unload_module(name) + self._loaded_modules.clear() diff --git a/src/window/spotlight/core/manager.py b/src/window/spotlight/core/manager.py new file mode 100644 index 0000000..b018a65 --- /dev/null +++ b/src/window/spotlight/core/manager.py @@ -0,0 +1,215 @@ +import json +from pathlib import Path + +from fabric.utils import logger + +from window.spotlight.api.context import PluginContext +from window.spotlight.api.plugin import SpotlightPlugin +from window.spotlight.core.loader import PluginLoader +from window.spotlight.core.registry import PluginEntry, PluginRegistry + + +class PluginManager: + """Manages plugin lifecycle: load, enable, disable, reload. + + Responsibilities: + - Orchestrate discovery (loader) and storage (registry) + - Instantiate and initialize enabled plugins + - Persist disabled state via TOML config + - Provide the active plugin set to the search pipeline + """ + + DISABLED_KEY = "disabled_plugins" + + def __init__( + self, + loader: PluginLoader, + registry: PluginRegistry, + context: PluginContext, + ): + self._loader = loader + self._registry = registry + self._context = context + self._disabled_ids: set[str] = set() + + @property + def registry(self) -> PluginRegistry: + return self._registry + + def load_all(self) -> None: + """Discover, register, and instantiate all plugins.""" + self._load_disabled_list() + + discovered = self._loader.discover() + for dp in discovered: + self._registry.register(dp) + + for entry in self._registry.get_all(): + if entry.id in self._disabled_ids: + entry.enabled = False + continue + self._instantiate(entry) + + logger.info( + f"[PluginManager] Loaded {len(self._registry.get_enabled())} plugins" + ) + + def enable(self, plugin_id: str) -> bool: + entry = self._registry.get(plugin_id) + if not entry or entry.enabled: + return False + + entry.enabled = True + self._disabled_ids.discard(plugin_id) + + if not entry.instance: + self._instantiate(entry) + + self._save_disabled_list() + return entry.instance is not None + + def disable(self, plugin_id: str) -> bool: + entry = self._registry.get(plugin_id) + if not entry or not entry.enabled: + return False + + self._call_cleanup(entry) + entry.enabled = False + self._disabled_ids.add(plugin_id) + self._save_disabled_list() + return True + + def reload(self, plugin_id: str) -> bool: + entry = self._registry.get(plugin_id) + if not entry: + return False + + if entry.instance: + try: + # Quick reload: just re-initialize same class instance + entry.instance.reload() + return True + except Exception as e: + logger.error(f"[PluginManager] Failed to reload {plugin_id}: {e}") + return False + + # Not instantiated yet, just re-instantiate + if entry.enabled: + return self._instantiate(entry) + return False + + def deep_reload(self, plugin_id: str) -> bool: + """Full re-import from disk. Picks up code changes. + + Destroys the old instance, removes the module from sys.modules, + re-discovers the plugin class, re-instantiates, and re-initializes. + Use during plugin development to avoid restarting Modus. + """ + entry = self._registry.get(plugin_id) + if not entry: + return False + + self._call_cleanup(entry) + if entry.module_name: + self._loader.unload_module(entry.module_name) + + from pathlib import Path + + fp = Path(entry.path) if entry.path else None + if fp and fp.exists(): + source = entry.source + if fp.is_file() and fp.suffix == ".py": + dp = self._loader._load_file_plugin(fp, source) + elif fp.is_dir(): + dp = self._loader._load_package_plugin(fp, source) + else: + dp = None + if dp and dp.plugin_class.id == plugin_id: + entry.plugin_class = dp.plugin_class + return self._instantiate(entry) + return False + + def reload_all(self) -> None: + for entry in self._registry.get_enabled(): + if entry.instance: + try: + entry.instance.reload() + except Exception as e: + logger.error(f"[PluginManager] Failed to reload {entry.id}: {e}") + + def reload_all_full(self) -> None: + """Reload all enabled plugins from disk. Accepts new classes, methods, etc.""" + for entry in self._registry.get_all(): + self.deep_reload(entry.id) + + def stop_all(self) -> None: + for entry in self._registry.get_all(): + self._call_cleanup(entry) + + def release_memory_all(self) -> None: + for entry in self._registry.get_enabled(): + if entry.instance: + try: + entry.instance.release_memory() + except Exception as e: + logger.error( + f"[PluginManager] Error releasing memory for {entry.id}: {e}" + ) + + def get_active_plugins(self) -> list[SpotlightPlugin]: + return [ + entry.instance for entry in self._registry.get_enabled() if entry.instance + ] + + def get_plugin(self, plugin_id: str) -> SpotlightPlugin | None: + entry = self._registry.get(plugin_id) + if entry and entry.enabled and entry.instance: + return entry.instance + return None + + def _instantiate(self, entry: PluginEntry) -> bool: + try: + instance = entry.plugin_class(self._context) + instance.initialize() + entry.instance = instance + logger.info(f"[PluginManager] Instantiated {entry.id} successfully") + return True + except Exception as e: + logger.error(f"[PluginManager] Failed to instantiate {entry.id}: {e}") + return False + + def _call_cleanup(self, entry: PluginEntry) -> None: + if not entry.instance: + return + try: + entry.instance.cleanup() + except Exception as e: + logger.error(f"[PluginManager] Error cleaning up {entry.id}: {e}") + entry.instance = None + + def _config_path(self) -> Path: + from utils.utils import toml_file + + return Path(toml_file("config.toml")).parent / "plugins.json" + + def _load_disabled_list(self) -> None: + path = self._config_path() + if not path.exists(): + self._disabled_ids = set() + return + try: + with open(path) as f: + data = json.load(f) + self._disabled_ids = set(data.get("disabled", [])) + except Exception as e: + logger.error(f"[PluginManager] Failed to load disabled list: {e}") + self._disabled_ids = set() + + def _save_disabled_list(self) -> None: + path = self._config_path() + try: + path.parent.mkdir(parents=True, exist_ok=True) + with open(path, "w") as f: + json.dump({"disabled": sorted(self._disabled_ids)}, f, indent=2) + except Exception as e: + logger.error(f"[PluginManager] Failed to save disabled list: {e}") diff --git a/src/window/spotlight/core/registry.py b/src/window/spotlight/core/registry.py new file mode 100644 index 0000000..0a757a4 --- /dev/null +++ b/src/window/spotlight/core/registry.py @@ -0,0 +1,112 @@ +from dataclasses import dataclass, field +from typing import Type + +from fabric.utils import logger + +from window.spotlight.api.plugin import SpotlightPlugin +from window.spotlight.core.loader import DiscoveredPlugin + + +@dataclass +class PluginEntry: + """Internal bookkeeping for a registered plugin.""" + + id: str + plugin_class: Type[SpotlightPlugin] + source: str + path: str + instance: SpotlightPlugin | None = None + enabled: bool = True + keywords: list[str] = field(default_factory=list) + module_name: str = "" + + +class PluginRegistry: + """Stores loaded plugin metadata and instances. + + Responsibilities: + - Register/unregister plugins + - Lookup by id, keyword + - Prevent duplicates + - Track enabled/disabled state + """ + + def __init__(self): + self._entries: dict[str, PluginEntry] = {} + self._keyword_map: dict[str, str] = {} # keyword -> plugin_id + + def register(self, discovered: DiscoveredPlugin) -> bool: + """Register a discovered plugin. Returns False if duplicate id.""" + cls = discovered.plugin_class + pid = cls.id + if not pid: + logger.warning( + f"[PluginRegistry] Plugin {cls.__name__} has no id, skipping" + ) + return False + + if pid in self._entries: + logger.warning( + f"[PluginRegistry] Duplicate plugin id '{pid}' from " + f"{discovered.path}, ignoring" + ) + return False + + entry = PluginEntry( + id=pid, + plugin_class=cls, + source=discovered.source, + path=discovered.path, + keywords=list(cls.keywords) if cls.keywords else [], + module_name=discovered.module_name, + ) + self._entries[pid] = entry + + for kw in entry.keywords: + if kw in self._keyword_map: + logger.warning( + f"[PluginRegistry] Keyword '{kw}' already mapped to " + f"'{self._keyword_map[kw]}', overriding with '{pid}'" + ) + self._keyword_map[kw] = pid + + return True + + def unregister(self, plugin_id: str) -> bool: + entry = self._entries.pop(plugin_id, None) + if not entry: + return False + for kw in entry.keywords: + if self._keyword_map.get(kw) == plugin_id: + del self._keyword_map[kw] + return True + + def get(self, plugin_id: str) -> PluginEntry | None: + return self._entries.get(plugin_id) + + def get_all(self) -> list[PluginEntry]: + return list(self._entries.values()) + + def get_enabled(self) -> list[PluginEntry]: + return [e for e in self._entries.values() if e.enabled] + + def get_searchable(self) -> list[PluginEntry]: + return [ + e for e in self._entries.values() if e.enabled and e.plugin_class.searchable + ] + + def get_by_keyword(self, keyword: str) -> PluginEntry | None: + pid = self._keyword_map.get(keyword) + if pid: + return self._entries.get(pid) + return None + + def is_registered(self, plugin_id: str) -> bool: + return plugin_id in self._entries + + def set_enabled(self, plugin_id: str, enabled: bool) -> bool: + entry = self._entries.get(plugin_id) + if not entry: + return False + entry.enabled = enabled + return True diff --git a/src/window/spotlight/core/search.py b/src/window/spotlight/core/search.py new file mode 100644 index 0000000..d9b69e9 --- /dev/null +++ b/src/window/spotlight/core/search.py @@ -0,0 +1,218 @@ +import threading +from dataclasses import dataclass, field +from typing import Callable + +from fabric.utils import GLib, logger + +from window.spotlight.api.plugin import SpotlightPlugin +from window.spotlight.api.result import SearchResult +from window.spotlight.core.manager import PluginManager +from window.spotlight.core.registry import PluginEntry + + +@dataclass +class SearchToken: + """Cancellation token for search operations.""" + + query: str + _cancelled: threading.Event = field(default_factory=threading.Event, repr=False) + + def cancel(self) -> None: + self._cancelled.set() + + @property + def is_cancelled(self) -> bool: + return self._cancelled.is_set() + + +class SearchPipeline: + """Async search pipeline that routes queries and merges results.""" + + def __init__(self, manager: PluginManager): + self._manager = manager + self._current_token: SearchToken | None = None + self._pending_threads: list[threading.Thread] = [] + self._pending_idle_ids: list[int] = [] + self._lock = threading.Lock() + + def search( + self, + query: str, + callback: Callable[[list[SearchResult]], None], + ) -> None: + self._cancel_current() + + if not query.strip(): + callback([]) + return + + token = SearchToken(query=query) + self._current_token = token + + keyword_plugin, _kw, _remaining = self._route(query) + + if keyword_plugin: + self._search_single(keyword_plugin, query, token, callback) + else: + self._global_search(query, token, callback) + + def search_single( + self, + entry: PluginEntry, + query: str, + callback: Callable[[list[SearchResult]], None], + ) -> None: + self._cancel_current() + token = SearchToken(query=query) + self._current_token = token + self._search_single(entry, query, token, callback) + + def cancel(self) -> None: + self._cancel_current() + + def _cancel_current(self) -> None: + if self._current_token: + self._current_token.cancel() + with self._lock: + for t in self._pending_threads: + t.join(timeout=0.5) + self._pending_threads.clear() + for source_id in self._pending_idle_ids: + GLib.source_remove(source_id) + self._pending_idle_ids.clear() + + def _route(self, query: str) -> tuple[PluginEntry | None, str, str]: + stripped = query.strip() + lower = stripped.lower() + registry = self._manager.registry + + for entry in registry.get_enabled(): + for keyword in entry.keywords: + kw = keyword.lower() + if lower.startswith(kw + " ") or lower == kw: + remaining = stripped[len(keyword) :].lstrip() + return entry, keyword, remaining + + return None, "", query + + def _search_single( + self, + entry: PluginEntry, + query: str, + token: SearchToken, + callback: Callable[[list[SearchResult]], None], + ) -> None: + instance = entry.instance + if not instance: + callback([]) + return + + def _run(): + try: + results = self._safe_search(instance, query, token) + self._stamp_results(results, entry) + if not token.is_cancelled: + source_id = GLib.idle_add(lambda: callback(results)) + with self._lock: + self._pending_idle_ids.append(source_id) + except Exception as e: + logger.error(f"[SearchPipeline] Error searching {entry.id}: {e}") + if not token.is_cancelled: + source_id = GLib.idle_add(lambda: callback([])) + with self._lock: + self._pending_idle_ids.append(source_id) + + thread = threading.Thread(target=_run, daemon=True) + with self._lock: + self._pending_threads.append(thread) + thread.start() + + def _global_search( + self, + query: str, + token: SearchToken, + callback: Callable[[list[SearchResult]], None], + ) -> None: + searchable = self._manager.registry.get_searchable() + if not searchable: + callback([]) + return + + results_lock = threading.Lock() + collected: dict[str, list[SearchResult]] = {} + remaining = len(searchable) + + def _on_all_done(): + merged = self._merge_results(collected) + callback(merged) + + def _plugin_done(plugin_id: str): + nonlocal remaining + with results_lock: + remaining -= 1 + if remaining <= 0: + source_id = GLib.idle_add(_on_all_done) + with self._lock: + self._pending_idle_ids.append(source_id) + + def _run(entry: PluginEntry): + instance = entry.instance + if not instance or token.is_cancelled: + _plugin_done(entry.id) + return + + try: + results = self._safe_search(instance, query, token) + self._stamp_results(results, entry) + with results_lock: + collected[entry.id] = results + except Exception as e: + logger.error(f"[SearchPipeline] Plugin {entry.id} crashed: {e}") + finally: + _plugin_done(entry.id) + + for entry in searchable: + thread = threading.Thread(target=_run, args=(entry,), daemon=True) + with self._lock: + self._pending_threads.append(thread) + thread.start() + + def _safe_search( + self, + plugin: SpotlightPlugin, + query: str, + token: SearchToken, + ) -> list[SearchResult]: + if token.is_cancelled: + return [] + results = plugin.search(query, token) + if token.is_cancelled: + return [] + return results or [] + + def _stamp_results(self, results: list[SearchResult], entry: PluginEntry) -> None: + for r in results: + if not r.plugin_name: + r.plugin_name = entry.plugin_class.name + if not r.plugin_id: + r.plugin_id = entry.id + + def _merge_results( + self, grouped: dict[str, list[SearchResult]] + ) -> list[SearchResult]: + enabled = {e.id: e for e in self._manager.registry.get_enabled()} + + ordered_ids = sorted( + grouped.keys(), + key=lambda pid: ( + enabled[pid].plugin_class.priority if pid in enabled else 999 + ), + ) + + merged: list[SearchResult] = [] + for pid in ordered_ids: + plugin_results = grouped[pid] + plugin_results.sort(key=lambda r: r.score, reverse=True) + merged.extend(plugin_results) + + return merged diff --git a/src/window/spotlight/main.py b/src/window/spotlight/main.py index 44110c1..73edd9b 100644 --- a/src/window/spotlight/main.py +++ b/src/window/spotlight/main.py @@ -97,10 +97,7 @@ class Spotlight(Box): self.children = self.header, self.scrolled_clip - # ------------------------------------------------------------------ # Search - # ------------------------------------------------------------------ - @debounce(200) def on_entry_changed(self, entry: Entry, *_): if getattr(entry, "_ignore_change", False): @@ -246,10 +243,7 @@ class Spotlight(Box): self._post_arrange() - # ------------------------------------------------------------------ # Rendering - # ------------------------------------------------------------------ - def _render_result(self, result: SearchResult) -> Button: rt = result.render_type if rt == "app": @@ -511,10 +505,7 @@ class Spotlight(Box): result.action() self.launched() - # ------------------------------------------------------------------ # Viewport management - # ------------------------------------------------------------------ - def _clear_viewport(self): old_results = self._current_results self._current_results = [] @@ -569,10 +560,7 @@ class Spotlight(Box): if self.viewport.children: self.set_selected_index(0) - # ------------------------------------------------------------------ # Selection & keyboard - # ------------------------------------------------------------------ - def on_slot_enter(self, button, event): children = self.viewport.children if button in children: @@ -611,10 +599,7 @@ class Spotlight(Box): return True return False - # ------------------------------------------------------------------ # Submit - # ------------------------------------------------------------------ - def on_entry_accept(self, entry: Entry, *_): if 0 <= self._selected_index < len(self.viewport.children): widget = self.viewport.children[self._selected_index] @@ -632,10 +617,7 @@ class Spotlight(Box): if self._current_results: self._activate_result(self._current_results[0]) - # ------------------------------------------------------------------ # External commands - # ------------------------------------------------------------------ - def handle_external(self, command: str, text: str = "") -> bool: if command == "deep_reload" and text: return self.plugin_service.manager.deep_reload(text.strip()) @@ -662,10 +644,10 @@ class SpotlightWindow(Window): ) super().__init__( name="spotlight-window", - title="fabric-spotlight", + title="modus-spotlight", anchor="top", margin="80px 0px 0px 0px", - keyboard_mode="on-demand", + keyboard_mode="exclusive", child=self.spotlight_box, visible=False, all_visible=False, diff --git a/src/window/spotlight/plugins/__init__.py b/src/window/spotlight/plugins/__init__.py new file mode 100644 index 0000000..30eec8f --- /dev/null +++ b/src/window/spotlight/plugins/__init__.py @@ -0,0 +1,24 @@ +""" +Modus Launcher plugins package. + +Built-in plugins are auto-discovered via pkgutil. +Each plugin module exposes a PLUGIN variable pointing to its class. +""" + +import importlib +import pkgutil + +__all__ = [] + +for module_info in pkgutil.iter_modules(__path__): + module_name = module_info.name + if module_name.startswith("_"): + continue + try: + module = importlib.import_module(f"{__name__}.{module_name}") + plugin_class = getattr(module, "PLUGIN", None) + if plugin_class is not None: + globals()[module_name] = plugin_class + __all__.append(module_name) + except Exception: + pass diff --git a/src/window/spotlight/plugins/application.py b/src/window/spotlight/plugins/application.py new file mode 100644 index 0000000..cc75f51 --- /dev/null +++ b/src/window/spotlight/plugins/application.py @@ -0,0 +1,66 @@ +from typing import Any + +from fabric.utils import DesktopApp, get_desktop_applications + +from utils.functions import fuzzy_score +from window.spotlight.api import SearchResult, SpotlightPlugin + + +class ApplicationPlugin(SpotlightPlugin): + id = "applications" + name = "Applications" + icon = "system-search-symbolic" + keywords: list[str] = [] + searchable = True + priority = 10 + + def __init__(self, context): + super().__init__(context) + self._desktop_apps: list[DesktopApp] = [] + + def initialize(self) -> None: + self._desktop_apps = get_desktop_applications() + + def cleanup(self) -> None: + self._desktop_apps = [] + + def search(self, query: str, token: Any) -> list[SearchResult]: + if not query: + return [] + + results: list[SearchResult] = [] + for app in self._desktop_apps: + display_name = getattr(app, "display_name", "") or "" + if not display_name: + continue + + raw_score = fuzzy_score(query, display_name) + if raw_score <= 0: + continue + + score = raw_score / 100.0 + + results.append( + SearchResult( + id=getattr(app, "desktop_id", "") or display_name, + title=display_name, + subtitle=getattr(app, "description", "") or "", + icon_name="", + score=score, + action=lambda a=app: self._launch_app(a), + render_type="app", + plugin_name=self.name, + plugin_id=self.id, + metadata={"desktop_app": app}, + ) + ) + + results.sort(key=lambda r: r.score, reverse=True) + return results[:10] + + @staticmethod + def _launch_app(app: DesktopApp) -> None: + app.launch() + + +PLUGIN = ApplicationPlugin diff --git a/src/window/spotlight/plugins/calculator.py b/src/window/spotlight/plugins/calculator.py new file mode 100644 index 0000000..7dfa3f0 --- /dev/null +++ b/src/window/spotlight/plugins/calculator.py @@ -0,0 +1,93 @@ +from typing import Any + +from utils.functions import run_command, spawn_detached +from window.spotlight.api import PluginContext, SearchResult, SpotlightPlugin + + +class CalculatorPlugin(SpotlightPlugin): + id = "calculator" + name = "calculator" + icon = "accessories-calculator-symbolic" + keywords = ["calc"] + searchable = True + priority = 90 + + def __init__(self, context: PluginContext) -> None: + super().__init__(context) + + def initialize(self) -> None: + pass + + def cleanup(self) -> None: + pass + + def detect(self, text: str) -> bool: + s = text.strip() + if not s: + return False + has_digit = any(c.isdigit() for c in s) + has_op = any(c in "+-*/^%()" for c in s) + return has_digit and has_op + + def search(self, query: str, token: Any) -> list[SearchResult]: + s = query.strip() + if not s: + return [] + + if s.lower().startswith("calc "): + s = s[5:].strip() + elif s.lower() == "calc": + return [ + SearchResult( + id="calc_hint", + title="Calculator", + subtitle="Type a math expression (e.g. 2+2)", + icon_name="accessories-calculator-symbolic", + score=100.0, + render_type="calc", + ) + ] + + if not s: + return [] + + has_digit = any(c.isdigit() for c in s) + has_op = any(c in "+-*/^%()" for c in s) + if not (has_digit and has_op): + return [] + + result = self._run_qalc(s) + if not result or not result.strip(): + return [] + + if " = " in result: + answer = result.split(" = ")[-1].strip() + else: + answer = result.strip() + + return [ + SearchResult( + id=f"calc_{s}", + title=f"{s} =", + subtitle=answer, + icon_name="accessories-calculator-symbolic", + score=90.0, + render_type="calc", + action=lambda s=s: self._run_qalc_async(s), + metadata={}, + ) + ] + + @staticmethod + def _run_qalc(expression: str) -> str: + result = run_command(["qalc", "-t", expression], timeout=10) + if result.returncode not in [0, -1]: + return "" + return result.stdout or "" + + @staticmethod + def _run_qalc_async(expression: str) -> None: + spawn_detached(["qalc", expression]) + + +PLUGIN = CalculatorPlugin diff --git a/src/window/spotlight/plugins/clipboard.py b/src/window/spotlight/plugins/clipboard.py new file mode 100644 index 0000000..3cae302 --- /dev/null +++ b/src/window/spotlight/plugins/clipboard.py @@ -0,0 +1,223 @@ +import os +import threading +from typing import Any + +from fabric.utils import Gio, GLib + +from shared.data import CLIPBOARD_DB_PATH, CLIPBOARD_THUMBS_DIR +from utils.functions import copy_image, copy_text, run_command, trigger_paste_shortcut +from window.spotlight.api import PluginContext, SearchResult, SpotlightPlugin + + +class ClipboardPlugin(SpotlightPlugin): + id = "clipboard" + name = "Clipboard" + icon = "edit-paste-symbolic" + searchable = True + priority = 80 + + MAX_RESULTS: int = 30 + + def __init__(self, context: PluginContext): + super().__init__(context) + self._history: list[dict] = [] + self._history_lock: threading.Lock = threading.Lock() + self._cache_dir: str = CLIPBOARD_THUMBS_DIR + self._monitor: Gio.FileMonitor | None = None + self._db_debounce_id: int = 0 + + def initialize(self) -> None: + os.makedirs(self._cache_dir, exist_ok=True) + self._load_history() + self._setup_file_monitor() + + def cleanup(self) -> None: + if self._monitor: + self._monitor.cancel() + self._monitor = None + if self._db_debounce_id: + GLib.source_remove(self._db_debounce_id) + self._db_debounce_id = 0 + + def release_memory(self) -> None: + with self._history_lock: + self._history.clear() + + def search(self, query: str, token: Any) -> list[SearchResult]: + q = query.strip() + if q.lower().startswith("clip "): + q = q[5:].strip() + elif q.lower() == "clip": + q = "" + + self._ensure_history() + return self._build_results(self._filter_entries(q), q) + + def _ensure_history(self) -> None: + with self._history_lock: + if self._history: + return + self._load_history() + + def _setup_file_monitor(self) -> None: + if not os.path.exists(CLIPBOARD_DB_PATH): + return + f = Gio.File.new_for_path(CLIPBOARD_DB_PATH) + self._monitor = f.monitor_file(Gio.FileMonitorFlags.NONE, None) + self._monitor.connect("changed", self._on_db_changed) + + def _on_db_changed(self, _monitor, _file, _other_file, event_type): + if event_type not in ( + Gio.FileMonitorEvent.CHANGES_DONE_HINT, + Gio.FileMonitorEvent.CREATED, + ): + return + if self._db_debounce_id: + GLib.source_remove(self._db_debounce_id) + self._db_debounce_id = GLib.timeout_add(300, self._on_debounced_db_change) + + def _on_debounced_db_change(self): + self._db_debounce_id = 0 + self._load_history() + + def _load_history(self) -> None: + result = run_command(["cliphist", "list"], timeout=5) + raw = result.stdout if isinstance(result.stdout, str) else "" + lines = raw.splitlines() + new_entries = [] + for line in lines[:100]: + if "\t" not in line: + continue + identifier, content = line.split("\t", 1) + content = content.strip() + entry = { + "type": "text", + "identifier": identifier, + "raw": line, + "content": content, + } + if "binary data" in content: + entry["type"] = "image" + new_entries.append(entry) + with self._history_lock: + self._history.clear() + self._history.extend(new_entries) + + def _filter_entries(self, query: str) -> list[dict]: + q = query.strip().lower() + with self._history_lock: + if not q: + return [ + e + for e in self._history + if not self._is_toml_data(e.get("content", "")) + ] + return [ + e + for e in self._history + if q in e.get("content", "").lower() + and not self._is_toml_data(e.get("content", "")) + ] + + def _build_results( + self, entries: list[dict], query: str = "" + ) -> list[SearchResult]: + results: list[SearchResult] = [] + for entry in entries[: self.MAX_RESULTS]: + content = entry["content"] + is_image = entry["type"] == "image" + render_type = "clipboard_image" if is_image else "clipboard_text" + display = content[:50] + "..." if len(content) > 50 else content + + metadata: dict[str, Any] = { + "raw": entry["raw"], + "identifier": entry["identifier"], + "type": entry["type"], + "image_path": "", + } + + if is_image: + image_path = entry.get("path", "") + if not image_path or not os.path.exists(image_path): + image_path = ( + self._cache_image(entry["raw"], entry["identifier"]) or "" + ) + if image_path: + entry["path"] = image_path + if image_path: + metadata["image_path"] = image_path + + results.append( + SearchResult( + id=f"clip_{entry['identifier']}", + title="Image" if is_image and not image_path else display, + subtitle=content[:80], + icon_name="edit-paste-symbolic", + score=self._score_entry(entry, query), + render_type=render_type, + plugin_id=self.id, + action=self._make_action(entry), + metadata=metadata, + ) + ) + return results + + @staticmethod + def _score_entry(entry: dict, query: str) -> float: + if not query.strip(): + return 80.0 + content_lower = entry["content"].lower() + q = query.lower() + pos = content_lower.find(q) + if pos == -1: + return 60.0 + if pos == 0: + return 100.0 + return max(60.0, 95.0 - pos * 0.5) + + @staticmethod + def _cliphist_decode(raw: str) -> bytes | None: + result = run_command(["cliphist", "decode"], input=raw.encode(), text=False) + return result.stdout if isinstance(result.stdout, (bytes, bytearray)) else None + + def _cache_image(self, raw_data: str, identifier: str) -> str | None: + img_path = os.path.join(self._cache_dir, f"{identifier}.png") + if os.path.exists(img_path): + return img_path + decoded = self._cliphist_decode(raw_data) + if not decoded: + return None + with open(img_path, "wb") as f: + f.write(decoded) + return str(img_path) + + def _make_action(self, entry: dict): + def _action(): + if entry["type"] == "image" and entry.get("path"): + copy_image(entry["path"]) + else: + decoded = self._cliphist_decode(entry["raw"]) + if not decoded: + return + copy_text(decoded.decode()) + GLib.timeout_add(5, lambda: (trigger_paste_shortcut(), False)[1]) + + return _action + + @staticmethod + def _is_toml_data(content: str) -> bool: + lines = content.strip().splitlines() + if len(lines) < 2: + return False + bracket_lines = sum( + 1 for ln in lines if ln.strip().startswith("[") and "]" in ln + ) + equals_lines = sum( + 1 + for ln in lines + if "=" in ln and ln.strip() and not ln.strip().startswith("[") + ) + return bracket_lines >= 1 and equals_lines >= 2 + + +PLUGIN = ClipboardPlugin diff --git a/src/window/spotlight/plugins/emoji.py b/src/window/spotlight/plugins/emoji.py new file mode 100644 index 0000000..f746026 --- /dev/null +++ b/src/window/spotlight/plugins/emoji.py @@ -0,0 +1,225 @@ +import contextlib +from pathlib import Path +from typing import Any +from urllib.request import Request, urlopen + +from fabric.utils import GLib, logger + +import shared.data as data +from utils.functions import copy_text, read_json_file, trigger_paste_shortcut +from window.spotlight.api import PluginContext, SearchResult, SpotlightPlugin + + +class EmojiPlugin(SpotlightPlugin): + id = "emoji" + name = "Emoji" + icon = "face-smile-symbolic" + keywords = ["emoji", "em"] + searchable = True + priority = 70 + + def __init__(self, context: PluginContext): + super().__init__(context) + self._emoji_data: dict[str, dict[str, Any]] = {} + + def initialize(self) -> None: + self._ensure_data() + + def cleanup(self) -> None: + self._emoji_data.clear() + + def release_memory(self) -> None: + self._emoji_data.clear() + + def _ensure_data(self): + if self._emoji_data: + return + cldr = self._load_cldr_annotations() + self._emoji_data = cldr or {} + logger.info(f"[Emoji] Loaded {len(self._emoji_data)} emojis") + + def detect(self, text: str) -> bool: + s = text.strip() + if not s: + return False + for ch in s: + cp = ord(ch) + if ( + 0x1F600 <= cp <= 0x1F64F + or 0x1F300 <= cp <= 0x1F5FF + or 0x1F680 <= cp <= 0x1F6FF + or 0x1F1E0 <= cp <= 0x1F1FF + or 0x2600 <= cp <= 0x27BF + or 0xFE00 <= cp <= 0xFE0F + or 0x1F900 <= cp <= 0x1F9FF + or cp == 0x200D + or cp == 0x20E3 + or 0xE0020 <= cp <= 0xE007F + ): + return True + return False + + def search(self, query: str, token: Any) -> list[SearchResult]: + q = query.strip() + if not q: + return [] + + for prefix in ("emoji ", "em "): + if q.lower().startswith(prefix): + q = q[len(prefix) :].strip() + break + if q.lower() in ("emoji", "em"): + q = "" + + if not q: + return [] + + self._ensure_data() + if not self._emoji_data: + logger.warning("[Emoji] No emoji data available") + return [] + return self._query_emojis(q) + + def _query_emojis(self, text: str) -> list[SearchResult]: + query = text.lower().strip() + scored: list[tuple[str, dict, float]] = [] + + for emoji_char, info in self._emoji_data.items(): + relevance = 0.0 + name = info.get("name", "").lower() + keywords = info.get("keywords", []) + + if query == emoji_char: + relevance = 1.0 + elif query == name: + relevance = 0.95 + elif name.startswith(query): + relevance = 0.9 + elif query in name: + relevance = 0.8 + 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: + scored.append((emoji_char, info, relevance)) + + if not scored: + return [] + + scored.sort(key=lambda x: x[2], reverse=True) + + results: list[SearchResult] = [] + for emoji_char, info, relevance in scored[:20]: + name = info.get("name", "Unknown").replace("_", " ").title() + keyword_list = info.get("keywords", []) + + results.append( + SearchResult( + id=emoji_char, + title=emoji_char, + subtitle=name, + icon_name="", + score=relevance * 100, + render_type="emoji", + action=lambda e=emoji_char: self._copy_to_clipboard(e), + plugin_name=self.name, + plugin_id=self.id, + metadata={ + "emoji": emoji_char, + "name": name, + "keywords": keyword_list, + }, + ) + ) + + return results + + def _copy_to_clipboard(self, emoji_char: str) -> None: + copy_text(emoji_char) + GLib.timeout_add(5, lambda: (trigger_paste_shortcut(), False)[1]) + + def handle_external(self, command: str, args: str) -> None: + pass + + def _load_cldr_annotations(self) -> dict[str, dict[str, Any]] | None: + try: + lang = "en" + annotations_path = self._get_annotations_path(lang) + + if not annotations_path.exists(): + url = ( + "https://raw.githubusercontent.com/unicode-org/cldr-json/main/" + f"cldr-json/cldr-annotations-full/annotations/{lang}/annotations.json" + ) + logger.info(f"[Emoji] Downloading CLDR annotations from {url}") + if not self._download_annotations(url, annotations_path): + logger.error("[Emoji] Failed to download CLDR annotations") + return None + + raw = self._read_annotations_file(annotations_path) + if raw is None: + logger.error("[Emoji] Failed to read annotations file") + return None + + result = self._normalize_annotations(raw) or None + if result: + logger.info(f"[Emoji] Normalized {len(result)} emojis") + else: + logger.error("[Emoji] Normalization returned empty") + return result + except Exception as e: + logger.error(f"[Emoji] Error loading CLDR annotations: {e}") + return None + + def _get_annotations_path(self, lang: str) -> Path: + cache_dir = Path(data.CACHE_DIR) / "emoji" + cache_dir.mkdir(parents=True, exist_ok=True) + return cache_dir / f"annotations_full_{lang}.json" + + def _download_annotations(self, url: str, dest: Path) -> bool: + headers = {"User-Agent": "Mozilla/5.0"} + try: + req = Request(url, headers=headers) + with urlopen(req, timeout=10) as response: + if response.status == 200: + dest.write_bytes(response.read()) + return True + return False + except Exception as e: + logger.error(f"[Emoji] Download error: {e}") + return False + + def _read_annotations_file(self, path: Path) -> dict[str, Any] | None: + raw = read_json_file(str(path)) + if raw: + return raw + with contextlib.suppress(OSError): + path.unlink() + return None + + def _normalize_annotations(self, raw: dict[str, Any]) -> dict[str, dict[str, Any]]: + annotations = ( + raw.get("annotations", {}).get("annotations", {}) + if isinstance(raw, dict) + else {} + ) + + if not annotations: + return {} + + normalized: dict[str, dict[str, Any]] = {} + for emoji_char, info in annotations.items(): + if not isinstance(info, dict): + continue + tts_list = info.get("tts", []) or [] + default_list = info.get("default", []) or [] + + title = tts_list[0] if tts_list else "" + normalized[emoji_char] = { + "name": title, + "keywords": default_list, + } + + return {k: v for k, v in normalized.items() if v.get("name")} diff --git a/src/window/spotlight/plugins/power_menu.py b/src/window/spotlight/plugins/power_menu.py new file mode 100644 index 0000000..fe959cd --- /dev/null +++ b/src/window/spotlight/plugins/power_menu.py @@ -0,0 +1,138 @@ +from typing import Any + +from fabric.utils import exec_shell_command_async + +from window.spotlight.api import SearchResult, SpotlightPlugin + + +class PowerMenuPlugin(SpotlightPlugin): + id = "power_menu" + name = "Power Menu" + icon = "system-shutdown-symbolic" + keywords = ["powermenu", "pm"] + searchable = True + priority = 95 + + POWER_OPTIONS = [ + { + "id": "shutdown", + "name": "Shutdown", + "icon": "system-shutdown", + "command": "systemctl poweroff", + "description": "Shutdown the system", + }, + { + "id": "restart", + "name": "Restart", + "icon": "system-reboot", + "command": "systemctl reboot", + "description": "Restart the system", + }, + { + "id": "suspend", + "name": "Suspend", + "icon": "system-suspend", + "command": "systemctl suspend", + "description": "Suspend the system to RAM", + }, + { + "id": "hibernate", + "name": "Hibernate", + "icon": "system-hibernate", + "command": "systemctl hibernate", + "description": "Hibernate the system to disk", + }, + { + "id": "logout", + "name": "Logout", + "icon": "system-log-out", + "command": "pkill -SIGTERM Hyprland", + "description": "Logout from the current session", + }, + { + "id": "lock", + "name": "Lock Screen", + "icon": "system-lock-screen", + "command": "hyprlock", + "description": "Lock the screen", + }, + ] + + def initialize(self) -> None: + pass + + def cleanup(self) -> None: + pass + + def search(self, query: str, token: Any) -> list[SearchResult]: + q = query.lower().strip() + if not q: + return [] + + for prefix in ("powermenu ", "pm "): + if q.startswith(prefix): + q = q[len(prefix) :].strip() + break + if q in ("powermenu", "pm"): + q = "" + + if not q: + return [ + SearchResult( + id=f"power_{option['id']}", + title=option["name"], + subtitle=option["description"], + icon_name=option["icon"], + score=100.0 - i, + render_type="power", + action=lambda cmd=option["command"]: exec_shell_command_async(cmd), + metadata={"power_option": option}, + ) + for i, option in enumerate(self.POWER_OPTIONS) + ] + + results: list[SearchResult] = [] + + for option in self.POWER_OPTIONS: + score = 0.0 + name = option["name"].lower() + desc = option["description"].lower() + + if q == name: + score = 100.0 + elif name.startswith(q): + score = 90.0 + elif len(q) >= 3 and q in name: + score = 80.0 + elif len(q) >= 3 and q in desc: + score = 60.0 + + if score > 0: + results.append( + SearchResult( + id=f"power_{option['id']}", + title=option["name"], + subtitle=option["description"], + icon_name=option["icon"], + score=score, + render_type="power", + action=lambda cmd=option["command"]: exec_shell_command_async( + cmd + ), + metadata={"power_option": option}, + ) + ) + + results.sort(key=lambda r: r.score, reverse=True) + return results + + def handle_external(self, command: str, args: str) -> None: + if not args: + return + for option in self.POWER_OPTIONS: + if args.lower() == option["name"].lower(): + exec_shell_command_async(option["command"]) + return + + +PLUGIN = PowerMenuPlugin diff --git a/src/window/spotlight/plugins/query.py b/src/window/spotlight/plugins/query.py new file mode 100644 index 0000000..347ad93 --- /dev/null +++ b/src/window/spotlight/plugins/query.py @@ -0,0 +1,134 @@ +import urllib.parse +from typing import Any + +from utils.functions import spawn_detached +from window.spotlight.api import PluginContext, SearchResult, SpotlightPlugin + + +class QueryPlugin(SpotlightPlugin): + id = "query" + name = "Query" + icon = "system-search-symbolic" + keywords = ["gg", "yt", "lk"] + keyword_icons = { + "gg": "edit-find-symbolic", + "yt": "video-x-generic-symbolic", + "lk": "insert-link-symbolic", + } + searchable = False + priority = 100 + + def __init__(self, context: PluginContext): + super().__init__(context) + + def initialize(self) -> None: + pass + + def cleanup(self) -> None: + pass + + def release_memory(self) -> None: + pass + + def search(self, query: str, token: Any) -> list[SearchResult]: + q = query.strip() + if not q: + return [] + + bare = q.lower() + if bare in ("gg", "yt", "lk"): + hints = { + "gg": ("Google Search", "Type 'gg ' to search"), + "yt": ("YouTube Search", "Type 'yt ' to search"), + "lk": ("Open Link", "Type 'lk ' to open"), + } + title, subtitle = hints[bare] + return [ + SearchResult( + id=f"query_{bare}_hint", + title=title, + subtitle=subtitle, + icon_name=self.keyword_icons.get(bare, self.icon), + score=100.0, + render_type="default", + ) + ] + + if q.lower().startswith("gg "): + search_q = q[3:].strip() + if not search_q: + return [] + return [ + SearchResult( + id="query_gg", + title=f"Google: {search_q}", + subtitle="Open in browser", + icon_name="system-search-symbolic", + score=100.0, + render_type="default", + action=lambda: self._google_search(search_q), + ) + ] + if q.lower().startswith("yt "): + search_q = q[3:].strip() + if not search_q: + return [] + return [ + SearchResult( + id="query_yt", + title=f"YouTube: {search_q}", + subtitle="Open in browser", + icon_name="system-search-symbolic", + score=100.0, + render_type="default", + action=lambda: self._youtube_search(search_q), + ) + ] + if q.lower().startswith("lk "): + link = q[3:].strip() + if not link: + return [] + if not link.startswith("http"): + link = "https://" + link + return [ + SearchResult( + id="query_lk", + title=f"Open: {link}", + subtitle="Open in browser", + icon_name="system-search-symbolic", + score=100.0, + render_type="default", + action=lambda: self._link_open(link), + ) + ] + return [] + + def on_submit(self, query: str) -> None: + q = query.strip() + if q.startswith("gg "): + self._google_search(q[3:].strip()) + elif q.startswith("yt "): + self._youtube_search(q[3:].strip()) + elif q.startswith("lk "): + self._link_open(q.split(" ", 1)[1].strip()) + + def _google_search(self, query: str) -> None: + url = f"https://www.google.com/search?q={urllib.parse.quote(query)}" + spawn_detached(["xdg-open", url]) + + def _youtube_search(self, query: str) -> None: + url = ( + f"https://www.youtube.com/results?search_query={urllib.parse.quote(query)}" + ) + spawn_detached(["xdg-open", url]) + + def _link_open(self, link: str) -> None: + if not link.startswith(("http://", "https://")): + link = "https://" + link + spawn_detached(["xdg-open", link]) + + def handle_external(self, command: str, args: str) -> None: + pass + + +PLUGIN = QueryPlugin diff --git a/src/window/spotlight/plugins/wallpapers.py b/src/window/spotlight/plugins/wallpapers.py new file mode 100644 index 0000000..e290de0 --- /dev/null +++ b/src/window/spotlight/plugins/wallpapers.py @@ -0,0 +1,118 @@ +import gc +import os +import random +from typing import Any + +from services.wallpaper import WallpaperService, create_thumbnail +from shared.data import WALLPAPERS_THUMBNAILS_SIZE +from utils.functions import fuzzy_filter +from window.spotlight.api import SearchResult, SpotlightPlugin + +MAX_THUMBS_PER_SEARCH = 5 + + +class WallpaperPlugin(SpotlightPlugin): + id = "wallpaper" + name = "Wallpaper" + icon = "wallpaper" + keywords = ["wall", "wr"] + searchable = True + priority = 85 + + def __init__(self, context): + super().__init__(context) + self._wallpaper_service: WallpaperService | None = None + + def initialize(self) -> None: + self._wallpaper_service = WallpaperService() + + def cleanup(self) -> None: + self._wallpaper_service = None + gc.collect() + + def release_memory(self) -> None: + if self._wallpaper_service: + self._wallpaper_service.invalidate_cache() + + def search(self, query: str, token: Any) -> list[SearchResult]: + service = self._wallpaper_service + if not service: + return [] + + q = query.strip() + for prefix in ("wall ", "wr "): + if q.lower().startswith(prefix): + q = q[len(prefix) :].strip() + break + if q.lower() in ("wall", "wr"): + q = "" + + all_files = service.get_all_wallpapers() + if not all_files: + return [] + + if q: + filtered = fuzzy_filter(q, all_files, key=os.path.basename, limit=50) + else: + filtered = all_files[:50] + + results: list[SearchResult] = [] + thumbs_created = 0 + + for image_path in filtered: + if token.is_cancelled or not self._wallpaper_service: + return [] + + thumbnail_path = service.get_thumbnail_path(image_path) + + if not service.has_thumbnail(image_path): + if thumbs_created >= MAX_THUMBS_PER_SEARCH: + continue + try: + create_thumbnail( + image_path, thumbnail_path, WALLPAPERS_THUMBNAILS_SIZE + ) + thumbs_created += 1 + except Exception: + continue + + filename = os.path.basename(image_path) + results.append( + SearchResult( + id=f"wall_{filename}", + title=filename, + subtitle=image_path, + icon_name="", + score=80.0, + render_type="wallpaper", + action=lambda p=image_path: self._set_wallpaper(p), + metadata={ + "image_path": image_path, + "thumbnail_path": thumbnail_path, + }, + ) + ) + + return results + + def _set_wallpaper(self, path: str) -> None: + service = self._wallpaper_service + if service: + service.set_wallpaper(path) + service.invalidate_cache() + + def apply_random_wallpaper(self) -> None: + service = self._wallpaper_service + if not service: + return + service.invalidate_cache() + files = service.get_all_wallpapers() + if files: + service.set_wallpaper(random.choice(files)) + + def handle_external(self, command: str, args: str) -> None: + if command == "wr": + self.apply_random_wallpaper() + + +PLUGIN = WallpaperPlugin