🔧 This commit includes various changes to the launcher, such as updating colors and icons in the config files. New plugins, like 'scripts' and 'kanban', have been added along with new features in existing plugins like 'bluetooth' and 'screencapture'. Notifications have been improved with better error handling and a new service for caching notifications. The scripts plugin now supports more system utilities and commands, providing better control over the system.

This commit is contained in:
sankalp committed 2025-06-19 21:55:00 +05:45
1 parent e0691986c2
commit 761e7bfd61
13 files changed
+746 -378

No files matched your search

+9 -9
View File
@@ -1,13 +1,13 @@
$wallpaper = /home/sankalp/Pictures/wallpapers/70.jpg
$wallpaper = /home/sankalp/Pictures/wallpapers/173.png
$background = 121318
$foreground = e2e2e9
$foreground = e3e1e9
$primary = aec6ff
$secondary = bfc6dc
$tertiary = dfbbde
$primary = bac3ff
$secondary = c3c5dd
$tertiary = e5bad8
$surface = 121318
$surface_bright = 37393e
$outline = 8e9099
$surface_bright = 38393f
$outline = 90909a
$error = ff5945
$shadow = 000000
@@ -15,13 +15,13 @@ $shadow = 000000
$red = ffb2b9
$green = 95d5a7
$yellow = b8cf84
$blue = b6c4ff
$blue = bcc2ff
$magenta = e4b7f3
$cyan = 82d3e2
$white = 82d3e0
$red_dim = ff4c5c
$green_dim = 4cb86a
$yellow_dim = 8baa43
$blue_dim = 5072ff
$blue_dim = 5665ff
$magenta_dim = c360e4
$cyan_dim = 30b5ce
+13 -1
View File
@@ -35,7 +35,7 @@
"description": "Caffeine - Prevent system from going idle"
},
"sc": {
"icon": "camera-video-symbolic",
"icon": "toolbox",
"description": "Screencapture - Record screen and audio"
},
"wall": {
@@ -105,6 +105,18 @@
"bt": {
"icon": "bluetooth",
"description": "Bluetooth - Manage Bluetooth devices and connections"
},
"script": {
"icon": "toolbox",
"description": "Scripts - Execute system scripts and utilities"
},
"cmd": {
"icon": "terminal",
"description": "Scripts - Quick access to system commands"
},
"util": {
"icon": "config",
"description": "Utilities - System utilities and toggles"
}
},
"settings": {
+84 -2
View File
@@ -132,12 +132,17 @@ class Launcher(Window):
# Hide trigger suggestions at startup
self._clear_results()
def show_launcher(self, trigger_keyword: str = None):
"""Show the launcher and focus the search entry.
def show_launcher(self, trigger_keyword: str = None, external: bool = False):
"""Show the launcher and focus the search entry, or execute command externally.
Args:
trigger_keyword: Optional trigger keyword to activate immediately (e.g., "google", "calc", "app")
external: If True, execute the command without showing the launcher UI
"""
if external and trigger_keyword:
# Execute command externally without showing launcher
return self._execute_external_command(trigger_keyword)
self.show_all()
if trigger_keyword:
@@ -481,6 +486,83 @@ class Launcher(Window):
# Don't hide the launcher - user should be able to continue typing
def _execute_external_command(self, command_string: str):
"""Execute a command externally without showing the launcher UI.
Args:
command_string: Full command string (e.g., "wall random", "calc 2+2")
Returns:
Result of the command execution or None if failed
"""
try:
# Parse the command to extract trigger and query
parts = command_string.strip().split(' ', 1)
if not parts:
return None
trigger_part = parts[0]
query_part = parts[1] if len(parts) > 1 else ""
# Find the plugin that handles this trigger
triggered_plugin = None
detected_trigger = None
for plugin in self.plugin_manager.get_active_plugins():
trigger = plugin.get_active_trigger(f"{trigger_part} ")
if trigger:
triggered_plugin = plugin
detected_trigger = trigger
break
if not triggered_plugin:
print(f"No plugin found for trigger: {trigger_part}")
return None
# Query the plugin with the remaining query
try:
results = triggered_plugin.query(query_part)
if not results:
print(f"No results found for query: {query_part}")
return None
# Find the first result that matches the query exactly or has highest relevance
best_result = None
for result in results:
# For exact matches like "random", execute immediately
if (hasattr(result, 'data') and result.data and
result.data.get('action') == query_part.strip()):
best_result = result
break
# For partial matches, take the first high-relevance result
elif not best_result and result.relevance >= 0.9:
best_result = result
# If no exact match, take the first result
if not best_result and results:
best_result = results[0]
if best_result:
# Execute the result action
try:
result_value = best_result.activate()
print(f"External command executed: {command_string}")
return result_value
except Exception as e:
print(f"Error executing result action: {e}")
return None
else:
print(f"No suitable result found for: {command_string}")
return None
except Exception as e:
print(f"Error querying plugin {triggered_plugin.name}: {e}")
return None
except Exception as e:
print(f"Error executing external command '{command_string}': {e}")
return None
def _update_results_display(self):
"""Update the results display."""
# Skip if still initializing or results_box not ready
+1
View File
@@ -98,6 +98,7 @@ class PluginManager:
"reminders",
"network",
"bluetooth",
"scripts",
]
for plugin_name in default_plugins:
+21 -65
View File
@@ -29,7 +29,7 @@ class BluetoothPlugin(PluginBase):
def initialize(self):
"""Initialize the bluetooth plugin."""
self.set_triggers(["bt", "bt "])
self.set_triggers(["bluetooth", "bt", "blue"])
def cleanup(self):
"""Cleanup the bluetooth plugin."""
@@ -41,7 +41,6 @@ class BluetoothPlugin(PluginBase):
if self._scan_timeout_id:
try:
from gi.repository import GLib
GLib.source_remove(self._scan_timeout_id)
self._scan_timeout_id = None
except Exception:
@@ -65,9 +64,7 @@ class BluetoothPlugin(PluginBase):
self.bluetooth_client.connect("device-added", self._on_device_changed)
self.bluetooth_client.connect("device-removed", self._on_device_changed)
self._bluetooth_available = True
print("✓ Bluetooth client initialized successfully")
except TimeoutError:
print("Warning: Bluetooth client initialization timed out")
self.bluetooth_client = None
except Exception as e:
print(f"Warning: Failed to initialize BluetoothClient: {e}")
@@ -117,15 +114,9 @@ class BluetoothPlugin(PluginBase):
battery_info = ""
if device.battery_percentage > 0:
battery_info = f" ({device.battery_percentage:.0f}%)"
return (
f"Connected to {device.alias}",
f"Device: {device.type}{battery_info}",
)
return f"Connected to {device.alias}", f"Device: {device.type}{battery_info}"
else:
return (
f"Connected to {len(connected_devices)} devices",
"Multiple devices connected",
)
return f"Connected to {len(connected_devices)} devices", "Multiple devices connected"
elif self.bluetooth_client.scanning:
return "Scanning for devices...", "Looking for nearby Bluetooth devices"
else:
@@ -135,11 +126,7 @@ class BluetoothPlugin(PluginBase):
"""Get appropriate icon for a bluetooth device based on its type."""
device_type = device.type.lower()
if (
"audio" in device_type
or "headphone" in device_type
or "headset" in device_type
):
if "audio" in device_type or "headphone" in device_type or "headset" in device_type:
return icons.headphones
elif "mouse" in device_type or "keyboard" in device_type:
return icons.keyboard
@@ -166,7 +153,6 @@ class BluetoothPlugin(PluginBase):
if self._scan_timeout_id:
try:
from gi.repository import GLib
GLib.source_remove(self._scan_timeout_id)
self._scan_timeout_id = None
except Exception:
@@ -194,14 +180,12 @@ class BluetoothPlugin(PluginBase):
def _stop_scanning(self):
"""Stop bluetooth scanning."""
if self.bluetooth_client and self.bluetooth_client.scanning:
print("Stopping bluetooth scanning...")
self.bluetooth_client.scanning = False
# Cancel any pending auto-stop timeout
if self._scan_timeout_id:
try:
from gi.repository import GLib
GLib.source_remove(self._scan_timeout_id)
self._scan_timeout_id = None
except Exception:
@@ -216,16 +200,13 @@ class BluetoothPlugin(PluginBase):
# Stop scanning if plugin hasn't been used for 10 seconds
import time
current_time = time.time()
if current_time - self._last_query_time > 10:
print("Stopping bluetooth scan due to inactivity...")
self._stop_scanning()
def _connect_device(self, device: BluetoothDevice):
"""Connect to a bluetooth device."""
if device and not device.connected:
print(f"Connecting to {device.alias}...")
device.connected = True # This will trigger the connection
# Force launcher refresh to show updated status
self._force_launcher_refresh()
@@ -234,7 +215,6 @@ class BluetoothPlugin(PluginBase):
def _disconnect_device(self, device: BluetoothDevice):
"""Disconnect from a bluetooth device."""
if device and device.connected:
print(f"Disconnecting from {device.alias}...")
device.connected = False # This will trigger the disconnection
# Force launcher refresh to show updated status
self._force_launcher_refresh()
@@ -242,18 +222,14 @@ class BluetoothPlugin(PluginBase):
def _create_connect_action(self, device: BluetoothDevice):
"""Create a connect action function with proper closure."""
def connect_action():
return self._connect_device(device)
return connect_action
def _create_disconnect_action(self, device: BluetoothDevice):
"""Create a disconnect action function with proper closure."""
def disconnect_action():
return self._disconnect_device(device)
return disconnect_action
def _force_launcher_refresh(self):
@@ -266,16 +242,13 @@ class BluetoothPlugin(PluginBase):
try:
# Try to access the launcher through the fabric Application
from fabric import Application
app = Application.get_default()
if app and hasattr(app, "launcher"):
if app and hasattr(app, 'launcher'):
launcher = app.launcher
if launcher and hasattr(launcher, "_perform_search"):
if launcher and hasattr(launcher, '_perform_search'):
# Get current query and trigger a search
current_query = (
launcher.query if hasattr(launcher, "query") else ""
)
current_query = launcher.query if hasattr(launcher, 'query') else ""
if not current_query:
# If no query, use the trigger to force refresh
current_query = "bt list"
@@ -286,15 +259,9 @@ class BluetoothPlugin(PluginBase):
# Fallback: try to find launcher instance through other means
import gc
for obj in gc.get_objects():
if (
hasattr(obj, "__class__")
and obj.__class__.__name__ == "Launcher"
):
if hasattr(obj, "_perform_search") and hasattr(
obj, "query"
):
if hasattr(obj, '__class__') and obj.__class__.__name__ == 'Launcher':
if hasattr(obj, '_perform_search') and hasattr(obj, 'query'):
current_query = obj.query if obj.query else "bt list"
obj._perform_search(current_query)
return False
@@ -314,16 +281,12 @@ class BluetoothPlugin(PluginBase):
"""Process bluetooth queries."""
# Update last query time to track plugin activity
import time
self._last_query_time = time.time()
# Schedule periodic check for inactive scanning
try:
from gi.repository import GLib
GLib.timeout_add_seconds(
15, lambda: (self._check_and_stop_inactive_scanning(), False)
)
GLib.timeout_add_seconds(15, lambda: (self._check_and_stop_inactive_scanning(), False))
except Exception:
pass
@@ -377,6 +340,7 @@ class BluetoothPlugin(PluginBase):
)
)
else:
results.append(
Result(
@@ -447,10 +411,7 @@ class BluetoothPlugin(PluginBase):
relevance = 1.0
elif device.connecting:
subtitle = f"Connecting... • {device.type}"
def action():
return None
action = lambda: None
relevance = 0.9
else:
subtitle = f"Disconnected • {device.type} • Click to connect"
@@ -470,7 +431,7 @@ class BluetoothPlugin(PluginBase):
"device_address": device.address,
"device_name": device.alias,
"connected": device.connected,
"keep_launcher_open": True,
"keep_launcher_open": True
},
)
)
@@ -509,14 +470,14 @@ class BluetoothPlugin(PluginBase):
return results
# Search for devices by name (like network plugin search)
if self.bluetooth_client and self.bluetooth_client.enabled:
devices = self.bluetooth_client.devices
matching_devices = [
device
for device in devices
if query_lower in device.alias.lower()
or query_lower in device.name.lower()
device for device in devices
if query_lower in device.alias.lower() or query_lower in device.name.lower()
]
if matching_devices:
@@ -540,10 +501,7 @@ class BluetoothPlugin(PluginBase):
relevance = 1.0
elif device.connecting:
subtitle = f"Connecting... • {device.type}"
def action():
return None
action = lambda: None
relevance = 0.9
else:
subtitle = f"Disconnected • {device.type} • Click to connect"
@@ -563,7 +521,7 @@ class BluetoothPlugin(PluginBase):
"device_address": device.address,
"device_name": device.alias,
"connected": device.connected,
"keep_launcher_open": True,
"keep_launcher_open": True
},
)
)
@@ -591,13 +549,11 @@ class BluetoothPlugin(PluginBase):
title="Bluetooth Not Available",
subtitle="Enable Bluetooth to search for devices",
icon_markup=icons.bluetooth_off,
action=lambda: self._toggle_bluetooth()
if self.bluetooth_client
else None,
action=lambda: self._toggle_bluetooth() if self.bluetooth_client else None,
relevance=0.5,
plugin_name=self.display_name,
data={"type": "info"},
)
)
return results
return results
+55 -16
View File
@@ -363,6 +363,21 @@ class Kanban(Gtk.Box):
self.set_vexpand(False)
self.set_hexpand(True)
# Make the widget focusable so it can receive key events
self.set_can_focus(True)
self.connect("key-press-event", self.on_key_press)
# Store pending add text for Enter key handling
self.pending_add_text = None
# Add a hidden Entry widget so the launcher will let us handle Enter events
from fabric.widgets.entry import Entry
self.hidden_entry = Entry(name="kanban-hidden-entry")
self.hidden_entry.set_size_request(1, 1) # Make it tiny
self.hidden_entry.set_opacity(0.0) # Make it invisible
self.hidden_entry.connect("activate", self.on_hidden_entry_activate)
self.add(self.hidden_entry)
self.grid = Gtk.Grid(
column_spacing=4,
column_homogeneous=True,
@@ -388,6 +403,29 @@ class Kanban(Gtk.Box):
self.load_state()
self.show_all()
def on_key_press(self, widget, event):
"""Handle key press events for the kanban widget."""
if event.keyval == Gdk.KEY_Return:
if self.pending_add_text:
self.columns[0].add_note(self.pending_add_text)
self.save_state()
self.pending_add_text = None
return True
return False
def on_hidden_entry_activate(self, entry):
"""Handle activation of the hidden entry (Enter key from launcher)."""
if self.pending_add_text:
self.columns[0].add_note(self.pending_add_text)
self.save_state()
self.pending_add_text = None
def set_pending_add_text(self, text):
"""Set the pending add text for Enter key handling."""
self.pending_add_text = text
# Don't grab focus - let the launcher manage focus
# The launcher will find our hidden entry when Enter is pressed
def save_state(self):
state = {
"columns": [
@@ -428,6 +466,7 @@ class KanbanPlugin(PluginBase):
self.display_name = "Kanban"
self.description = "Kanban board for task management"
self._current_widget = None
self._pending_add_text = None
def initialize(self):
"""Initialize the Kanban plugin."""
@@ -442,42 +481,42 @@ class KanbanPlugin(PluginBase):
self._current_widget.get_parent().remove(self._current_widget)
self._current_widget.destroy()
self._current_widget = None
self._pending_add_text = None
def query(self, query_string: str) -> List[Result]:
"""Process Kanban queries."""
results = []
# Clean up any existing widget first
self.cleanup()
# Only create a new widget if we don't have one
if not self._current_widget:
kanban_widget = Kanban()
self._current_widget = kanban_widget
else:
kanban_widget = self._current_widget
# Create a new kanban widget each time to avoid parent issues
kanban_widget = Kanban()
self._current_widget = kanban_widget
# Check if this is an "add" command
# Check if this is an "add" command and store it for Enter key handling
if query_string.startswith("add "):
# Extract the note text
note_text = query_string[4:].strip()
if note_text:
# Create an action that will add the note when executed
def add_task_action():
# Add to the "To Do" column (first column)
kanban_widget.columns[0].add_note(note_text)
kanban_widget.save_state()
return None
# Store the pending add command in the widget
kanban_widget.set_pending_add_text(note_text)
else:
kanban_widget.set_pending_add_text(None)
else:
kanban_widget.set_pending_add_text(None)
add_task_action()
# Always show the kanban board directly when triggered
results.append(
Result(
title="Kanban Board",
subtitle="Task management with kanban-style columns",
icon_markup=icons.kanban,
action=lambda: None, # No action needed since widget is displayed directly
action=lambda: None, # No action needed, widget handles Enter
relevance=1.0,
plugin_name=self.display_name,
custom_widget=kanban_widget,
data={"type": "kanban_board"},
data={"type": "kanban_board", "keep_launcher_open": True},
)
)
+54 -235
View File
@@ -44,6 +44,7 @@ class ScreencapturePlugin(PluginBase):
"ss-both": "Take a screenshot of both displays",
"screenshot-hdmi": "Take a screenshot of HDMI display",
"ss-hdmi": "Take a screenshot of HDMI display",
# Recording commands (with audio)
"record": "Start recording main display with audio",
"rec": "Start recording main display with audio",
@@ -51,6 +52,7 @@ class ScreencapturePlugin(PluginBase):
"rec-region": "Start recording selected region with audio",
"record-hdmi": "Start recording HDMI display with audio",
"rec-hdmi": "Start recording HDMI display with audio",
# Recording commands (no audio)
"record-noaudio": "Start recording main display without audio",
"rec-noaudio": "Start recording main display without audio",
@@ -58,6 +60,7 @@ class ScreencapturePlugin(PluginBase):
"rec-noaudio-region": "Start recording selected region without audio",
"record-noaudio-hdmi": "Start recording HDMI display without audio",
"rec-noaudio-hdmi": "Start recording HDMI display without audio",
# High-quality recording commands
"record-hq": "Start high-quality recording of main display",
"rec-hq": "Start high-quality recording of main display",
@@ -65,13 +68,16 @@ class ScreencapturePlugin(PluginBase):
"rec-hq-region": "Start high-quality recording of selected region",
"record-hq-hdmi": "Start high-quality recording of HDMI display",
"rec-hq-hdmi": "Start high-quality recording of HDMI display",
# GIF recording commands
"record-gif": "Start GIF recording of main display",
"rec-gif": "Start GIF recording of main display",
"record-gif-region": "Start GIF recording of selected region",
"rec-gif-region": "Start GIF recording of selected region",
# Control commands
"stop": "Stop current recording",
# Conversion commands
"convert-webm": "Convert recordings to WebM format",
"conv-webm": "Convert recordings to WebM format",
@@ -94,7 +100,9 @@ class ScreencapturePlugin(PluginBase):
"""Check if recording is currently active."""
try:
result = subprocess.run(
[self.script_path, "status"], capture_output=True, text=True
[self.script_path, "status"],
capture_output=True,
text=True
)
return result.stdout.strip() == "true"
except Exception:
@@ -107,246 +115,57 @@ class ScreencapturePlugin(PluginBase):
command_info = {
# Screenshot commands
"screenshot": (
"Take Screenshot (eDP-1)",
"Capture the main display",
ssfull,
lambda: self._run_script("screenshot", "eDP-1"),
),
"ss": (
"Take Screenshot (eDP-1)",
"Capture the main display",
ssfull,
lambda: self._run_script("screenshot", "eDP-1"),
),
"screenshot-region": (
"Take Region Screenshot",
"Capture a selected region",
ssregion,
lambda: self._run_script("screenshot", "selection"),
),
"ss-region": (
"Take Region Screenshot",
"Capture a selected region",
ssregion,
lambda: self._run_script("screenshot", "selection"),
),
"screenshot-both": (
"Take Screenshot (Both Displays)",
"Capture both displays combined",
screenshots,
lambda: self._run_script("screenshot", "both"),
),
"ss-both": (
"Take Screenshot (Both Displays)",
"Capture both displays combined",
screenshots,
lambda: self._run_script("screenshot", "both"),
),
"screenshot-hdmi": (
"Take Screenshot (HDMI-A-1)",
"Capture HDMI display",
ssfull,
lambda: self._run_script("screenshot", "HDMI-A-1"),
),
"ss-hdmi": (
"Take Screenshot (HDMI-A-1)",
"Capture HDMI display",
ssfull,
lambda: self._run_script("screenshot", "HDMI-A-1"),
),
"screenshot": ("Take Screenshot (eDP-1)", "Capture the main display", ssfull, lambda: self._run_script("screenshot", "eDP-1")),
"ss": ("Take Screenshot (eDP-1)", "Capture the main display", ssfull, lambda: self._run_script("screenshot", "eDP-1")),
"screenshot-region": ("Take Region Screenshot", "Capture a selected region", ssregion, lambda: self._run_script("screenshot", "selection")),
"ss-region": ("Take Region Screenshot", "Capture a selected region", ssregion, lambda: self._run_script("screenshot", "selection")),
"screenshot-both": ("Take Screenshot (Both Displays)", "Capture both displays combined", screenshots, lambda: self._run_script("screenshot", "both")),
"ss-both": ("Take Screenshot (Both Displays)", "Capture both displays combined", screenshots, lambda: self._run_script("screenshot", "both")),
"screenshot-hdmi": ("Take Screenshot (HDMI-A-1)", "Capture HDMI display", ssfull, lambda: self._run_script("screenshot", "HDMI-A-1")),
"ss-hdmi": ("Take Screenshot (HDMI-A-1)", "Capture HDMI display", ssfull, lambda: self._run_script("screenshot", "HDMI-A-1")),
# Recording commands (with audio)
"record": (
"Start Recording (eDP-1)",
"Record the main display with audio",
screenrecord,
lambda: self._run_script("record", "eDP-1"),
),
"rec": (
"Start Recording (eDP-1)",
"Record the main display with audio",
screenrecord,
lambda: self._run_script("record", "eDP-1"),
),
"record-region": (
"Start Region Recording",
"Record a selected region with audio",
screenrecord,
lambda: self._run_script("record", "selection"),
),
"rec-region": (
"Start Region Recording",
"Record a selected region with audio",
screenrecord,
lambda: self._run_script("record", "selection"),
),
"record-hdmi": (
"Start Recording (HDMI-A-1)",
"Record HDMI display with audio",
screenrecord,
lambda: self._run_script("record", "HDMI-A-1"),
),
"rec-hdmi": (
"Start Recording (HDMI-A-1)",
"Record HDMI display with audio",
screenrecord,
lambda: self._run_script("record", "HDMI-A-1"),
),
"record": ("Start Recording (eDP-1)", "Record the main display with audio", screenrecord, lambda: self._run_script("record", "eDP-1")),
"rec": ("Start Recording (eDP-1)", "Record the main display with audio", screenrecord, lambda: self._run_script("record", "eDP-1")),
"record-region": ("Start Region Recording", "Record a selected region with audio", screenrecord, lambda: self._run_script("record", "selection")),
"rec-region": ("Start Region Recording", "Record a selected region with audio", screenrecord, lambda: self._run_script("record", "selection")),
"record-hdmi": ("Start Recording (HDMI-A-1)", "Record HDMI display with audio", screenrecord, lambda: self._run_script("record", "HDMI-A-1")),
"rec-hdmi": ("Start Recording (HDMI-A-1)", "Record HDMI display with audio", screenrecord, lambda: self._run_script("record", "HDMI-A-1")),
# Recording commands (no audio)
"record-noaudio": (
"Start Recording No Audio (eDP-1)",
"Record the main display without audio",
screenrecord,
lambda: self._run_script("record-noaudio", "eDP-1"),
),
"rec-noaudio": (
"Start Recording No Audio (eDP-1)",
"Record the main display without audio",
screenrecord,
lambda: self._run_script("record-noaudio", "eDP-1"),
),
"record-noaudio-region": (
"Start Region Recording No Audio",
"Record a selected region without audio",
screenrecord,
lambda: self._run_script("record-noaudio", "selection"),
),
"rec-noaudio-region": (
"Start Region Recording No Audio",
"Record a selected region without audio",
screenrecord,
lambda: self._run_script("record-noaudio", "selection"),
),
"record-noaudio-hdmi": (
"Start Recording No Audio (HDMI-A-1)",
"Record HDMI display without audio",
screenrecord,
lambda: self._run_script("record-noaudio", "HDMI-A-1"),
),
"rec-noaudio-hdmi": (
"Start Recording No Audio (HDMI-A-1)",
"Record HDMI display without audio",
screenrecord,
lambda: self._run_script("record-noaudio", "HDMI-A-1"),
),
"record-noaudio": ("Start Recording No Audio (eDP-1)", "Record the main display without audio", screenrecord, lambda: self._run_script("record-noaudio", "eDP-1")),
"rec-noaudio": ("Start Recording No Audio (eDP-1)", "Record the main display without audio", screenrecord, lambda: self._run_script("record-noaudio", "eDP-1")),
"record-noaudio-region": ("Start Region Recording No Audio", "Record a selected region without audio", screenrecord, lambda: self._run_script("record-noaudio", "selection")),
"rec-noaudio-region": ("Start Region Recording No Audio", "Record a selected region without audio", screenrecord, lambda: self._run_script("record-noaudio", "selection")),
"record-noaudio-hdmi": ("Start Recording No Audio (HDMI-A-1)", "Record HDMI display without audio", screenrecord, lambda: self._run_script("record-noaudio", "HDMI-A-1")),
"rec-noaudio-hdmi": ("Start Recording No Audio (HDMI-A-1)", "Record HDMI display without audio", screenrecord, lambda: self._run_script("record-noaudio", "HDMI-A-1")),
# High-quality recording commands
"record-hq": (
"Start HQ Recording (eDP-1)",
"High-quality recording for YouTube",
screenrecord,
lambda: self._run_script("record-hq", "eDP-1"),
),
"rec-hq": (
"Start HQ Recording (eDP-1)",
"High-quality recording for YouTube",
screenrecord,
lambda: self._run_script("record-hq", "eDP-1"),
),
"record-hq-region": (
"Start HQ Region Recording",
"High-quality region recording",
screenrecord,
lambda: self._run_script("record-hq", "selection"),
),
"rec-hq-region": (
"Start HQ Region Recording",
"High-quality region recording",
screenrecord,
lambda: self._run_script("record-hq", "selection"),
),
"record-hq-hdmi": (
"Start HQ Recording (HDMI-A-1)",
"High-quality HDMI recording",
screenrecord,
lambda: self._run_script("record-hq", "HDMI-A-1"),
),
"rec-hq-hdmi": (
"Start HQ Recording (HDMI-A-1)",
"High-quality HDMI recording",
screenrecord,
lambda: self._run_script("record-hq", "HDMI-A-1"),
),
"record-hq": ("Start HQ Recording (eDP-1)", "High-quality recording for YouTube", screenrecord, lambda: self._run_script("record-hq", "eDP-1")),
"rec-hq": ("Start HQ Recording (eDP-1)", "High-quality recording for YouTube", screenrecord, lambda: self._run_script("record-hq", "eDP-1")),
"record-hq-region": ("Start HQ Region Recording", "High-quality region recording", screenrecord, lambda: self._run_script("record-hq", "selection")),
"rec-hq-region": ("Start HQ Region Recording", "High-quality region recording", screenrecord, lambda: self._run_script("record-hq", "selection")),
"record-hq-hdmi": ("Start HQ Recording (HDMI-A-1)", "High-quality HDMI recording", screenrecord, lambda: self._run_script("record-hq", "HDMI-A-1")),
"rec-hq-hdmi": ("Start HQ Recording (HDMI-A-1)", "High-quality HDMI recording", screenrecord, lambda: self._run_script("record-hq", "HDMI-A-1")),
# GIF recording commands
"record-gif": (
"Start GIF Recording (eDP-1)",
"Record as optimized GIF",
screenrecord,
lambda: self._run_script("record-gif", "eDP-1"),
),
"rec-gif": (
"Start GIF Recording (eDP-1)",
"Record as optimized GIF",
screenrecord,
lambda: self._run_script("record-gif", "eDP-1"),
),
"record-gif-region": (
"Start GIF Region Recording",
"Record selected region as GIF",
screenrecord,
lambda: self._run_script("record-gif", "selection"),
),
"rec-gif-region": (
"Start GIF Region Recording",
"Record selected region as GIF",
screenrecord,
lambda: self._run_script("record-gif", "selection"),
),
"record-gif": ("Start GIF Recording (eDP-1)", "Record as optimized GIF", screenrecord, lambda: self._run_script("record-gif", "eDP-1")),
"rec-gif": ("Start GIF Recording (eDP-1)", "Record as optimized GIF", screenrecord, lambda: self._run_script("record-gif", "eDP-1")),
"record-gif-region": ("Start GIF Region Recording", "Record selected region as GIF", screenrecord, lambda: self._run_script("record-gif", "selection")),
"rec-gif-region": ("Start GIF Region Recording", "Record selected region as GIF", screenrecord, lambda: self._run_script("record-gif", "selection")),
# Control commands
"stop": (
"Stop Recording",
"Stop the current screen recording",
stop,
lambda: self._run_script("record", "stop"),
),
"stop": ("Stop Recording", "Stop the current screen recording", stop, lambda: self._run_script("record", "stop")),
# Conversion commands
"convert-webm": (
"Convert to WebM",
"Convert MKV recordings to WebM format",
screenrecord,
lambda: self._run_script("convert", "webm"),
),
"conv-webm": (
"Convert to WebM",
"Convert MKV recordings to WebM format",
screenrecord,
lambda: self._run_script("convert", "webm"),
),
"convert-iphone": (
"Convert for iPhone",
"Convert recordings for iPhone compatibility",
screenrecord,
lambda: self._run_script("convert", "iphone"),
),
"conv-iphone": (
"Convert for iPhone",
"Convert recordings for iPhone compatibility",
screenrecord,
lambda: self._run_script("convert", "iphone"),
),
"convert-youtube": (
"Convert for YouTube",
"Convert recordings for YouTube upload",
screenrecord,
lambda: self._run_script("convert", "youtube"),
),
"conv-youtube": (
"Convert for YouTube",
"Convert recordings for YouTube upload",
screenrecord,
lambda: self._run_script("convert", "youtube"),
),
"convert-gif": (
"Convert to GIF",
"Convert video recordings to GIF",
screenrecord,
lambda: self._run_script("convert", "gif"),
),
"conv-gif": (
"Convert to GIF",
"Convert video recordings to GIF",
screenrecord,
lambda: self._run_script("convert", "gif"),
),
"convert-webm": ("Convert to WebM", "Convert MKV recordings to WebM format", screenrecord, lambda: self._run_script("convert", "webm")),
"conv-webm": ("Convert to WebM", "Convert MKV recordings to WebM format", screenrecord, lambda: self._run_script("convert", "webm")),
"convert-iphone": ("Convert for iPhone", "Convert recordings for iPhone compatibility", screenrecord, lambda: self._run_script("convert", "iphone")),
"conv-iphone": ("Convert for iPhone", "Convert recordings for iPhone compatibility", screenrecord, lambda: self._run_script("convert", "iphone")),
"convert-youtube": ("Convert for YouTube", "Convert recordings for YouTube upload", screenrecord, lambda: self._run_script("convert", "youtube")),
"conv-youtube": ("Convert for YouTube", "Convert recordings for YouTube upload", screenrecord, lambda: self._run_script("convert", "youtube")),
"convert-gif": ("Convert to GIF", "Convert video recordings to GIF", screenrecord, lambda: self._run_script("convert", "gif")),
"conv-gif": ("Convert to GIF", "Convert video recordings to GIF", screenrecord, lambda: self._run_script("convert", "gif")),
}
if command in command_info:
+265
View File
@@ -0,0 +1,265 @@
"""
Scripts plugin for the launcher.
Provides access to various system scripts and utilities.
"""
import os
import subprocess
from typing import List
from fabric.utils import exec_shell_command_async
import utils.icons as icons
from modules.launcher.plugin_base import PluginBase
from modules.launcher.result import Result
class ScriptsPlugin(PluginBase):
"""
Plugin for executing system scripts and utilities.
"""
def __init__(self):
super().__init__()
self.display_name = "Scripts"
self.description = "System scripts and utilities"
def initialize(self):
"""Initialize the scripts plugin."""
self.set_triggers(["script", "scripts", "cmd", "util"])
def cleanup(self):
"""Cleanup the scripts plugin."""
pass
def _check_gamemode_status(self) -> bool:
"""Check if gamemode (animations disabled) is currently active."""
try:
# Use the gamemode script's check function
script_path = self._get_script_path("gamemode.sh")
result = subprocess.run(
["bash", script_path, "check"],
capture_output=True,
text=True,
timeout=2
)
if result.returncode == 0:
# Script returns "t" for true (gamemode on), "f" for false (gamemode off)
return result.stdout.strip() == "t"
except Exception:
pass
return False
def _toggle_gamemode(self):
"""Toggle gamemode using the gamemode script."""
script_path = self._get_script_path("gamemode.sh")
exec_shell_command_async(f"bash '{script_path}'")
def _run_colorpicker(self, format_type: str):
"""Run color picker with specified format."""
script_path = self._get_script_path("hyprpicker.sh")
exec_shell_command_async(f"bash '{script_path}' -{format_type}")
def _reload_css(self):
"""Reload application CSS."""
exec_shell_command_async("$fabricSend 'app.set_css()'")
def _restart_modus(self):
"""Restart Modus application."""
exec_shell_command_async("killall modus; uwsm-app python ~/Modus/main.py")
def _run_system_command(self, command: str):
"""Run a system command."""
exec_shell_command_async(command)
def _toggle_service(self, service_name: str):
"""Toggle a systemd service."""
exec_shell_command_async(f"systemctl --user is-active {service_name} && systemctl --user stop {service_name} || systemctl --user start {service_name}")
def _get_script_path(self, script_name: str) -> str:
"""Get the full path to a script from the scripts directory."""
# Get the current working directory (should be Modus root)
current_dir = os.getcwd()
# Try scripts directory relative to current working directory
script_path = os.path.join(current_dir, "scripts", script_name)
if os.path.exists(script_path):
return script_path
# Try relative to this file's location
plugin_dir = os.path.dirname(os.path.abspath(__file__))
modus_root = os.path.dirname(os.path.dirname(os.path.dirname(plugin_dir)))
script_path = os.path.join(modus_root, "scripts", script_name)
if os.path.exists(script_path):
return script_path
# Fallback to relative path
return f"scripts/{script_name}"
def query(self, query_string: str) -> List[Result]:
"""Search for script actions based on query."""
query = query_string.strip().lower()
results = []
# Check gamemode status for dynamic display
gamemode_active = self._check_gamemode_status()
gamemode_icon = icons.gamemode if gamemode_active else icons.gamemode_off
gamemode_status = "ON" if gamemode_active else "OFF"
# Define script actions
script_actions = {
# Gamemode
"gamemode": {
"title": f"Toggle Gamemode (Currently {gamemode_status})",
"subtitle": "Toggle Hyprland game mode - disable animations, gaps, and effects",
"icon": gamemode_icon,
"action": self._toggle_gamemode,
"keywords": ["gamemode", "game", "performance", "toggle"],
"relevance": 1.0
},
# Color Picker
"colorpicker-hex": {
"title": "Color Picker (HEX)",
"subtitle": "Pick a color and copy HEX value to clipboard",
"icon": icons.colorpicker,
"action": lambda: self._run_colorpicker("hex"),
"keywords": ["color", "picker", "hex", "colorpicker"],
"relevance": 0.9
},
"colorpicker-rgb": {
"title": "Color Picker (RGB)",
"subtitle": "Pick a color and copy RGB value to clipboard",
"icon": icons.colorpicker,
"action": lambda: self._run_colorpicker("rgb"),
"keywords": ["color", "picker", "rgb", "colorpicker"],
"relevance": 0.9
},
"colorpicker-hsv": {
"title": "Color Picker (HSV)",
"subtitle": "Pick a color and copy HSV value to clipboard",
"icon": icons.colorpicker,
"action": lambda: self._run_colorpicker("hsv"),
"keywords": ["color", "picker", "hsv", "colorpicker"],
"relevance": 0.9
},
# System Controls
"reload-css": {
"title": "Reload CSS",
"subtitle": "Reload application stylesheets",
"icon": icons.reload,
"action": self._reload_css,
"keywords": ["reload", "css", "style", "stylesheet"],
"relevance": 0.6
},
"restart-modus": {
"title": "Restart Modus",
"subtitle": "Restart the Modus application",
"icon": icons.reload,
"action": self._restart_modus,
"keywords": ["restart", "modus", "reload", "reboot"],
"relevance": 0.6
},
# System Utilities
"kill-all": {
"title": "Kill All Processes",
"subtitle": "Kill all user processes (useful for cleanup)",
"icon": icons.close,
"action": lambda: self._run_system_command("killall -u $USER"),
"keywords": ["kill", "killall", "processes", "cleanup"],
"relevance": 0.5
},
"clear-cache": {
"title": "Clear Cache",
"subtitle": "Clear system and user cache directories",
"icon": icons.trash,
"action": lambda: self._run_system_command("rm -rf ~/.cache/* /tmp/*"),
"keywords": ["clear", "cache", "clean", "tmp", "cleanup"],
"relevance": 0.5
},
# Night Light / Blue Light Filter
"night-light": {
"title": "Toggle Night Light",
"subtitle": "Toggle blue light filter for eye comfort",
"icon": icons.night,
"action": lambda: self._run_system_command("hyprctl keyword decoration:screen_shader ~/.config/hypr/shaders/blue_light_filter.glsl"),
"keywords": ["night", "light", "blue", "filter", "eye", "comfort"],
"relevance": 0.6
},
# Clipboard
"clear-clipboard": {
"title": "Clear Clipboard",
"subtitle": "Clear clipboard contents",
"icon": icons.clipboard,
"action": lambda: self._run_system_command("wl-copy --clear"),
"keywords": ["clipboard", "clear", "empty"],
"relevance": 0.5
},
}
# If no query, show all actions
if not query:
for action_id, action_data in script_actions.items():
results.append(Result(
title=action_data["title"],
subtitle=action_data["subtitle"],
icon_markup=action_data["icon"],
action=action_data["action"],
relevance=action_data["relevance"],
plugin_name=self.display_name,
data={"action_id": action_id}
))
else:
# Search through actions
for action_id, action_data in script_actions.items():
relevance = self._calculate_relevance(query, action_data["keywords"], action_data["title"])
if relevance > 0:
results.append(Result(
title=action_data["title"],
subtitle=action_data["subtitle"],
icon_markup=action_data["icon"],
action=action_data["action"],
relevance=relevance,
plugin_name=self.display_name,
data={"action_id": action_id}
))
return sorted(results, key=lambda x: x.relevance, reverse=True)
def _calculate_relevance(self, query: str, keywords: List[str], title: str) -> float:
"""Calculate relevance score for a script action."""
query_lower = query.lower()
title_lower = title.lower()
# Check for exact matches in keywords
for keyword in keywords:
if query_lower == keyword.lower():
return 1.0
# Check if query starts with any keyword
for keyword in keywords:
if keyword.lower().startswith(query_lower):
return 0.9
# Check if title contains query
if query_lower in title_lower:
return 0.8
# Check if any keyword contains query
for keyword in keywords:
if query_lower in keyword.lower():
return 0.7
return 0.0
+4
View File
@@ -2,3 +2,7 @@
Modus services package.
Contains background services and utilities for the shell.
"""
from .notification import CustomNotifications
notification_service = CustomNotifications()
+226
View File
@@ -0,0 +1,226 @@
import json
import os
from typing import List
from fabric import Signal
from fabric.notifications import Notification, Notifications
from loguru import logger
import gi
gi.require_version('GLib', '2.0')
from gi.repository import GLib
import config.data as data
NOTIFICATION_CACHE_FILE = f"{data.CACHE_DIR}/notifications.json"
class CustomNotifications(Notifications):
"""A service to manage the notifications."""
@Signal
def clear_all(self, value: bool) -> None:
"""Signal emitted when notifications are emptied."""
# Implement as needed for your application
@Signal
def notification_count(self, value: int) -> None:
"""Signal emitted when a new notification is added."""
# Implement as needed for your application
@Signal
def dnd(self, value: bool) -> None:
"""Signal emitted when dnd is toggled."""
# Implement as needed for your application
@property
def count(self) -> int:
"""Return the count of notifications."""
return len(self.all_notifications)
@property
def dont_disturb(self) -> bool:
"""Return the pause status."""
return self._dont_disturb
@dont_disturb.setter
def dont_disturb(self, value: bool):
"""Set the pause status."""
self._dont_disturb = value
self.emit("dnd", value)
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.all_notifications = []
self._count = 0 # Will be updated to highest ID when loading
self.deserialized_notifications = []
self._dont_disturb = False
self._load_notifications()
def _load_notifications(self):
"""Read notifications from the cache file."""
if os.path.exists(NOTIFICATION_CACHE_FILE):
try:
with open(NOTIFICATION_CACHE_FILE, "r") as file:
notifications = json.load(file)
def validate_with_id(notif):
"""Helper to validate and return ID if valid."""
try:
self._deserialize_notification(notif)
return (True, notif, notif.get("id", 0))
except Exception as e:
msg = f"[Notification] Invalid: {str(e)[:50]}"
logger.error(msg)
return (False, None, 0)
# Validate all notifications at once
results = [validate_with_id(n) for n in notifications]
# Process results and find highest ID
valid_notifications = []
highest_id = self._count # Start with current count
for is_valid, notif, notif_id in results:
if is_valid:
valid_notifications.append(notif)
highest_id = max(highest_id, notif_id)
self.all_notifications = valid_notifications
self._count = highest_id # Update to highest ID seen
self._write_notifications(self.all_notifications)
except (json.JSONDecodeError, KeyError, ValueError, IndexError) as e:
logger.error(f"[Notification] {e}")
self.all_notifications = []
self._count = 0
def remove_notification(self, id: int):
"""Remove the notification of given id."""
item = next((p for p in self.all_notifications if p["id"] == id), None)
if item:
self.all_notifications.remove(item)
self._write_notifications(self.all_notifications)
self.emit("notification_count", len(self.all_notifications))
# Emit clear_all signal if there are no notifications left
if len(self.all_notifications) == 0:
self.emit("clear_all", True)
def cache_notification(self, data: Notification):
"""Cache the notification without widget_config and max_count."""
# First clean up any invalid notifications
self._cleanup_invalid_notifications()
# Create the new notification
self._count += 1
new_id = self._count
timestamp = GLib.get_real_time() / 1000000 # Convert microseconds to seconds
serialized_data = data.serialize()
serialized_data.update({
"id": new_id,
"app_name": data.app_name,
"timestamp": timestamp
})
# Append the new notification
self.all_notifications.append(serialized_data)
self._write_notifications(self.all_notifications)
self.emit("notification_count", len(self.all_notifications))
def _cleanup_invalid_notifications(self):
"""Remove any invalid notifications."""
def validate_with_id(notif):
"""Helper to validate and return result with ID."""
try:
self._deserialize_notification(notif)
return (True, notif, None)
except Exception as e:
msg = f"[Notification] Removing invalid: {str(e)[:50]}"
logger.debug(msg)
return (False, None, notif.get("id", 0))
# Validate all notifications at once
results = [validate_with_id(n) for n in self.all_notifications]
# Process results
valid_notifications = []
invalid_count = 0
for is_valid, notif, invalid_id in results:
if is_valid:
valid_notifications.append(notif)
else:
invalid_count += 1
self.emit("notification-closed", invalid_id, "dismissed-by-limit")
if invalid_count > 0:
self.all_notifications = valid_notifications
self._write_notifications(self.all_notifications)
self.emit("notification_count", len(self.all_notifications))
def _deserialize_notification(self, notification):
"""Deserialize a notification."""
try:
notif = Notification.deserialize(notification)
# Store the timestamp in the notification object
notif.timestamp = notification.get("timestamp", GLib.get_real_time() / 1000000)
return notif
except Exception as e:
logger.error(
f"[Notification] Deserialize failed for: {notification}, Error: {e}"
)
raise
def clear_all_notifications(self):
"""Empty the notifications."""
logger.info("[Notification] Clearing all notifications")
# Clear notifications but preserve the highest ID we've seen
highest_id = self._count
self.all_notifications = []
self._write_notifications(self.all_notifications)
self.emit("notification_count", 0)
self.emit("clear_all", True)
# Restore the ID counter so new notifications get unique IDs
self._count = highest_id
def _write_notifications(self, data):
"""Write the notifications to the cache file."""
try:
with open(NOTIFICATION_CACHE_FILE, "w") as f:
json.dump(data, f, indent=4, ensure_ascii=False)
logger.info("[Notification] Notifications written successfully.")
except Exception as e:
logger.error(f"[Notification] Failed to write notifications: {e}")
def get_deserialized(self) -> List[Notification]:
"""Return the notifications."""
def deserialize_with_id(notif):
"""Helper to deserialize and return result with ID."""
try:
return (self._deserialize_notification(notif), None)
except Exception as e:
msg = f"[Notification] Deserialize failed: {str(e)[:50]}"
logger.error(msg)
return (None, notif.get("id"))
# Process all notifications at once
results = [
deserialize_with_id(notification) for notification in self.all_notifications
]
# Split into successful and failed
deserialized = []
invalid_ids = []
for result, error_id in results:
if result is not None:
deserialized.append(result)
elif error_id is not None:
invalid_ids.append(error_id)
# Clean up invalid notifications
for invalid_id in invalid_ids:
self.remove_notification(invalid_id)
return deserialized
+12 -12
View File
@@ -1,20 +1,20 @@
:vars {
--foreground: #e2e2e9;
--foreground: #e3e1e9;
--background: #121318;
--cursor: #e2e2e9;
--primary: #aec6ff;
--on-primary: #122f60;
--secondary: #bfc6dc;
--on-secondary: #293041;
--tertiary: #dfbbde;
--on-tertiary: #402843;
--cursor: #e3e1e9;
--primary: #bac3ff;
--on-primary: #222c61;
--secondary: #c3c5dd;
--on-secondary: #2c2f42;
--tertiary: #e5bad8;
--on-tertiary: #44263e;
--surface: #121318;
--surface-bright: #37393e;
--surface-bright: #38393f;
--error: #ffb4ab;
--error-dim: #ff8678;
--on-error: #690005;
--error-container: #93000a;
--outline: #8e9099;
--outline: #90909a;
--shadow: #000000;
--red: #ffb2b9;
--red-dim: #ff7f8b;
@@ -22,8 +22,8 @@
--green-dim: #70c789;
--yellow: #b8cf84;
--yellow-dim: #a3c15f;
--blue: #b6c4ff;
--blue-dim: #839bff;
--blue: #bcc2ff;
--blue-dim: #8994ff;
--magenta: #e4b7f3;
--magenta-dim: #d48bec;
--cyan: #82d3e2;
+2 -1
View File
@@ -2,9 +2,10 @@ import math
from typing import cast
import cairo
from fabric.widgets.image import Image
from gi.repository import Gtk
from fabric.widgets.image import Image
class CustomImage(Image):
def do_render_rectangle(
-37
View File
@@ -1,37 +0,0 @@
import math
import cairo
from typing import cast
from fabric.widgets.image import Image
from gi.repository import Gtk
class CustomImage(Image):
def do_render_rectangle(
self, cr: cairo.Context, width: int, height: int, radius: int = 0
):
cr.move_to(radius, 0)
cr.line_to(width - radius, 0)
cr.arc(width - radius, radius, radius, -(math.pi / 2), 0)
cr.line_to(width, height - radius)
cr.arc(width - radius, height - radius, radius, 0, (math.pi / 2))
cr.line_to(radius, height)
cr.arc(radius, height - radius, radius, (math.pi / 2), math.pi)
cr.line_to(0, radius)
cr.arc(radius, radius, radius, math.pi, (3 * (math.pi / 2)))
cr.close_path()
def do_draw(self, cr: cairo.Context):
context = self.get_style_context()
width, height = self.get_allocated_width(), self.get_allocated_height()
cr.save()
self.do_render_rectangle(
cr,
width,
height,
cast(int, context.get_property("border-radius", Gtk.StateFlags.NORMAL)),
)
cr.clip()
Image.do_draw(self, cr)
cr.restore()