282 lines
10 KiB
Python
282 lines
10 KiB
Python
"""
|
|
EU-Utility - Plugin Manager
|
|
|
|
Handles discovery, loading, and lifecycle of plugins.
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import json
|
|
import importlib
|
|
import importlib.util
|
|
from pathlib import Path
|
|
from typing import Dict, List, Type, Optional
|
|
|
|
from core.base_plugin import BasePlugin
|
|
|
|
|
|
class PluginManager:
|
|
"""Manages loading and lifecycle of plugins."""
|
|
|
|
PLUGIN_DIRS = [
|
|
"plugins", # Built-in plugins
|
|
"user_plugins", # User-installed plugins
|
|
]
|
|
|
|
def __init__(self, overlay_window):
|
|
self.overlay = overlay_window
|
|
self.plugins: Dict[str, BasePlugin] = {}
|
|
self.plugin_classes: Dict[str, Type[BasePlugin]] = {}
|
|
self.config = self._load_config()
|
|
|
|
# Add plugin dirs and project root to path
|
|
# Project root needed for 'plugins.base_plugin' imports
|
|
project_root = Path(__file__).parent.parent.absolute()
|
|
if str(project_root) not in sys.path:
|
|
sys.path.insert(0, str(project_root))
|
|
|
|
for plugin_dir in self.PLUGIN_DIRS:
|
|
path = Path(plugin_dir).absolute()
|
|
if path.exists() and str(path) not in sys.path:
|
|
sys.path.insert(0, str(path))
|
|
|
|
# Ensure plugins package is loaded
|
|
try:
|
|
import plugins
|
|
except ImportError:
|
|
pass # plugins package will be loaded when needed
|
|
|
|
def _load_config(self) -> dict:
|
|
"""Load plugin configuration."""
|
|
config_path = Path("config/plugins.json")
|
|
if config_path.exists():
|
|
try:
|
|
return json.loads(config_path.read_text())
|
|
except json.JSONDecodeError:
|
|
pass
|
|
# Default: no plugins enabled - all disabled by default
|
|
return {"enabled": [], "settings": {}}
|
|
|
|
def save_config(self) -> None:
|
|
"""Save plugin configuration."""
|
|
config_path = Path("config/plugins.json")
|
|
config_path.parent.mkdir(parents=True, exist_ok=True)
|
|
config_path.write_text(json.dumps(self.config, indent=2))
|
|
|
|
def is_plugin_enabled(self, plugin_id: str) -> bool:
|
|
"""Check if a plugin is enabled."""
|
|
# Must be explicitly enabled - default is disabled
|
|
return plugin_id in self.config.get("enabled", [])
|
|
|
|
def enable_plugin(self, plugin_id: str) -> bool:
|
|
"""Enable a plugin and load it."""
|
|
if plugin_id in self.config.get("enabled", []):
|
|
return True
|
|
|
|
# Add to enabled list
|
|
if "enabled" not in self.config:
|
|
self.config["enabled"] = []
|
|
self.config["enabled"].append(plugin_id)
|
|
self.save_config()
|
|
|
|
# Load the plugin
|
|
plugin_class = self.plugin_classes.get(plugin_id)
|
|
if plugin_class:
|
|
return self.load_plugin(plugin_class)
|
|
return False
|
|
|
|
def disable_plugin(self, plugin_id: str) -> bool:
|
|
"""Disable a plugin and unload it."""
|
|
if "enabled" in self.config and plugin_id in self.config["enabled"]:
|
|
self.config["enabled"].remove(plugin_id)
|
|
self.save_config()
|
|
|
|
# Unload if loaded
|
|
if plugin_id in self.plugins:
|
|
self.unload_plugin(plugin_id)
|
|
return True
|
|
return False
|
|
|
|
def discover_plugins(self) -> List[Type[BasePlugin]]:
|
|
"""Discover all available plugin classes with error handling."""
|
|
discovered = []
|
|
|
|
# First, ensure plugins package is loaded
|
|
self._ensure_plugins_package()
|
|
|
|
for plugin_dir in self.PLUGIN_DIRS:
|
|
base_path = Path(plugin_dir)
|
|
if not base_path.exists():
|
|
continue
|
|
|
|
# Find plugin folders
|
|
for item in base_path.iterdir():
|
|
if not item.is_dir():
|
|
continue
|
|
if item.name.startswith("__"):
|
|
continue
|
|
if item.name == "__pycache__":
|
|
continue
|
|
|
|
plugin_file = item / "plugin.py"
|
|
|
|
if not plugin_file.exists():
|
|
continue
|
|
|
|
try:
|
|
# Load the plugin module using normal import
|
|
# This ensures proper package resolution
|
|
module_name = f"{plugin_dir}.{item.name}.plugin"
|
|
|
|
if module_name in sys.modules:
|
|
module = sys.modules[module_name]
|
|
else:
|
|
# Try normal import first
|
|
try:
|
|
module = importlib.import_module(module_name)
|
|
except ImportError:
|
|
# Fall back to spec loading
|
|
spec = importlib.util.spec_from_file_location(
|
|
module_name, plugin_file
|
|
)
|
|
if not spec or not spec.loader:
|
|
continue
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[module_name] = module
|
|
spec.loader.exec_module(module)
|
|
|
|
# Find plugin class
|
|
for attr_name in dir(module):
|
|
attr = getattr(module, attr_name)
|
|
if (isinstance(attr, type) and
|
|
issubclass(attr, BasePlugin) and
|
|
attr != BasePlugin and
|
|
not attr.__name__.startswith("Base")):
|
|
discovered.append(attr)
|
|
break
|
|
|
|
except Exception as e:
|
|
print(f"[PluginManager] Failed to discover {item.name}: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
continue
|
|
|
|
return discovered
|
|
|
|
def _ensure_plugins_package(self):
|
|
"""Ensure the plugins package is loaded in sys.modules."""
|
|
if 'plugins' not in sys.modules:
|
|
try:
|
|
import plugins
|
|
except ImportError:
|
|
# Create a minimal plugins module if it doesn't exist
|
|
import types
|
|
plugins_module = types.ModuleType('plugins')
|
|
plugins_module.__path__ = [str(Path('plugins').absolute())]
|
|
sys.modules['plugins'] = plugins_module
|
|
|
|
def load_plugin(self, plugin_class: Type[BasePlugin]) -> bool:
|
|
"""Instantiate and initialize a plugin with error handling."""
|
|
try:
|
|
plugin_id = f"{plugin_class.__module__}.{plugin_class.__name__}"
|
|
|
|
# Check if already loaded
|
|
if plugin_id in self.plugins:
|
|
return True
|
|
|
|
# Check if disabled
|
|
if plugin_id in self.config.get("disabled", []):
|
|
print(f"[PluginManager] Skipping disabled plugin: {plugin_class.name}")
|
|
return False
|
|
|
|
# Get plugin config
|
|
plugin_config = self.config.get("settings", {}).get(plugin_id, {})
|
|
|
|
# Create instance
|
|
try:
|
|
instance = plugin_class(self.overlay, plugin_config)
|
|
except Exception as e:
|
|
print(f"[PluginManager] Failed to create {plugin_class.name}: {e}")
|
|
return False
|
|
|
|
# Initialize with error handling
|
|
try:
|
|
instance.initialize()
|
|
except Exception as e:
|
|
print(f"[PluginManager] Failed to initialize {plugin_class.name}: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return False
|
|
|
|
# Store
|
|
self.plugins[plugin_id] = instance
|
|
self.plugin_classes[plugin_id] = plugin_class
|
|
|
|
print(f"[PluginManager] ✓ Loaded: {instance.name} v{instance.version}")
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"[PluginManager] Failed to load {plugin_class.__name__}: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return False
|
|
|
|
def load_all_plugins(self) -> None:
|
|
"""Load only enabled plugins."""
|
|
discovered = self.discover_plugins()
|
|
|
|
for plugin_class in discovered:
|
|
plugin_id = f"{plugin_class.__module__}.{plugin_class.__name__}"
|
|
|
|
# Only load if explicitly enabled
|
|
if self.is_plugin_enabled(plugin_id):
|
|
self.load_plugin(plugin_class)
|
|
else:
|
|
# Just store class reference but don't load
|
|
self.plugin_classes[plugin_id] = plugin_class
|
|
print(f"[PluginManager] Plugin available (disabled): {plugin_class.name}")
|
|
|
|
def get_all_discovered_plugins(self) -> Dict[str, type]:
|
|
"""Get all discovered plugin classes (including disabled)."""
|
|
return self.plugin_classes.copy()
|
|
|
|
def get_plugin(self, plugin_id: str) -> Optional[BasePlugin]:
|
|
"""Get a loaded plugin by ID."""
|
|
return self.plugins.get(plugin_id)
|
|
|
|
def get_plugin_ui(self, plugin_id: str):
|
|
"""Get UI widget for a plugin."""
|
|
plugin = self.plugins.get(plugin_id)
|
|
if plugin:
|
|
return plugin.get_ui()
|
|
return None
|
|
|
|
def get_all_plugins(self) -> Dict[str, BasePlugin]:
|
|
"""Get all loaded plugins."""
|
|
return self.plugins.copy()
|
|
|
|
def unload_plugin(self, plugin_id: str) -> None:
|
|
"""Shutdown and unload a plugin."""
|
|
if plugin_id in self.plugins:
|
|
try:
|
|
self.plugins[plugin_id].shutdown()
|
|
except Exception as e:
|
|
print(f"Error shutting down {plugin_id}: {e}")
|
|
del self.plugins[plugin_id]
|
|
|
|
def shutdown_all(self) -> None:
|
|
"""Shutdown all plugins."""
|
|
for plugin_id in list(self.plugins.keys()):
|
|
self.unload_plugin(plugin_id)
|
|
|
|
def trigger_hotkey(self, hotkey: str) -> bool:
|
|
"""Trigger plugin by hotkey. Returns True if handled."""
|
|
for plugin_id, plugin in self.plugins.items():
|
|
if plugin.hotkey == hotkey and plugin.enabled:
|
|
try:
|
|
plugin.on_hotkey()
|
|
return True
|
|
except Exception as e:
|
|
print(f"Error triggering hotkey for {plugin_id}: {e}")
|
|
return False
|