♻️ refactor(launcher, plugin_manager): build trigger map during activation, improve plugin loading
This commit is contained in:
3 files changed
+48
-126
No files matched your search
+11
-44
@@ -367,41 +367,10 @@ class Launcher(Window):
|
||||
print(f"Error in triggered plugin {triggered_plugin.name}: {e}")
|
||||
all_results = []
|
||||
else:
|
||||
# No trigger detected - search for applications and other general content
|
||||
# No trigger detected - only show trigger suggestions
|
||||
self.triggered_plugin = None
|
||||
self.active_trigger = ""
|
||||
|
||||
all_results = []
|
||||
|
||||
# For general queries, load applications plugin on-demand
|
||||
if "applications" not in self.plugin_manager.active_plugins:
|
||||
app_plugin = self.plugin_manager.get_plugin_for_trigger("app")
|
||||
if app_plugin:
|
||||
try:
|
||||
app_results = app_plugin.query(query)
|
||||
all_results.extend(app_results)
|
||||
except Exception as e:
|
||||
print(f"Error in applications plugin: {e}")
|
||||
else:
|
||||
# Applications plugin already loaded, use it
|
||||
app_plugin = self.plugin_manager.plugins.get("applications")
|
||||
if app_plugin:
|
||||
try:
|
||||
app_results = app_plugin.query(query)
|
||||
all_results.extend(app_results)
|
||||
except Exception as e:
|
||||
print(f"Error in applications plugin: {e}")
|
||||
|
||||
# Search across any other loaded plugins for general queries
|
||||
for plugin in self.plugin_manager.get_active_plugins():
|
||||
if plugin.name != "applications": # Skip applications as we handled it above
|
||||
try:
|
||||
if plugin.handles_query(query):
|
||||
plugin_results = plugin.query(query)
|
||||
all_results.extend(plugin_results)
|
||||
except Exception as e:
|
||||
print(f"Error in plugin {plugin.name}: {e}")
|
||||
|
||||
# Show trigger suggestions if query matches trigger prefixes
|
||||
trigger_suggestions = self._get_trigger_suggestions(query)
|
||||
all_results = trigger_suggestions
|
||||
@@ -476,7 +445,7 @@ class Launcher(Window):
|
||||
|
||||
def _detect_trigger(self, query: str) -> Tuple[Optional[object], str]:
|
||||
"""
|
||||
Detect if query starts with a trigger keyword and lazy load the plugin.
|
||||
Detect if query starts with a trigger keyword.
|
||||
|
||||
Args:
|
||||
query: The search query
|
||||
@@ -487,15 +456,7 @@ class Launcher(Window):
|
||||
if not query.strip():
|
||||
return None, ""
|
||||
|
||||
# First check if any trigger matches without loading plugins
|
||||
plugin_name, trigger = self.plugin_manager.find_trigger_plugin(query)
|
||||
if plugin_name and trigger:
|
||||
# Load the plugin on demand
|
||||
plugin = self.plugin_manager.get_plugin_for_trigger(trigger)
|
||||
if plugin:
|
||||
return plugin, trigger
|
||||
|
||||
# Fallback: check already loaded plugins
|
||||
# Check all plugins for triggers
|
||||
for plugin in self.plugin_manager.get_active_plugins():
|
||||
trigger = plugin.get_active_trigger(query)
|
||||
if trigger:
|
||||
@@ -620,8 +581,14 @@ class Launcher(Window):
|
||||
trigger_part = parts[0]
|
||||
query_part = parts[1] if len(parts) > 1 else ""
|
||||
|
||||
# Find the plugin that handles this trigger (lazy load if needed)
|
||||
triggered_plugin = self.plugin_manager.get_plugin_for_trigger(trigger_part)
|
||||
# Find the plugin that handles this trigger
|
||||
triggered_plugin = None
|
||||
|
||||
for plugin in self.plugin_manager.get_active_plugins():
|
||||
trigger = plugin.get_active_trigger(f"{trigger_part} ")
|
||||
if trigger:
|
||||
triggered_plugin = plugin
|
||||
break
|
||||
|
||||
if not triggered_plugin:
|
||||
print(f"No plugin found for trigger: {trigger_part}")
|
||||
|
||||
@@ -8,23 +8,22 @@ from modules.launcher.plugin_base import PluginBase
|
||||
|
||||
class PluginManager:
|
||||
"""
|
||||
Manages launcher plugins with lazy loading support.
|
||||
Manages launcher plugins.
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self.plugins: Dict[str, PluginBase] = {} # Actually loaded plugin instances
|
||||
self.plugin_classes: Dict[str, Type[PluginBase]] = {} # Available plugin classes
|
||||
self.active_plugins: List[str] = [] # List of loaded plugin names
|
||||
self.trigger_map: Dict[str, str] = {} # Maps triggers to plugin names
|
||||
self.plugins: Dict[str, PluginBase] = {}
|
||||
self.plugin_classes: Dict[str, Type[PluginBase]] = {}
|
||||
self.active_plugins: List[str] = []
|
||||
|
||||
# Load built-in plugins (discover classes only)
|
||||
# Load built-in plugins
|
||||
self._load_builtin_plugins()
|
||||
|
||||
# Load external plugins (discover classes only)
|
||||
# Load external plugins
|
||||
self._load_external_plugins()
|
||||
|
||||
# Build trigger mapping without activating plugins
|
||||
self._build_trigger_map()
|
||||
# Activate default plugins
|
||||
self._activate_default_plugins()
|
||||
|
||||
def _load_builtin_plugins(self):
|
||||
"""Load built-in plugins from the plugins directory."""
|
||||
@@ -75,42 +74,35 @@ class PluginManager:
|
||||
except Exception as e:
|
||||
print(f"Failed to load plugin {plugin_name}: {e}")
|
||||
|
||||
def _build_trigger_map(self):
|
||||
"""Build a mapping of triggers to plugin names without loading plugins."""
|
||||
for plugin_name, plugin_class in self.plugin_classes.items():
|
||||
try:
|
||||
# Get triggers from class without instantiating
|
||||
triggers = self._get_class_triggers(plugin_class)
|
||||
for trigger in triggers:
|
||||
self.trigger_map[trigger.lower()] = plugin_name
|
||||
except Exception as e:
|
||||
print(f"Failed to get triggers for plugin {plugin_name}: {e}")
|
||||
def _activate_default_plugins(self):
|
||||
"""Activate default plugins."""
|
||||
default_plugins = [
|
||||
"applications",
|
||||
"calculator",
|
||||
"system",
|
||||
"clipboard",
|
||||
"power",
|
||||
"caffeine",
|
||||
"screencapture",
|
||||
"emoji",
|
||||
"wallpaper",
|
||||
"websearch",
|
||||
"kanban",
|
||||
"calendar",
|
||||
"reminders",
|
||||
"network",
|
||||
"bluetooth",
|
||||
"otp",
|
||||
"password",
|
||||
"bookmarks",
|
||||
"process",
|
||||
]
|
||||
|
||||
def _get_class_triggers(self, plugin_class: Type[PluginBase]) -> List[str]:
|
||||
"""Get triggers from a plugin class without full initialization."""
|
||||
# Check if the plugin class has a class-level triggers attribute
|
||||
if hasattr(plugin_class, 'TRIGGERS'):
|
||||
return plugin_class.TRIGGERS
|
||||
|
||||
# Fallback: create a minimal instance to get triggers
|
||||
try:
|
||||
temp_instance = plugin_class()
|
||||
# Try to get triggers without full initialization if possible
|
||||
if hasattr(temp_instance, '_default_triggers'):
|
||||
triggers = temp_instance._default_triggers
|
||||
else:
|
||||
# Last resort: minimal initialization
|
||||
temp_instance.initialize()
|
||||
triggers = temp_instance.get_triggers()
|
||||
temp_instance.cleanup()
|
||||
return triggers
|
||||
except Exception as e:
|
||||
print(f"Failed to get triggers for {plugin_class.__name__}: {e}")
|
||||
# Fallback: return empty list if we can't get triggers
|
||||
return []
|
||||
for plugin_name in default_plugins:
|
||||
self.activate_plugin(plugin_name)
|
||||
|
||||
def activate_plugin(self, plugin_name: str) -> bool:
|
||||
"""Activate a plugin by name (lazy loading)."""
|
||||
"""Activate a plugin by name."""
|
||||
if plugin_name in self.plugins:
|
||||
# Already activated
|
||||
return True
|
||||
@@ -130,7 +122,6 @@ class PluginManager:
|
||||
self.plugins[plugin_name] = plugin_instance
|
||||
self.active_plugins.append(plugin_name)
|
||||
|
||||
print(f"Lazy loaded plugin: {plugin_name}")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
@@ -164,42 +155,6 @@ class PluginManager:
|
||||
self.plugins[name] for name in self.active_plugins if name in self.plugins
|
||||
]
|
||||
|
||||
def get_plugin_for_trigger(self, trigger: str) -> PluginBase:
|
||||
"""Get plugin instance for a trigger, loading it if necessary."""
|
||||
trigger_lower = trigger.lower()
|
||||
|
||||
# Check if we have a plugin for this trigger
|
||||
plugin_name = self.trigger_map.get(trigger_lower)
|
||||
if not plugin_name:
|
||||
return None
|
||||
|
||||
# Load the plugin if not already loaded
|
||||
if plugin_name not in self.plugins:
|
||||
if not self.activate_plugin(plugin_name):
|
||||
return None
|
||||
|
||||
return self.plugins.get(plugin_name)
|
||||
|
||||
def find_trigger_plugin(self, query: str) -> tuple:
|
||||
"""Find plugin and trigger for a query without loading the plugin."""
|
||||
query_lower = query.lower().strip()
|
||||
|
||||
# Sort triggers by length (longest first) to match more specific triggers first
|
||||
sorted_triggers = sorted(self.trigger_map.items(), key=lambda x: len(x[0]), reverse=True)
|
||||
|
||||
for trigger, plugin_name in sorted_triggers:
|
||||
trigger_lower = trigger.lower()
|
||||
|
||||
# Exact match with trigger
|
||||
if query_lower == trigger_lower:
|
||||
return plugin_name, trigger
|
||||
|
||||
# Match trigger followed by space
|
||||
if query_lower.startswith(trigger_lower + " "):
|
||||
return plugin_name, trigger
|
||||
|
||||
return None, ""
|
||||
|
||||
def get_plugin_names(self) -> List[str]:
|
||||
"""Get list of available plugin names."""
|
||||
return list(self.plugin_classes.keys())
|
||||
|
||||
@@ -58,8 +58,8 @@ class CalculatorPlugin(PluginBase):
|
||||
self._cache_cleanup_interval = 300 # 5 minutes
|
||||
|
||||
def initialize(self):
|
||||
"""Initialize the calculator plugin."""
|
||||
self.set_triggers(["=", "calc"])
|
||||
"""Initialize the files plugin."""
|
||||
self.set_triggers(["="])
|
||||
|
||||
def _cleanup_cache(self):
|
||||
"""Clean up old cache entries."""
|
||||
@@ -95,7 +95,7 @@ class CalculatorPlugin(PluginBase):
|
||||
self._conversion_cache[cache_key] = result
|
||||
subtitle = f"{value} {from_unit} = {result:.6g} {to_unit}"
|
||||
|
||||
|
||||
|
||||
|
||||
return [
|
||||
Result(
|
||||
|
||||
Reference in new issue
Block a user