""" EU-Utility Premium - Plugin API ================================ Plugin API surface for the plugin system. This module defines the contracts that plugins must implement to be loaded by the PluginManager. Example: from premium.plugins.api import PluginAPI, PluginManifest, PluginContext class MyPlugin(PluginAPI): manifest = PluginManifest( name="My Plugin", version="1.0.0", author="Your Name" ) def on_init(self, ctx: PluginContext): self.ctx = ctx ctx.logger.info("Plugin initialized!") def on_activate(self): self.ctx.logger.info("Plugin activated!") """ from __future__ import annotations import logging from abc import ABC, abstractmethod from dataclasses import dataclass, field from datetime import datetime from enum import Enum, auto from pathlib import Path from typing import Any, Callable, Dict, List, Optional, Set, Type, TYPE_CHECKING if TYPE_CHECKING: from PyQt6.QtWidgets import QWidget # ============================================================================= # PERMISSION LEVELS # ============================================================================= class PermissionLevel(Enum): """Permission levels for plugin sandboxing. Plugins must declare which permissions they need. The user must approve these permissions before the plugin runs. """ FILE_READ = auto() # Read files from disk FILE_WRITE = auto() # Write files to disk NETWORK = auto() # Access network (API calls) UI = auto() # Create/manipulate UI widgets MEMORY = auto() # Access game memory (dangerous) PROCESS = auto() # Access other processes SYSTEM = auto() # System-level access (very dangerous) # ============================================================================= # PLUGIN STATE # ============================================================================= class PluginState(Enum): """Lifecycle states for plugins.""" DISCOVERED = auto() # Found but not loaded LOADING = auto() # Currently loading LOADED = auto() # Code loaded, not initialized INITIALIZING = auto() # Currently initializing INACTIVE = auto() # Initialized but not active ACTIVATING = auto() # Currently activating ACTIVE = auto() # Fully active and running DEACTIVATING = auto() # Currently deactivating UNLOADING = auto() # Currently unloading UNLOADED = auto() # Unloaded from memory ERROR = auto() # Error state # ============================================================================= # PLUGIN MANIFEST # ============================================================================= @dataclass class PluginManifest: """Plugin manifest - metadata about a plugin. This is loaded from plugin.json in the plugin directory. Example plugin.json: { "name": "My Plugin", "version": "1.0.0", "author": "Your Name", "description": "Does cool things", "entry_point": "main.py", "permissions": ["file_read", "ui"], "dependencies": { "other_plugin": ">=1.0.0" } } """ name: str version: str author: str description: str = "" entry_point: str = "main.py" permissions: Set[PermissionLevel] = field(default_factory=set) dependencies: Dict[str, str] = field(default_factory=dict) min_api_version: str = "3.0.0" tags: List[str] = field(default_factory=list) homepage: str = "" icon: str = "" @classmethod def from_json(cls, path: Path) -> PluginManifest: """Load manifest from JSON file.""" import json with open(path, 'r', encoding='utf-8') as f: data = json.load(f) # Parse permissions permission_map = { 'file_read': PermissionLevel.FILE_READ, 'file_write': PermissionLevel.FILE_WRITE, 'network': PermissionLevel.NETWORK, 'ui': PermissionLevel.UI, 'memory': PermissionLevel.MEMORY, 'process': PermissionLevel.PROCESS, 'system': PermissionLevel.SYSTEM, } permissions = set() for perm_str in data.get('permissions', []): if perm_str in permission_map: permissions.add(permission_map[perm_str]) return cls( name=data['name'], version=data['version'], author=data.get('author', 'Unknown'), description=data.get('description', ''), entry_point=data.get('entry_point', 'main.py'), permissions=permissions, dependencies=data.get('dependencies', {}), min_api_version=data.get('min_api_version', '3.0.0'), tags=data.get('tags', []), homepage=data.get('homepage', ''), icon=data.get('icon', ''), ) def to_json(self, path: Path) -> None: """Save manifest to JSON file.""" import json # Convert permissions back to strings permission_reverse_map = { PermissionLevel.FILE_READ: 'file_read', PermissionLevel.FILE_WRITE: 'file_write', PermissionLevel.NETWORK: 'network', PermissionLevel.UI: 'ui', PermissionLevel.MEMORY: 'memory', PermissionLevel.PROCESS: 'process', PermissionLevel.SYSTEM: 'system', } data = { 'name': self.name, 'version': self.version, 'author': self.author, 'description': self.description, 'entry_point': self.entry_point, 'permissions': [permission_reverse_map[p] for p in self.permissions], 'dependencies': self.dependencies, 'min_api_version': self.min_api_version, 'tags': self.tags, 'homepage': self.homepage, 'icon': self.icon, } with open(path, 'w', encoding='utf-8') as f: json.dump(data, f, indent=2) # ============================================================================= # PLUGIN CONTEXT # ============================================================================= @dataclass class PluginContext: """Context passed to plugins during initialization. This provides plugins with access to system resources while maintaining sandbox boundaries. Attributes: plugin_id: Unique plugin identifier manifest: Plugin manifest data_dir: Directory for plugin data storage config: Plugin configuration dictionary logger: Logger instance for this plugin event_bus: Event bus for publishing/subscribing to events state_store: State store for accessing global state widget_api: API for creating UI widgets nexus_api: API for Entropia Nexus data permissions: Set of granted permissions """ plugin_id: str manifest: PluginManifest data_dir: Path config: Dict[str, Any] logger: logging.Logger event_bus: Optional[Any] = None state_store: Optional[Any] = None widget_api: Optional[Any] = None nexus_api: Optional[Any] = None permissions: Set[PermissionLevel] = field(default_factory=set) def has_permission(self, permission: PermissionLevel) -> bool: """Check if plugin has a specific permission.""" return permission in self.permissions def require_permission(self, permission: PermissionLevel) -> None: """Require a permission or raise an error.""" if not self.has_permission(permission): raise PluginPermissionError( f"Plugin '{self.manifest.name}' requires permission: {permission.name}" ) # ============================================================================= # PLUGIN INSTANCE # ============================================================================= @dataclass class PluginInstance: """Represents a loaded plugin instance. Tracks the lifecycle state and metadata of a plugin. """ plugin_id: str manifest: PluginManifest state: PluginState = PluginState.DISCOVERED instance: Optional[PluginAPI] = None load_time: Optional[datetime] = None activate_time: Optional[datetime] = None error_message: Optional[str] = None error_traceback: Optional[str] = None def is_active(self) -> bool: """Check if plugin is currently active.""" return self.state == PluginState.ACTIVE def has_error(self) -> bool: """Check if plugin is in error state.""" return self.state == PluginState.ERROR def to_dict(self) -> Dict[str, Any]: """Convert to dictionary for serialization.""" return { 'plugin_id': self.plugin_id, 'name': self.manifest.name, 'version': self.manifest.version, 'state': self.state.name, 'is_active': self.is_active(), 'has_error': self.has_error(), 'load_time': self.load_time.isoformat() if self.load_time else None, 'activate_time': self.activate_time.isoformat() if self.activate_time else None, 'error_message': self.error_message, } # ============================================================================= # PLUGIN ERRORS # ============================================================================= class PluginError(Exception): """Base exception for plugin-related errors.""" pass class PluginLoadError(PluginError): """Error loading a plugin (invalid code, missing files, etc).""" pass class PluginInitError(PluginError): """Error initializing a plugin.""" pass class PluginPermissionError(PluginError): """Plugin tried to use a permission it doesn't have.""" pass class PluginDependencyError(PluginError): """Error with plugin dependencies.""" pass class PluginVersionError(PluginError): """Error with plugin version compatibility.""" pass class PluginAPIError(PluginError): """Error with plugin API usage.""" pass # ============================================================================= # PLUGIN API BASE CLASS # ============================================================================= class PluginAPI(ABC): """Base class for all plugins. Plugins must inherit from this class and implement the lifecycle methods. Example: class MyPlugin(PluginAPI): manifest = PluginManifest( name="My Plugin", version="1.0.0", author="Your Name" ) def on_init(self, ctx: PluginContext): self.ctx = ctx self.config = ctx.config def on_activate(self): # Start doing work pass def on_deactivate(self): # Stop doing work pass def on_shutdown(self): # Cleanup resources pass def create_widget(self) -> Optional[QWidget]: # Return UI widget for dashboard return None """ # Must be defined by subclass manifest: PluginManifest def __init__(self): self.ctx: Optional[PluginContext] = None self._initialized = False self._active = False def _set_context(self, ctx: PluginContext) -> None: """Set the plugin context (called by PluginManager).""" self.ctx = ctx # ========== Lifecycle Methods ========== @abstractmethod def on_init(self, ctx: PluginContext) -> None: """Called when plugin is initialized. Use this to set up initial state, load config, etc. Don't start any background work here - use on_activate for that. Args: ctx: Plugin context with resources and configuration """ pass def on_activate(self) -> None: """Called when plugin is activated. Start background tasks, register event handlers, etc. """ pass def on_deactivate(self) -> None: """Called when plugin is deactivated. Stop background tasks, unregister event handlers. """ pass def on_shutdown(self) -> None: """Called when plugin is being unloaded. Clean up all resources, save state, etc. """ pass # ========== UI Methods ========== def create_widget(self) -> Optional[Any]: """Create a widget for the dashboard. Returns: QWidget or None if plugin has no UI """ return None def get_settings_widget(self) -> Optional[Any]: """Create a settings widget. Returns: QWidget or None if plugin has no settings """ return None # ========== Utility Methods ========== def log(self, level: str, message: str) -> None: """Log a message through the plugin's logger.""" if self.ctx and self.ctx.logger: getattr(self.ctx.logger, level.lower(), self.ctx.logger.info)(message) def emit_event(self, event_type: str, data: Dict[str, Any]) -> None: """Emit an event to the event bus.""" if self.ctx and self.ctx.event_bus: self.ctx.event_bus.emit(event_type, data, source=self.ctx.plugin_id) def save_config(self) -> bool: """Save plugin configuration to disk.""" if not self.ctx: return False config_path = self.ctx.data_dir / "config.json" try: import json with open(config_path, 'w', encoding='utf-8') as f: json.dump(self.ctx.config, f, indent=2) return True except Exception as e: self.log('error', f"Failed to save config: {e}") return False # ============================================================================= # EXPORTS # ============================================================================= __all__ = [ # Permissions 'PermissionLevel', # State 'PluginState', # Manifest 'PluginManifest', # Context 'PluginContext', # Instance 'PluginInstance', # Errors 'PluginError', 'PluginLoadError', 'PluginInitError', 'PluginPermissionError', 'PluginDependencyError', 'PluginVersionError', 'PluginAPIError', # Base class 'PluginAPI', ]