feat(spotlight): add plugin system with core engine and examples
- Plugin API (plugin.py, context.py, result.py) - Core engine (search.py, manager.py, registry.py, loader.py) - Built-in plugins: app, calculator, clipboard, emoji, power_menu, query, wallpapers - Spotlight docs and example plugins - Spotlight CSS styling
This commit is contained in:
24 files changed
+3130
-20
No files matched your search
+533
@@ -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 <query>` 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 <query>` -> 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/<plugin_name>/` using `uv venv`.
|
||||
3. It runs `uv pip install -r requirements.txt --python <venv>/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'`
|
||||
@@ -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
|
||||
@@ -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 <city>' 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
|
||||
@@ -0,0 +1 @@
|
||||
requests>=2.28
|
||||
@@ -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),
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
@@ -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"]
|
||||
@@ -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
|
||||
@@ -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.
|
||||
"""
|
||||
@@ -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
|
||||
@@ -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",
|
||||
]
|
||||
@@ -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 <plugins_dir>/.venvs/<plugin_name>/
|
||||
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()
|
||||
@@ -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}")
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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,
|
||||
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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")}
|
||||
@@ -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
|
||||
@@ -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 <query>' to search"),
|
||||
"yt": ("YouTube Search", "Type 'yt <query>' to search"),
|
||||
"lk": ("Open Link", "Type 'lk <url>' 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
|
||||
@@ -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
|
||||
Reference in new issue
Block a user