sankalp/Modus
1
0 Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases 1 Wiki Activity

♻️ refactor(launcher): improve code structure and plugin functionality

Browse Source
This commit is contained in:
sankalp committed 2025-07-30 13:04:22 +05:45
1 parent 4822693298
commit 5495233667
31 files changed
+536 -618
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

modules/launcher/plugins/bash_scripts.py
+115 -46
View File
@@ -82,6 +82,7 @@ class BashScriptsPlugin(PluginBase):
"""Get reference to the battery service."""
try:
from services.battery import Battery
self._battery_service = Battery()
except Exception as e:
print(f"BashScriptsPlugin: Error getting battery service: {e}")
@@ -95,7 +96,10 @@ class BashScriptsPlugin(PluginBase):
try:
# Check if profile proxy is available
if not hasattr(self._battery_service, '_profile_proxy') or not self._battery_service._profile_proxy:
if (
not hasattr(self._battery_service, "_profile_proxy")
or not self._battery_service._profile_proxy
):
print("BashScriptsPlugin: Power profile proxy not available")
return False
@@ -114,7 +118,7 @@ class BashScriptsPlugin(PluginBase):
profile_mapping = {
"power-saver": ["power-saver", "powersave", "power_saver"],
"balanced": ["balanced", "balance"],
"performance": ["performance", "performance-mode"]
"performance": ["performance", "performance-mode"],
}
# Find the actual profile name
@@ -129,18 +133,20 @@ class BashScriptsPlugin(PluginBase):
if actual_profile:
self._battery_service._profile_proxy.ActiveProfile = actual_profile
print(f"BashScriptsPlugin: Successfully set power profile to {actual_profile}")
print(
f"BashScriptsPlugin: Successfully set power profile to {actual_profile}"
)
return True
else:
print(f"BashScriptsPlugin: No matching profile found for '{profile}' in available profiles: {available_profiles}")
print(
f"BashScriptsPlugin: No matching profile found for '{profile}' in available profiles: {available_profiles}"
)
return False
except Exception as e:
print(f"BashScriptsPlugin: Error setting power profile: {e}")
return False
def _load_scripts_cache(self):
"""Load scripts cache from JSON file."""
try:
@@ -150,7 +156,9 @@ class BashScriptsPlugin(PluginBase):
self._scripts_cache = cache_data.get("scripts", {})
self._last_cache_update = cache_data.get("last_update", 0)
else:
print("BashScriptsPlugin: No cache file found, will build cache in background")
print(
"BashScriptsPlugin: No cache file found, will build cache in background"
)
except Exception as e:
print(f"BashScriptsPlugin: Error loading scripts cache: {e}")
self._scripts_cache = {}
@@ -172,16 +180,17 @@ class BashScriptsPlugin(PluginBase):
def _start_background_cache_update(self):
"""Start background thread to update scripts cache."""
current_time = time.time()
# Check if cache needs updating
if (current_time - self._last_cache_update > self._cache_update_interval or
not self._scripts_cache):
if (
current_time - self._last_cache_update > self._cache_update_interval
or not self._scripts_cache
):
if not self._cache_building:
self._cache_building = True
self._cache_thread = threading.Thread(
target=self._build_scripts_cache_background,
daemon=True
target=self._build_scripts_cache_background, daemon=True
)
self._cache_thread.start()
@@ -192,11 +201,15 @@ class BashScriptsPlugin(PluginBase):
new_cache = {}
# Scan Modus scripts directory for discovered scripts
if os.path.exists(self.modus_scripts_dir) and os.path.isdir(self.modus_scripts_dir):
if os.path.exists(self.modus_scripts_dir) and os.path.isdir(
self.modus_scripts_dir
):
try:
self._scan_directory_for_scripts(self.modus_scripts_dir, new_cache)
except (PermissionError, FileNotFoundError, OSError) as e:
print(f"BashScriptsPlugin: Error scanning Modus scripts directory: {e}")
print(
f"BashScriptsPlugin: Error scanning Modus scripts directory: {e}"
)
# Update cache atomically
self._scripts_cache = new_cache
@@ -228,11 +241,13 @@ class BashScriptsPlugin(PluginBase):
cache[script_name] = {
"path": script_path,
"name": script_name,
"description": self._get_script_description(script_path),
"description": self._get_script_description(
script_path
),
"type": "discovered",
"executable": os.access(script_path, os.X_OK),
"args": [],
"category": os.path.basename(directory)
"category": os.path.basename(directory),
}
except (PermissionError, FileNotFoundError, OSError) as e:
@@ -244,14 +259,16 @@ class BashScriptsPlugin(PluginBase):
"""Check if a file is a bash script."""
try:
# Check file extension first (most common case)
if file_path.endswith(('.sh', '.bash')):
if file_path.endswith((".sh", ".bash")):
return True
# For files without extension, check shebang
try:
with open(file_path, 'rb') as f:
first_line = f.readline(100).decode('utf-8', errors='ignore')
if first_line.startswith('#!') and ('bash' in first_line or 'sh' in first_line):
with open(file_path, "rb") as f:
first_line = f.readline(100).decode("utf-8", errors="ignore")
if first_line.startswith("#!") and (
"bash" in first_line or "sh" in first_line
):
return True
except (PermissionError, FileNotFoundError, UnicodeDecodeError):
pass
@@ -263,13 +280,13 @@ class BashScriptsPlugin(PluginBase):
def _get_script_description(self, script_path: str) -> str:
"""Extract description from script comments."""
try:
with open(script_path, 'r', encoding='utf-8', errors='ignore') as f:
with open(script_path, "r", encoding="utf-8", errors="ignore") as f:
lines = f.readlines()
# Look for description in first few comment lines
for line in lines[:10]:
line = line.strip()
if line.startswith('#') and not line.startswith('#!'):
if line.startswith("#") and not line.startswith("#!"):
# Remove leading # and whitespace
desc = line[1:].strip()
if desc and len(desc) > 5: # Meaningful description
@@ -284,7 +301,9 @@ class BashScriptsPlugin(PluginBase):
query = query_string.strip()
# Start background update if needed (non-blocking)
if not self._scripts_cache or (time.time() - self._last_cache_update > self._cache_update_interval):
if not self._scripts_cache or (
time.time() - self._last_cache_update > self._cache_update_interval
):
self._start_background_cache_update()
results = []
@@ -313,11 +332,15 @@ class BashScriptsPlugin(PluginBase):
max_results = 20
# Sort scripts by name for consistent ordering
sorted_scripts = sorted(self._scripts_cache.items(), key=lambda x: x[1].get("name", ""))
sorted_scripts = sorted(
self._scripts_cache.items(), key=lambda x: x[1].get("name", "")
)
# Add scripts (limit to max_results)
for script_name, script_info in sorted_scripts:
script_results = self._create_script_results_with_args(script_name, script_info, 0.8)
script_results = self._create_script_results_with_args(
script_name, script_info, 0.8
)
for script_result in script_results:
if len(results) < max_results:
results.append(script_result)
@@ -345,7 +368,10 @@ class BashScriptsPlugin(PluginBase):
description_lower = script_info.get("description", "").lower()
# Skip if no match at all
if query_lower not in script_name_lower and query_lower not in description_lower:
if (
query_lower not in script_name_lower
and query_lower not in description_lower
):
continue
# Categorize matches
@@ -359,11 +385,15 @@ class BashScriptsPlugin(PluginBase):
description_matches.append((script_name, script_info, 0.5))
# Combine results in priority order
all_matches = exact_matches + prefix_matches + partial_matches + description_matches
all_matches = (
exact_matches + prefix_matches + partial_matches + description_matches
)
# Convert to Result objects
for script_name, script_info, relevance in all_matches:
script_results = self._create_script_results_with_args(script_name, script_info, relevance)
script_results = self._create_script_results_with_args(
script_name, script_info, relevance
)
for script_result in script_results:
if len(results) < max_results:
results.append(script_result)
@@ -387,10 +417,12 @@ class BashScriptsPlugin(PluginBase):
description_lower = info["description"].lower()
# Check for exact match or partial match
if (query_lower == cmd_lower or
cmd_lower.startswith(query_lower) or
query_lower in cmd_lower or
query_lower in description_lower):
if (
query_lower == cmd_lower
or cmd_lower.startswith(query_lower)
or query_lower in cmd_lower
or query_lower in description_lower
):
relevance = 1.0 if query_lower == cmd_lower else 0.9
@@ -401,7 +433,11 @@ class BashScriptsPlugin(PluginBase):
action=self._create_power_action(info["profile"]),
relevance=relevance,
plugin_name=self.display_name,
data={"command": cmd, "profile": info["profile"], "type": "power_manager"}
data={
"command": cmd,
"profile": info["profile"],
"type": "power_manager",
},
)
results.append(result)
@@ -419,7 +455,11 @@ class BashScriptsPlugin(PluginBase):
action=self._create_power_action(info["profile"]),
relevance=0.8,
plugin_name=self.display_name,
data={"command": cmd, "profile": info["profile"], "type": "power_manager"}
data={
"command": cmd,
"profile": info["profile"],
"type": "power_manager",
},
)
results.append(result)
@@ -427,6 +467,7 @@ class BashScriptsPlugin(PluginBase):
def _create_power_action(self, profile: str):
"""Create an action function for setting power profile."""
def action():
success = self._set_power_profile(profile)
if success:
@@ -439,16 +480,21 @@ class BashScriptsPlugin(PluginBase):
if app and hasattr(app, "launcher"):
launcher = app.launcher
if launcher and hasattr(launcher, "search_entry"):
def clear_search():
launcher.search_entry.set_text("")
return False
# Use a small delay to ensure the action completes first
GLib.timeout_add(50, clear_search)
except Exception as e:
print(f"BashScriptsPlugin: Could not clear search query: {e}")
return action
def _create_script_result(self, script_name: str, script_info: Dict, relevance: float) -> Result:
def _create_script_result(
self, script_name: str, script_info: Dict, relevance: float
) -> Result:
"""Create a Result object for a script."""
script_path = script_info.get("path", "")
description = script_info.get("description", "")
@@ -465,7 +511,9 @@ class BashScriptsPlugin(PluginBase):
if not executable:
subtitle_parts.append("(not executable)")
subtitle = " | ".join(subtitle_parts) if subtitle_parts else f"Execute: {script_name}"
subtitle = (
" | ".join(subtitle_parts) if subtitle_parts else f"Execute: {script_name}"
)
# Choose icon based on script type and status
if not executable:
@@ -482,10 +530,16 @@ class BashScriptsPlugin(PluginBase):
action=self._create_script_action(script_name, script_info),
relevance=relevance,
plugin_name=self.display_name,
data={"script_name": script_name, "script_path": script_path, "type": script_type}
data={
"script_name": script_name,
"script_path": script_path,
"type": script_type,
},
)
def _create_script_results_with_args(self, script_name: str, script_info: Dict, relevance: float) -> List[Result]:
def _create_script_results_with_args(
self, script_name: str, script_info: Dict, relevance: float
) -> List[Result]:
"""Create multiple Result objects for scripts that support arguments."""
results = []
@@ -495,7 +549,7 @@ class BashScriptsPlugin(PluginBase):
variants = [
("-rgb", "Pick RGB color"),
("-hex", "Pick HEX color"),
("-hsv", "Pick HSV color")
("-hsv", "Pick HSV color"),
]
for arg, desc in variants:
@@ -503,32 +557,48 @@ class BashScriptsPlugin(PluginBase):
title=f"{script_name} {arg}",
subtitle=f"{desc} | [scripts]",
icon_markup=icons.terminal,
action=self._create_script_action_with_args(script_name, script_info, [arg]),
relevance=relevance + 0.1, # Slightly higher relevance for specific variants
action=self._create_script_action_with_args(
script_name, script_info, [arg]
),
relevance=relevance
+ 0.1, # Slightly higher relevance for specific variants
plugin_name=self.display_name,
data={"script_name": script_name, "script_path": script_info.get("path", ""), "type": script_info.get("type", "discovered"), "args": [arg]}
data={
"script_name": script_name,
"script_path": script_info.get("path", ""),
"type": script_info.get("type", "discovered"),
"args": [arg],
},
)
results.append(variant_result)
else:
# For other scripts, create the basic result
basic_result = self._create_script_result(script_name, script_info, relevance)
basic_result = self._create_script_result(
script_name, script_info, relevance
)
results.append(basic_result)
return results
def _create_script_action(self, script_name: str, script_info: Dict):
"""Create an action function for executing a script."""
def action():
self._execute_script(script_name, script_info)
return action
def _create_script_action_with_args(self, script_name: str, script_info: Dict, args: List[str]):
def _create_script_action_with_args(
self, script_name: str, script_info: Dict, args: List[str]
):
"""Create an action function for executing a script with specific arguments."""
def action():
# Create a modified script_info with the specific arguments
modified_script_info = script_info.copy()
modified_script_info["args"] = args
self._execute_script(script_name, modified_script_info)
return action
def _execute_script(self, script_name: str, script_info: Dict):
@@ -549,4 +619,3 @@ class BashScriptsPlugin(PluginBase):
except Exception as e:
print(f"BashScriptsPlugin: Error executing script '{script_name}': {e}")
modules/launcher/plugins/calculator.py
-2
View File
@@ -95,8 +95,6 @@ class CalculatorPlugin(PluginBase):
self._conversion_cache[cache_key] = result
subtitle = f"{value} {from_unit} = {result:.6g} {to_unit}"
return [
Result(
title=f"{result:.6g} {to_unit}",
modules/launcher/plugins/clipboard.py
+3 -3
View File
@@ -380,9 +380,9 @@ class ClipboardPlugin(PluginBase):
result = Result(
title=display_text,
subtitle="Text from clipboard",
description=content
if len(content) <= 100
else content[:97] + "...",
description=(
content if len(content) <= 100 else content[:97] + "..."
),
icon_name="edit-paste",
relevance=relevance,
plugin_name=self.name,
modules/launcher/plugins/process.py
-1
View File
@@ -342,7 +342,6 @@ class ProcessPlugin(PluginBase):
if self._kill_process(pid, silent=True):
killed_count += 1
except Exception as e:
print(f"✗ Error killing process group: {e}")
modules/launcher/plugins/reminders.py
+3 -2
View File
@@ -408,8 +408,9 @@ class RemindersPlugin(PluginBase):
target_time_str
})",
icon_markup=icons.timer_on,
action=lambda ts=time_str,
msg=message: self._create_and_confirm_reminder(ts, msg),
action=lambda ts=time_str, msg=message: self._create_and_confirm_reminder(
ts, msg
),
relevance=1.0,
plugin_name=self.display_name,
data={
modules/launcher/plugins/screencapture.py
+3 -2
View File
@@ -466,8 +466,9 @@ class ScreencapturePlugin(PluginBase):
title=f"Convert {file_param} to {format_type.upper()}",
subtitle=f"Convert specified file to {format_type} format",
icon_markup=screenrecord,
action=lambda fp=file_param,
ft=format_type: self._run_script_with_file(ft, fp),
action=lambda fp=file_param, ft=format_type: self._run_script_with_file(
ft, fp
),
relevance=1.0,
plugin_name=self.display_name,
)
modules/launcher/plugins/system.py
+27 -16
View File
@@ -57,7 +57,9 @@ class SystemPlugin(PluginBase):
self._bin_cache = set(cache_data.get("binaries", []))
self._last_bin_update = cache_data.get("last_update", 0)
else:
print("SystemPlugin: No cache file found, will build cache in background")
print(
"SystemPlugin: No cache file found, will build cache in background"
)
except Exception as e:
print(f"SystemPlugin: Error loading binary cache: {e}")
self._bin_cache = set()
@@ -72,7 +74,7 @@ class SystemPlugin(PluginBase):
cache_data = {
"binaries": sorted(list(self._bin_cache)),
"last_update": self._last_bin_update,
"cache_version": "1.0"
"cache_version": "1.0",
}
with open(self.bin_cache_file, "w", encoding="utf-8") as f:
@@ -85,14 +87,15 @@ class SystemPlugin(PluginBase):
current_time = time.time()
# Check if cache needs updating
if (current_time - self._last_bin_update > self._bin_update_interval or
not self._bin_cache):
if (
current_time - self._last_bin_update > self._bin_update_interval
or not self._bin_cache
):
if not self._cache_building:
self._cache_building = True
self._cache_thread = threading.Thread(
target=self._build_bin_cache_background,
daemon=True
target=self._build_bin_cache_background, daemon=True
)
self._cache_thread.start()
@@ -113,8 +116,9 @@ class SystemPlugin(PluginBase):
# Use os.scandir for better performance than os.listdir
with os.scandir(path) as entries:
for entry in entries:
if (entry.is_file(follow_symlinks=False) and
os.access(entry.path, os.X_OK)):
if entry.is_file(follow_symlinks=False) and os.access(
entry.path, os.X_OK
):
new_cache.add(entry.name)
except (PermissionError, FileNotFoundError, OSError):
continue
@@ -126,7 +130,6 @@ class SystemPlugin(PluginBase):
# Save to disk
self._save_bin_cache()
except Exception as e:
print(f"SystemPlugin: Error building binary cache: {e}")
finally:
@@ -151,7 +154,9 @@ class SystemPlugin(PluginBase):
# Check system binaries
# Start background update if needed (non-blocking) - but only if cache is empty or very old
if not self._bin_cache or (time.time() - self._last_bin_update > self._bin_update_interval):
if not self._bin_cache or (
time.time() - self._last_bin_update > self._bin_update_interval
):
self._start_background_cache_update()
# Optimize search with early termination and result limiting
@@ -174,19 +179,25 @@ class SystemPlugin(PluginBase):
display_command = full_command
command_to_execute = full_command
relevance = 1.0
exact_matches.append((binary, display_command, command_to_execute, relevance))
exact_matches.append(
(binary, display_command, command_to_execute, relevance)
)
elif binary_lower.startswith(binary_query):
# Prefix match - high priority
display_command = binary
command_to_execute = binary
relevance = 0.9
prefix_matches.append((binary, display_command, command_to_execute, relevance))
prefix_matches.append(
(binary, display_command, command_to_execute, relevance)
)
else:
# Partial match - lower priority
display_command = binary
command_to_execute = binary
relevance = 0.7
partial_matches.append((binary, display_command, command_to_execute, relevance))
partial_matches.append(
(binary, display_command, command_to_execute, relevance)
)
# Early termination if we have enough good matches
if len(exact_matches) + len(prefix_matches) >= max_results:
@@ -196,7 +207,9 @@ class SystemPlugin(PluginBase):
all_matches = exact_matches + prefix_matches + partial_matches
# Convert to Result objects (limit to max_results)
for binary, display_command, command_to_execute, relevance in all_matches[:max_results]:
for binary, display_command, command_to_execute, relevance in all_matches[
:max_results
]:
result = Result(
title=display_command,
subtitle=f"Execute: {display_command}",
@@ -210,8 +223,6 @@ class SystemPlugin(PluginBase):
return results # Already sorted by priority
def _create_action(self, command: Union[str, List[str]]):
"""Create an action function for the given command."""
modules/launcher/plugins/tmux.py
+72 -46
View File
@@ -46,8 +46,7 @@ class TmuxPlugin(PluginBase):
"""Start background thread to refresh session cache."""
if not self.refresh_thread or not self.refresh_thread.is_alive():
self.refresh_thread = threading.Thread(
target=self._refresh_sessions_background,
daemon=True
target=self._refresh_sessions_background, daemon=True
)
self.refresh_thread.start()
@@ -59,7 +58,7 @@ class TmuxPlugin(PluginBase):
if current_time - self._last_cache_update > self._cache_ttl:
self._sessions_cache = self._get_tmux_sessions()
self._last_cache_update = current_time
# Sleep for a short interval
self.stop_refresh.wait(1)
except Exception as e:
@@ -73,12 +72,18 @@ class TmuxPlugin(PluginBase):
["tmux", "list-sessions", "-F", "#{session_name}"],
capture_output=True,
text=True,
timeout=5
timeout=5,
)
if result.returncode == 0:
return [s.strip() for s in result.stdout.strip().split('\n') if s.strip()]
return [
s.strip() for s in result.stdout.strip().split("\n") if s.strip()
]
return []
except (subprocess.TimeoutExpired, subprocess.CalledProcessError, FileNotFoundError) as e:
except (
subprocess.TimeoutExpired,
subprocess.CalledProcessError,
FileNotFoundError,
) as e:
print(f"TmuxPlugin: Error getting tmux sessions: {e}")
return []
@@ -92,28 +97,32 @@ class TmuxPlugin(PluginBase):
if current_time - self._last_cache_update > self._cache_ttl:
self._sessions_cache = self._get_tmux_sessions()
self._last_cache_update = current_time
sessions = self._sessions_cache
# Handle specific commands
if query.startswith("new ") or query.startswith("create "):
session_name = query.split(" ", 1)[1].strip() if " " in query else ""
results.append(self._create_new_session_result(session_name))
elif query.startswith("kill ") or query.startswith("delete "):
session_name = query.split(" ", 1)[1].strip() if " " in query else ""
if session_name:
matching_sessions = [s for s in sessions if session_name.lower() in s.lower()]
matching_sessions = [
s for s in sessions if session_name.lower() in s.lower()
]
for session in matching_sessions:
results.append(self._create_kill_session_result(session))
elif query.startswith("rename "):
parts = query.split(" ", 2)
if len(parts) >= 3:
old_name, new_name = parts[1], parts[2]
if old_name in sessions:
results.append(self._create_rename_session_result(old_name, new_name))
results.append(
self._create_rename_session_result(old_name, new_name)
)
else:
# Show existing sessions for attachment
if sessions:
@@ -122,13 +131,19 @@ class TmuxPlugin(PluginBase):
filtered_sessions = [s for s in sessions if query in s.lower()]
else:
filtered_sessions = sessions
for session in filtered_sessions:
results.append(self._create_attach_session_result(session))
# Always show option to create new session
if not query or "new" in query or "create" in query:
results.append(self._create_new_session_result(query if query and not any(cmd in query for cmd in ["new", "create"]) else ""))
results.append(
self._create_new_session_result(
query
if query and not any(cmd in query for cmd in ["new", "create"])
else ""
)
)
return results
@@ -140,7 +155,7 @@ class TmuxPlugin(PluginBase):
icon_markup=icons.terminal,
action=lambda: self._attach_to_session(session_name),
relevance=0.9,
data={"type": "attach", "session": session_name}
data={"type": "attach", "session": session_name},
)
def _create_new_session_result(self, session_name: str = "") -> Result:
@@ -152,7 +167,7 @@ class TmuxPlugin(PluginBase):
icon_markup=icons.plus,
action=lambda: self._create_session(session_name),
relevance=0.8,
data={"type": "create", "session": session_name}
data={"type": "create", "session": session_name},
)
def _create_kill_session_result(self, session_name: str) -> Result:
@@ -163,7 +178,7 @@ class TmuxPlugin(PluginBase):
icon_markup=icons.trash,
action=lambda: self._kill_session(session_name),
relevance=0.7,
data={"type": "kill", "session": session_name}
data={"type": "kill", "session": session_name},
)
def _create_rename_session_result(self, old_name: str, new_name: str) -> Result:
@@ -174,14 +189,15 @@ class TmuxPlugin(PluginBase):
icon_markup=icons.config,
action=lambda: self._rename_session(old_name, new_name),
relevance=0.6,
data={"type": "rename", "old_session": old_name, "new_session": new_name}
data={"type": "rename", "old_session": old_name, "new_session": new_name},
)
def _attach_to_session(self, session_name: str):
"""Attach to an existing tmux session."""
try:
terminal_cmd = self._get_terminal_command(f"tmux attach-session -t '{session_name}'")
terminal_cmd = self._get_terminal_command(
f"tmux attach-session -t '{session_name}'"
)
exec_shell_command_async(terminal_cmd)
print(f"TmuxPlugin: Attaching to session '{session_name}'")
except Exception as e:
@@ -197,25 +213,25 @@ class TmuxPlugin(PluginBase):
while str(counter) in existing_sessions:
counter += 1
session_name = str(counter)
# Clean the session name
clean_name = session_name.strip().replace(" ", "_")
# Create session
subprocess.run(
["tmux", "new-session", "-d", "-s", clean_name],
check=True,
timeout=10
["tmux", "new-session", "-d", "-s", clean_name], check=True, timeout=10
)
# Launch terminal and attach
terminal_cmd = self._get_terminal_command(f"tmux attach-session -t '{clean_name}'")
terminal_cmd = self._get_terminal_command(
f"tmux attach-session -t '{clean_name}'"
)
exec_shell_command_async(terminal_cmd)
# Refresh cache
self._sessions_cache = self._get_tmux_sessions()
self._last_cache_update = time.time()
print(f"TmuxPlugin: Created and attached to session '{clean_name}'")
except Exception as e:
print(f"TmuxPlugin: Error creating session '{session_name}': {e}")
@@ -224,15 +240,13 @@ class TmuxPlugin(PluginBase):
"""Kill a tmux session."""
try:
subprocess.run(
["tmux", "kill-session", "-t", session_name],
check=True,
timeout=10
["tmux", "kill-session", "-t", session_name], check=True, timeout=10
)
# Refresh cache
self._sessions_cache = self._get_tmux_sessions()
self._last_cache_update = time.time()
print(f"TmuxPlugin: Killed session '{session_name}'")
except Exception as e:
print(f"TmuxPlugin: Error killing session '{session_name}': {e}")
@@ -244,32 +258,39 @@ class TmuxPlugin(PluginBase):
subprocess.run(
["tmux", "rename-session", "-t", old_name, clean_name],
check=True,
timeout=10
timeout=10,
)
# Refresh cache
self._sessions_cache = self._get_tmux_sessions()
self._last_cache_update = time.time()
print(f"TmuxPlugin: Renamed session '{old_name}' to '{clean_name}'")
except Exception as e:
print(f"TmuxPlugin: Error renaming session '{old_name}' to '{new_name}': {e}")
print(
f"TmuxPlugin: Error renaming session '{old_name}' to '{new_name}': {e}"
)
def _get_terminal_command(self, cmd: str) -> str:
"""Get terminal command based on configured terminal or available terminals."""
# First try to use the configured terminal command
if hasattr(data, 'TERMINAL_COMMAND') and data.TERMINAL_COMMAND:
if hasattr(data, "TERMINAL_COMMAND") and data.TERMINAL_COMMAND:
parts = data.TERMINAL_COMMAND.split()
terminal = parts[0]
try:
# Check if the configured terminal is available
subprocess.run(["which", terminal], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
subprocess.run(
["which", terminal],
check=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
return f"{data.TERMINAL_COMMAND} {cmd}"
except subprocess.CalledProcessError:
# If configured terminal is not available, fall back to defaults
pass
# Fallback to checking available terminals
terminals = [
("kitty", f"kitty -e {cmd}"),
@@ -279,14 +300,19 @@ class TmuxPlugin(PluginBase):
("konsole", f"konsole -e {cmd}"),
("xfce4-terminal", f"xfce4-terminal -e '{cmd}'"),
]
for term, term_cmd in terminals:
try:
# Check if terminal is available
subprocess.run(["which", term], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
subprocess.run(
["which", term],
check=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
return term_cmd
except subprocess.CalledProcessError:
continue
# Default fallback
return f"kitty -e {cmd}"
modules/launcher/plugins/wallpaper.py
+18 -15
View File
@@ -132,6 +132,7 @@ class WallpaperPlugin(PluginBase):
if app and hasattr(app, "launcher"):
launcher = app.launcher
if launcher and hasattr(launcher, "search_entry"):
def clear_and_refresh():
# Clear the search entry to just the trigger
launcher.search_entry.set_text("wall ")
@@ -150,11 +151,9 @@ class WallpaperPlugin(PluginBase):
import gc
for obj in gc.get_objects():
if (
hasattr(obj, "__class__")
and obj.__class__.__name__ == "Launcher"
):
if hasattr(obj, "__class__") and obj.__class__.__name__ == "Launcher":
if hasattr(obj, "search_entry") and hasattr(obj, "_perform_search"):
def clear_and_refresh():
obj.search_entry.set_text("wall ")
obj.search_entry.set_position(-1)
@@ -554,8 +553,9 @@ class WallpaperPlugin(PluginBase):
status_text
}",
icon_markup=icons.palette,
action=lambda c=hex_color,
s=scheme: self._apply_hex_color_direct(c, s),
action=lambda c=hex_color, s=scheme: self._apply_hex_color_direct(
c, s
),
relevance=1.0,
plugin_name=self.display_name,
data={
@@ -595,10 +595,11 @@ class WallpaperPlugin(PluginBase):
status_text
}",
icon_markup=icons.palette,
action=lambda c=hex_color,
s=scheme: self._apply_hex_color_direct(c, s)
if not matugen_enabled
else None,
action=lambda c=hex_color, s=scheme: (
self._apply_hex_color_direct(c, s)
if not matugen_enabled
else None
),
relevance=0.9,
plugin_name=self.display_name,
data={
@@ -683,11 +684,13 @@ class WallpaperPlugin(PluginBase):
title=f"Random Hex Color{indicator_text}",
subtitle=f"Generate and apply random color • {status_text}",
icon_markup=icons.palette,
action=lambda s=scheme: self._apply_hex_color_direct(
self._generate_random_hex_color(), s
)
if not matugen_enabled
else None,
action=lambda s=scheme: (
self._apply_hex_color_direct(
self._generate_random_hex_color(), s
)
if not matugen_enabled
else None
),
relevance=0.8,
plugin_name=self.display_name,
data={