✨ feat(osd): add OSD for brightness, volume, and microphone controls

This commit is contained in:
sankalp committed 2025-06-16 18:58:34 +05:45
1 parent 99421153ab
commit 90ebffe6dd
15 files changed
+520 -1231

No files matched your search

-52
View File
@@ -1,52 +0,0 @@
"""
Plugin system for the launcher module.
Allows extending search functionality with different content types.
"""
from abc import ABC, abstractmethod
from typing import List, Dict, Any, Optional
class LauncherPlugin(ABC):
"""Base class for launcher plugins"""
@property
@abstractmethod
def name(self) -> str:
"""Name of the plugin"""
pass
@property
@abstractmethod
def category(self) -> str:
"""Category this plugin belongs to"""
pass
@property
def icon_name(self) -> str:
"""Icon name for this plugin's category"""
return "application-x-executable-symbolic"
@abstractmethod
def search(self, query: str) -> List[Dict[str, Any]]:
"""
Search for items matching the query
Returns a list of dictionaries with at least:
- title: str
- description: str
- icon_name: str (optional)
- action: callable
"""
pass
def get_action_items(self, query: str) -> List[Dict[str, Any]]:
"""
Get quick action items for the query
Returns a list of dictionaries with at least:
- title: str
- icon_name: str
- action: callable
"""
return []
-78
View File
@@ -1,78 +0,0 @@
from typing import List, Dict, Any
from fabric.utils import get_desktop_applications, DesktopApp, exec_shell_command_async
from . import LauncherPlugin
class ApplicationsPlugin(LauncherPlugin):
"""Plugin for searching desktop applications"""
@property
def name(self) -> str:
return "Applications"
@property
def category(self) -> str:
return "Applications"
@property
def icon_name(self) -> str:
return "application-x-executable-symbolic"
def search(self, query: str) -> List[Dict[str, Any]]:
all_apps = get_desktop_applications()
# Filter apps based on query
filtered_apps = [
app
for app in all_apps
if query.casefold()
in (
(app.display_name or "")
+ (" " + app.name + " ")
+ (app.generic_name or "")
).casefold()
]
# Sort by relevance - exact matches first
if query:
exact_matches = []
partial_matches = []
for app in filtered_apps:
name = (app.display_name or app.name or "").casefold()
if name == query.casefold() or name.startswith(query.casefold()):
exact_matches.append(app)
else:
partial_matches.append(app)
filtered_apps = exact_matches + partial_matches
# Convert to result format
results = []
for app in filtered_apps:
results.append(
{
"title": app.display_name or app.name or "Unknown",
"description": app.description or "Application",
"icon": app, # Pass the whole app object for icon rendering
"action": lambda app=app: self.launch_app(app),
"app": app, # Store the original app object
}
)
return results
def launch_app(self, app: DesktopApp):
command = (
" ".join([arg for arg in app.command_line.split() if "%" not in arg])
if app.command_line
else None
)
(
exec_shell_command_async(
f"hyprctl dispatch exec -- {command}",
lambda *_: print(f"Launched {app.name}"),
)
if command
else None
)
-54
View File
@@ -1,54 +0,0 @@
import re
from typing import List, Dict, Any
from . import LauncherPlugin
class CalculatorPlugin(LauncherPlugin):
"""Plugin for calculator functionality"""
@property
def name(self) -> str:
return "Calculator"
@property
def category(self) -> str:
return "Utilities"
@property
def icon_name(self) -> str:
return "accessories-calculator-symbolic"
def search(self, query: str) -> List[Dict[str, Any]]:
# Only process if query looks like a math expression
if not (query.isdigit() or re.match(r"^[\d+\-*/().]+$", query)):
return []
try:
result = eval(query)
return [
{
"title": f"{query} = {result}",
"description": "Calculator result",
"icon_name": "accessories-calculator-symbolic",
"action": lambda: None, # No action needed
}
]
except:
return []
def get_action_items(self, query: str) -> List[Dict[str, Any]]:
# Only process if query looks like a math expression
if not (query.isdigit() or re.match(r"^[\d+\-*/().]+$", query)):
return []
try:
result = eval(query)
return [
{
"title": f"Calculate: {query} = {result}",
"icon_name": "accessories-calculator-symbolic",
"action": lambda: None, # No action needed
}
]
except:
return []
-259
View File
@@ -1,259 +0,0 @@
import re
import subprocess
from typing import List, Dict, Any
import gi
gi.require_version('GdkPixbuf', '2.0')
from gi.repository import GdkPixbuf
from . import LauncherPlugin
class CliphistPlugin(LauncherPlugin):
"""Plugin for clipboard history functionality using cliphist"""
@property
def name(self) -> str:
return "Clipboard History"
@property
def category(self) -> str:
return "Utilities"
@property
def icon_name(self) -> str:
return "edit-paste-symbolic"
def search(self, query: str) -> List[Dict[str, Any]]:
if not query:
return []
query_lower = query.lower()
# Show clipboard history when user types "clip" (like Albert launcher)
if query_lower == "clip" or query_lower == "clipboard":
return self._get_clipboard_history()
# Also show clipboard history if query starts with "clip " for further filtering
if query_lower.startswith("clip ") and len(query) > 5:
search_term = query[5:].strip() # Remove "clip " prefix
return self._get_clipboard_history(search_term)
return []
def _get_clipboard_history(self, search_term: str = "") -> List[Dict[str, Any]]:
"""Get clipboard history, optionally filtered by search term"""
try:
# Get clipboard history from cliphist
result = subprocess.run(
["cliphist", "list"],
capture_output=True,
text=True,
timeout=5
)
if result.returncode != 0:
return []
entries = result.stdout.strip().split('\n')
if not entries or entries == ['']:
return []
results = []
search_lower = search_term.lower() if search_term else ""
for entry in entries[:50]: # Limit to first 50 entries for performance
if not entry.strip():
continue
# Parse entry format: "ID\tCONTENT"
parts = entry.split('\t', 1)
if len(parts) != 2:
continue
entry_id, content = parts
content = content.strip()
# Check if this is an image
is_image = self._is_image_data(content)
# Skip binary data entries that aren't images
if content.startswith('[[') and 'binary data' in content and not is_image:
continue
# If we have a search term, filter by it
if search_term and search_lower not in content.lower():
continue
# Prepare display content and icon
if is_image:
display_content = "[Image]"
# Try to load image preview
image_pixbuf = self._load_image_preview(entry_id)
if image_pixbuf:
result_item = {
"title": display_content,
"image_pixbuf": image_pixbuf, # Custom field for image preview
"action": lambda eid=entry_id: self.copy_to_clipboard(eid),
"clip_id": entry_id,
}
else:
result_item = {
"title": display_content,
"icon_name": "image-x-generic-symbolic",
"action": lambda eid=entry_id: self.copy_to_clipboard(eid),
"clip_id": entry_id,
}
else:
# Truncate long content for display
display_content = content
if len(display_content) > 80:
display_content = display_content[:77] + "..."
# Replace newlines with spaces for display
display_content = display_content.replace('\n', ' ').replace('\r', ' ')
result_item = {
"title": display_content,
"icon_name": "edit-paste-symbolic",
"action": lambda eid=entry_id: self.copy_to_clipboard(eid),
"clip_id": entry_id,
}
results.append(result_item)
# # Limit results to prevent overwhelming the UI
# if len(results) >= 15:
# break
# Sort results by relevance if we have a search term
if search_term:
def sort_key(item):
title = item["title"].lower()
if title.startswith(search_lower):
return 0 # Starts with query
elif search_lower in title:
return 1 # Contains query
else:
return 2 # Other matches
results.sort(key=sort_key)
return results
except (subprocess.CalledProcessError, subprocess.TimeoutExpired, FileNotFoundError):
return []
except Exception as e:
print(f"Error in cliphist plugin: {e}")
return []
def copy_to_clipboard(self, entry_id: str):
"""Copy clipboard history entry back to clipboard"""
try:
# Use cliphist decode to get the content and pipe it to wl-copy
decode_process = subprocess.Popen(
["cliphist", "decode", entry_id],
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL
)
copy_process = subprocess.Popen(
["wl-copy"],
stdin=decode_process.stdout,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL
)
decode_process.stdout.close() # Allow decode_process to receive SIGPIPE
copy_process.wait(timeout=2)
decode_process.wait(timeout=2)
print(f"Copied clipboard history entry {entry_id}")
except (subprocess.CalledProcessError, subprocess.TimeoutExpired, FileNotFoundError):
try:
# Fallback to xclip
decode_process = subprocess.Popen(
["cliphist", "decode", entry_id],
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
text=True
)
content, _ = decode_process.communicate(timeout=2)
if decode_process.returncode == 0:
subprocess.Popen(
["xclip", "-selection", "clipboard"],
input=content,
text=True,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL
)
print(f"Copied clipboard history entry {entry_id} (xclip)")
except (subprocess.CalledProcessError, subprocess.TimeoutExpired, FileNotFoundError):
print(f"Could not copy clipboard history entry {entry_id}")
except Exception as e:
print(f"Error copying clipboard entry: {e}")
def _is_image_data(self, content):
"""Determine if clipboard content is likely an image"""
return (
content.startswith("data:image/") or
content.startswith("\x89PNG") or
content.startswith("GIF8") or
content.startswith("\xff\xd8\xff") or
re.match(r'^\s*<img\s+', content) is not None or
"binary" in content.lower() and any(ext in content.lower() for ext in ["jpg", "jpeg", "png", "bmp", "gif"])
)
def _load_image_preview(self, item_id):
"""Load image preview for clipboard item"""
try:
result = subprocess.run(
["cliphist", "decode", item_id],
capture_output=True,
check=True,
timeout=2
)
# Try to load the image data
loader = GdkPixbuf.PixbufLoader()
loader.write(result.stdout)
loader.close()
pixbuf = loader.get_pixbuf()
if pixbuf:
# Scale to thumbnail size
width, height = pixbuf.get_width(), pixbuf.get_height()
max_size = 128 # Match launcher icon size
if width > height:
new_width = max_size
new_height = int(height * (max_size / width))
else:
new_height = max_size
new_width = int(width * (max_size / height))
scaled_pixbuf = pixbuf.scale_simple(
new_width, new_height, GdkPixbuf.InterpType.BILINEAR
)
return scaled_pixbuf
except Exception as e:
pass
return None
def get_action_items(self, query: str) -> List[Dict[str, Any]]:
"""Get quick action items for clipboard history"""
if not query or len(query) < 2:
return []
return [
{
"title": f'Search clipboard history for "{query}"',
"icon_name": "edit-paste-symbolic",
"action": lambda: None, # No specific action needed
}
]
-122
View File
@@ -1,122 +0,0 @@
import json
import os
import subprocess
from typing import List, Dict, Any
from . import LauncherPlugin
class EmojiPlugin(LauncherPlugin):
"""Plugin for emoji search and clipboard functionality"""
def __init__(self):
self._emojis = {}
self._load_emojis()
def _load_emojis(self):
"""Load emoji data from the JSON file"""
try:
emoji_file_path = os.path.expanduser("~/Modus/config/emoji.json")
with open(emoji_file_path, "r", encoding="utf-8") as f:
self._emojis = json.load(f)
except Exception as e:
print(f"Failed to load emoji data: {e}")
self._emojis = {}
@property
def name(self) -> str:
return "Emoji"
@property
def category(self) -> str:
return "Utilities"
@property
def icon_name(self) -> str:
return "face-smile-symbolic"
def search(self, query: str) -> List[Dict[str, Any]]:
if not query or len(query) < 2: # Require at least 2 characters
return []
results = []
query_lower = query.lower()
# Search through emojis
for emoji_char, emoji_data in self._emojis.items():
# Search in name, slug, and group
searchable_text = (
emoji_data.get("name", "").lower()
+ " "
+ emoji_data.get("slug", "").lower()
+ " "
+ emoji_data.get("group", "").lower()
)
if query_lower in searchable_text:
results.append(
{
"title": emoji_data.get('name', 'Unknown'),
"description": f"{emoji_data.get('group', 'Unknown')}",
"emoji_icon": emoji_char, # Store the emoji character for display
"action": lambda emoji=emoji_char: self.copy_emoji(emoji),
}
)
# Limit results to prevent overwhelming the UI
if len(results) >= 20:
break
# Sort results by relevance (exact matches first)
def sort_key(item):
name = item["title"].lower()
# Check if query matches the emoji name
if name.startswith(query_lower):
return 0 # Name starts with query
elif query_lower in name:
return 1 # Partial name match
else:
return 2 # Other matches
results.sort(key=sort_key)
return results
def copy_emoji(self, emoji: str):
"""Copy emoji to clipboard using wl-copy"""
# Use a fire-and-forget approach to avoid blocking
try:
# Start wl-copy in the background and don't wait for it
subprocess.Popen(
["wl-copy", emoji], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL
)
print(f"Copied emoji to clipboard: {emoji}")
except FileNotFoundError:
try:
# Fallback to xclip
process = subprocess.Popen(
["xclip", "-selection", "clipboard"],
stdin=subprocess.PIPE,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
text=True,
)
process.stdin.write(emoji)
process.stdin.close()
print(f"Copied emoji to clipboard (xclip): {emoji}")
except (FileNotFoundError, OSError):
print(f"Could not copy emoji to clipboard: {emoji}")
except OSError:
print(f"Could not copy emoji to clipboard: {emoji}")
def get_action_items(self, query: str) -> List[Dict[str, Any]]:
"""Get quick action items for emoji search"""
if not query or len(query) < 2:
return []
# Return a quick action to search for emojis
return [
{
"title": f'Search emojis for "{query}"',
"icon_name": "face-smile-symbolic",
"action": lambda: None, # No specific action needed
}
]
-202
View File
@@ -1,202 +0,0 @@
import subprocess
from typing import List, Dict, Any
import utils.icons as icons
from . import LauncherPlugin
class PowerMenuPlugin(LauncherPlugin):
"""Plugin for system power management functionality"""
@property
def name(self) -> str:
return "Power Menu"
@property
def category(self) -> str:
return "System"
@property
def icon_name(self) -> str:
return "system-shutdown-symbolic"
def search(self, query: str) -> List[Dict[str, Any]]:
if not query:
return []
query_lower = query.lower()
# Show power menu when user types "sys", "system", "power", etc.
power_triggers = ["sys", "system", "power", "shutdown", "restart", "reboot", "logout", "suspend", "sleep", "lock"]
if any(query_lower.startswith(trigger) for trigger in power_triggers):
return self._get_power_options(query_lower)
return []
def _get_power_options(self, query: str = "") -> List[Dict[str, Any]]:
"""Get available power management options"""
power_options = [
{
"title": "Shutdown",
"description": "Turn off the computer completely",
"icon_markup": icons.shutdown,
"action": self.shutdown_system,
"keywords": ["shutdown", "power off", "turn off", "sys", "halt"],
},
{
"title": "Restart",
"description": "Restart the computer",
"icon_markup": icons.reboot,
"action": self.restart_system,
"keywords": ["restart", "reboot", "sys"],
},
{
"title": "Logout",
"description": "End current user session",
"icon_markup": icons.logout,
"action": self.logout_session,
"keywords": ["logout", "sign out", "end session", "sys"],
},
{
"title": "Suspend",
"description": "Suspend to RAM (sleep mode)",
"icon_markup": icons.suspend,
"action": self.suspend_system,
"keywords": ["suspend", "sleep", "sys"],
},
{
"title": "Lock Screen",
"description": "Lock the current session",
"icon_markup": icons.lock,
"action": self.lock_screen,
"keywords": ["lock", "lock screen", "sys"],
},
]
# Filter options based on query if it's more specific
if len(query) > 3: # More than just "sys"
filtered_options = []
for option in power_options:
# Check if query matches any keywords
if any(query in keyword for keyword in option["keywords"]):
filtered_options.append(option)
if filtered_options:
power_options = filtered_options
# Convert to launcher result format
results = []
for option in power_options:
results.append({
"title": option["title"],
"description": option["description"],
"icon_markup": option["icon_markup"], # Use markup for custom icons
"action": option["action"],
"power_option": True, # Mark as power option for special handling
})
return results
def shutdown_system(self):
"""Shutdown the system"""
print("Initiating system shutdown...")
try:
# Try systemctl first (systemd)
subprocess.run(["systemctl", "poweroff"], check=True)
except (subprocess.CalledProcessError, FileNotFoundError):
try:
# Fallback to shutdown command
subprocess.run(["shutdown", "-h", "now"], check=True)
except (subprocess.CalledProcessError, FileNotFoundError):
print("Could not shutdown system - no suitable command found")
def restart_system(self):
"""Restart the system"""
print("Initiating system restart...")
try:
# Try systemctl first (systemd)
subprocess.run(["systemctl", "reboot"], check=True)
except (subprocess.CalledProcessError, FileNotFoundError):
try:
# Fallback to shutdown command
subprocess.run(["shutdown", "-r", "now"], check=True)
except (subprocess.CalledProcessError, FileNotFoundError):
print("Could not restart system - no suitable command found")
def logout_session(self):
"""Logout from current session"""
try:
# Try different logout methods based on desktop environment
# Hyprland
subprocess.run(["hyprctl", "dispatch", "exit"], check=True)
except (subprocess.CalledProcessError, FileNotFoundError):
try:
# GNOME
subprocess.run(["gnome-session-quit", "--logout", "--no-prompt"], check=True)
except (subprocess.CalledProcessError, FileNotFoundError):
try:
# KDE
subprocess.run(["qdbus", "org.kde.ksmserver", "/KSMServer", "logout", "0", "0", "0"], check=True)
except (subprocess.CalledProcessError, FileNotFoundError):
try:
# Generic X11
subprocess.run(["pkill", "-KILL", "-u", subprocess.getoutput("whoami")], check=True)
except (subprocess.CalledProcessError, FileNotFoundError):
print("Could not logout - no suitable method found")
def suspend_system(self):
"""Suspend the system"""
try:
# Try systemctl first (systemd)
subprocess.run(["systemctl", "suspend"], check=True)
except (subprocess.CalledProcessError, FileNotFoundError):
try:
# Fallback to pm-suspend
subprocess.run(["pm-suspend"], check=True)
except (subprocess.CalledProcessError, FileNotFoundError):
print("Could not suspend system - no suitable command found")
def lock_screen(self):
"""Lock the screen"""
try:
# Try different lock methods
# Hyprland
subprocess.run(["hyprlock"], check=True)
except (subprocess.CalledProcessError, FileNotFoundError):
try:
# swaylock (Sway/Wayland)
subprocess.run(["swaylock"], check=True)
except (subprocess.CalledProcessError, FileNotFoundError):
try:
# i3lock (X11)
subprocess.run(["i3lock"], check=True)
except (subprocess.CalledProcessError, FileNotFoundError):
try:
# GNOME
subprocess.run(["gnome-screensaver-command", "--lock"], check=True)
except (subprocess.CalledProcessError, FileNotFoundError):
try:
# KDE
subprocess.run(["qdbus", "org.freedesktop.ScreenSaver", "/ScreenSaver", "Lock"], check=True)
except (subprocess.CalledProcessError, FileNotFoundError):
print("Could not lock screen - no suitable command found")
def get_action_items(self, query: str) -> List[Dict[str, Any]]:
"""Get quick action items for power management"""
if not query or len(query) < 2:
return []
query_lower = query.lower()
# Show quick actions for system-related queries
if any(trigger in query_lower for trigger in ["sys", "power", "shutdown", "restart"]):
return [
{
"title": f'System power options for "{query}"',
"icon_name": "system-shutdown-symbolic",
"action": lambda: None, # No specific action needed
}
]
return []
-49
View File
@@ -1,49 +0,0 @@
from typing import List, Dict, Any
from fabric.utils import exec_shell_command_async
from . import LauncherPlugin
class WebSearchPlugin(LauncherPlugin):
"""Plugin for web search functionality"""
@property
def name(self) -> str:
return "Web Search"
@property
def category(self) -> str:
return "Web"
@property
def icon_name(self) -> str:
return "web-browser-symbolic"
def search(self, query: str) -> List[Dict[str, Any]]:
if not query:
return []
return [
{
"title": f'Search the web for "{query}"',
"description": "Open in default browser",
"icon_name": "web-browser-symbolic",
"action": lambda q=query: self.search_web(q),
}
]
def get_action_items(self, query: str) -> List[Dict[str, Any]]:
if not query:
return []
return [
{
"title": f'Search web for "{query}"',
"icon_name": "web-browser-symbolic",
"action": lambda q=query: self.search_web(q),
}
]
def search_web(self, query: str):
# Use xdg-open to open the default browser with a search query
search_url = f"https://www.google.com/search?q={query.replace(' ', '+')}"
exec_shell_command_async(f"xdg-open '{search_url}'", lambda *_: None)