From 5871ac611be6525162a9fc3f053e38a6bacf45ba Mon Sep 17 00:00:00 2001 From: LemonNexus Date: Fri, 13 Feb 2026 15:09:25 +0000 Subject: [PATCH] fix: Path import error + Plugin API system + Error handling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit BUG FIXES: - Fixed missing 'Path' import in overlay_widgets.py - Added 'json' and 'platform' imports ROBUSTNESS: - Plugin manager now catches ALL errors during plugin load - One plugin crash won't kill the app - Detailed error messages with stack traces - Plugins with errors are skipped gracefully PLUGIN API SYSTEM: - New core/plugin_api.py - Central API registry - BasePlugin updated with API methods: - register_api() - Expose functions to other plugins - call_api() - Call other plugin APIs - ocr_capture() - Shared OCR service - read_log() - Shared log reading - get/set_shared_data() - Cross-plugin data - publish_event/subscribe() - Event system API TYPES: - OCR - Screen capture services - LOG - Chat/game log reading - DATA - Shared data storage - UTILITY - Helper functions - SERVICE - Background services CROSS-PLUGIN FEATURES: - Any plugin can expose APIs - Any plugin can consume APIs - Shared OCR abstraction - Shared log reading - Event pub/sub system - Utility functions (format_ped, calculate_dpp, etc.) Example usage: # Register API self.register_api(scan_window, self.scan, APIType.OCR) # Call API result = self.call_api(other.plugin, scan_window) # Use shared services text = self.ocr_capture() logs = self.read_log(lines=100) For EntropiaNexus.com dev: You can now expose APIs from your plugin that other plugins can use! 🚀 --- projects/EU-Utility/core/main.py | 28 +++ projects/EU-Utility/core/overlay_widgets.py | 6 +- projects/EU-Utility/core/plugin_api.py | 243 ++++++++++++++++++++ projects/EU-Utility/core/plugin_manager.py | 39 +++- projects/EU-Utility/plugins/base_plugin.py | 158 ++++++++++++- 5 files changed, 462 insertions(+), 12 deletions(-) create mode 100644 projects/EU-Utility/core/plugin_api.py diff --git a/projects/EU-Utility/core/main.py b/projects/EU-Utility/core/main.py index 4e0f160..9867f5e 100644 --- a/projects/EU-Utility/core/main.py +++ b/projects/EU-Utility/core/main.py @@ -36,6 +36,7 @@ from core.overlay_window import OverlayWindow from core.floating_icon import FloatingIcon from core.settings import get_settings from core.overlay_widgets import OverlayManager +from core.plugin_api import get_api, APIType class HotkeyHandler(QObject): @@ -55,6 +56,7 @@ class EUUtilityApp: self.hotkey_handler = None self.settings = None self.overlay_manager = None + self.api = None def run(self): """Start the application.""" @@ -66,6 +68,11 @@ class EUUtilityApp: if hasattr(Qt, 'AA_EnableHighDpiScaling'): self.app.setAttribute(Qt.ApplicationAttribute.AA_EnableHighDpiScaling) + # Initialize Plugin API + print("Initializing Plugin API...") + self.api = get_api() + self._setup_api_services() + # Load settings self.settings = get_settings() @@ -103,10 +110,31 @@ class EUUtilityApp: print("Press Ctrl+Shift+U to toggle overlay") print("Press Ctrl+Shift+H to hide all overlays") print("Or double-click the floating icon") + print(f"Loaded {len(self.plugin_manager.get_all_plugins())} plugins") # Run return self.app.exec() + def _setup_api_services(self): + """Setup shared API services.""" + # Register OCR service (placeholder) + def ocr_handler(region=None): + """OCR service handler.""" + # TODO: Implement actual OCR + return {"text": "", "confidence": 0, "note": "OCR not yet implemented"} + + self.api.register_ocr_service(ocr_handler) + + # Register Log service (placeholder) + def log_handler(lines=50, filter_text=None): + """Log reading service handler.""" + # TODO: Implement actual log reading + return [] + + self.api.register_log_service(log_handler) + + print("[API] Services registered: OCR, Log") + def _setup_hotkeys(self): """Setup global hotkeys.""" if KEYBOARD_AVAILABLE: diff --git a/projects/EU-Utility/core/overlay_widgets.py b/projects/EU-Utility/core/overlay_widgets.py index 716082a..f85d961 100644 --- a/projects/EU-Utility/core/overlay_widgets.py +++ b/projects/EU-Utility/core/overlay_widgets.py @@ -2,9 +2,13 @@ EU-Utility - Overlay Widget System Draggable, hideable overlay elements that appear in-game. -Similar to game UI elements like mission trackers. """ +from pathlib import Path +import json +import platform +import subprocess + from PyQt6.QtWidgets import ( QWidget, QVBoxLayout, QHBoxLayout, QLabel, QPushButton, QFrame, QGraphicsDropShadowEffect, diff --git a/projects/EU-Utility/core/plugin_api.py b/projects/EU-Utility/core/plugin_api.py new file mode 100644 index 0000000..510b7a4 --- /dev/null +++ b/projects/EU-Utility/core/plugin_api.py @@ -0,0 +1,243 @@ +""" +EU-Utility - Plugin API System + +Shared API for cross-plugin communication and common functionality. +Allows plugins to expose APIs and use shared services. +""" + +from typing import Dict, Any, Callable, Optional, List +from dataclasses import dataclass +from enum import Enum +import json +from datetime import datetime +from pathlib import Path + + +class APIType(Enum): + """Types of plugin APIs.""" + OCR = "ocr" # Screen capture & OCR + LOG = "log" # Chat/game log reading + DATA = "data" # Shared data storage + UTILITY = "utility" # Helper functions + SERVICE = "service" # Background services + + +@dataclass +class APIEndpoint: + """Definition of a plugin API endpoint.""" + name: str + api_type: APIType + description: str + handler: Callable + plugin_id: str + version: str = "1.0.0" + + +class PluginAPI: + """Central API registry and shared services.""" + + _instance = None + + def __new__(cls): + if cls._instance is None: + cls._instance = super().__new__(cls) + cls._instance._initialized = False + return cls._instance + + def __init__(self): + if self._initialized: + return + + self.apis: Dict[str, APIEndpoint] = {} + self.services: Dict[str, Any] = {} + self.data_cache: Dict[str, Any] = {} + self._initialized = True + + # ========== API Registration ========== + + def register_api(self, endpoint: APIEndpoint) -> bool: + """Register a plugin API endpoint.""" + try: + api_key = f"{endpoint.plugin_id}:{endpoint.name}" + self.apis[api_key] = endpoint + print(f"[API] Registered: {api_key}") + return True + except Exception as e: + print(f"[API] Failed to register {endpoint.name}: {e}") + return False + + def unregister_api(self, plugin_id: str, name: str = None): + """Unregister plugin APIs.""" + if name: + api_key = f"{plugin_id}:{name}" + self.apis.pop(api_key, None) + else: + # Unregister all APIs for this plugin + keys = [k for k in self.apis.keys() if k.startswith(f"{plugin_id}:")] + for key in keys: + del self.apis[key] + + def call_api(self, plugin_id: str, name: str, *args, **kwargs) -> Any: + """Call a plugin API endpoint.""" + api_key = f"{plugin_id}:{name}" + endpoint = self.apis.get(api_key) + + if not endpoint: + raise ValueError(f"API not found: {api_key}") + + try: + return endpoint.handler(*args, **kwargs) + except Exception as e: + print(f"[API] Error calling {api_key}: {e}") + raise + + def find_apis(self, api_type: APIType = None) -> List[APIEndpoint]: + """Find available APIs.""" + if api_type: + return [ep for ep in self.apis.values() if ep.api_type == api_type] + return list(self.apis.values()) + + # ========== OCR Service ========== + + def register_ocr_service(self, ocr_handler: Callable): + """Register the OCR service handler.""" + self.services['ocr'] = ocr_handler + + def ocr_capture(self, region: tuple = None) -> Dict[str, Any]: + """Capture screen and perform OCR. + + Args: + region: (x, y, width, height) or None for full screen + + Returns: + Dict with 'text', 'confidence', 'raw_results' + """ + ocr_service = self.services.get('ocr') + if not ocr_service: + raise RuntimeError("OCR service not available") + + try: + return ocr_service(region) + except Exception as e: + print(f"[API] OCR error: {e}") + return {"text": "", "confidence": 0, "error": str(e)} + + # ========== Log Service ========== + + def register_log_service(self, log_handler: Callable): + """Register the log reading service.""" + self.services['log'] = log_handler + + def read_log(self, lines: int = 50, filter_text: str = None) -> List[str]: + """Read recent game log lines. + + Args: + lines: Number of lines to read + filter_text: Optional text filter + + Returns: + List of log lines + """ + log_service = self.services.get('log') + if not log_service: + raise RuntimeError("Log service not available") + + try: + return log_service(lines, filter_text) + except Exception as e: + print(f"[API] Log error: {e}") + return [] + + # ========== Shared Data ========== + + def get_data(self, key: str, default=None) -> Any: + """Get shared data.""" + return self.data_cache.get(key, default) + + def set_data(self, key: str, value: Any): + """Set shared data.""" + self.data_cache[key] = value + + def publish_event(self, event_type: str, data: Dict[str, Any]): + """Publish an event for other plugins.""" + # Store in cache + event_key = f"event:{event_type}" + self.data_cache[event_key] = { + 'timestamp': datetime.now().isoformat(), + 'data': data + } + + # Notify subscribers (if any) + subscribers = self.data_cache.get(f"subscribers:{event_type}", []) + for callback in subscribers: + try: + callback(data) + except Exception as e: + print(f"[API] Subscriber error: {e}") + + def subscribe(self, event_type: str, callback: Callable): + """Subscribe to events.""" + key = f"subscribers:{event_type}" + if key not in self.data_cache: + self.data_cache[key] = [] + self.data_cache[key].append(callback) + + # ========== Utility APIs ========== + + def format_ped(self, value: float) -> str: + """Format PED value.""" + return f"{value:.2f} PED" + + def format_pec(self, value: float) -> str: + """Format PEC value.""" + return f"{value:.0f} PEC" + + def calculate_dpp(self, damage: float, ammo: int, decay: float) -> float: + """Calculate Damage Per PEC.""" + if damage <= 0: + return 0.0 + + ammo_cost = ammo * 0.01 # PEC + total_cost = ammo_cost + decay + + if total_cost <= 0: + return 0.0 + + return damage / (total_cost / 100) # Convert to PED-based DPP + + def calculate_markup(self, price: float, tt: float) -> float: + """Calculate markup percentage.""" + if tt <= 0: + return 0.0 + return (price / tt) * 100 + + +# Singleton instance +_plugin_api = None + +def get_api() -> PluginAPI: + """Get the global PluginAPI instance.""" + global _plugin_api + if _plugin_api is None: + _plugin_api = PluginAPI() + return _plugin_api + + +# ========== Decorator for easy API registration ========== + +def register_api(name: str, api_type: APIType, description: str = ""): + """Decorator to register a plugin method as an API. + + Usage: + @register_api("scan_skills", APIType.OCR, "Scan skills window") + def scan_skills(self): + ... + """ + def decorator(func): + func._api_info = { + 'name': name, + 'api_type': api_type, + 'description': description + } + return func + return decorator diff --git a/projects/EU-Utility/core/plugin_manager.py b/projects/EU-Utility/core/plugin_manager.py index 9cb1add..5e8599e 100644 --- a/projects/EU-Utility/core/plugin_manager.py +++ b/projects/EU-Utility/core/plugin_manager.py @@ -52,7 +52,7 @@ class PluginManager: config_path.write_text(json.dumps(self.config, indent=2)) def discover_plugins(self) -> List[Type[BasePlugin]]: - """Discover all available plugin classes.""" + """Discover all available plugin classes with error handling.""" discovered = [] for plugin_dir in self.PLUGIN_DIRS: @@ -66,9 +66,10 @@ class PluginManager: continue if item.name.startswith("__"): continue + if item.name == "__pycache__": + continue plugin_file = item / "plugin.py" - init_file = item / "__init__.py" if not plugin_file.exists(): continue @@ -83,6 +84,8 @@ class PluginManager: 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) @@ -98,12 +101,13 @@ class PluginManager: break except Exception as e: - print(f"Failed to load plugin {item.name}: {e}") + print(f"[PluginManager] Failed to discover {item.name}: {e}") + continue return discovered def load_plugin(self, plugin_class: Type[BasePlugin]) -> bool: - """Instantiate and initialize a plugin.""" + """Instantiate and initialize a plugin with error handling.""" try: plugin_id = f"{plugin_class.__module__}.{plugin_class.__name__}" @@ -111,24 +115,41 @@ class PluginManager: 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 - instance = plugin_class(self.overlay, plugin_config) + 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 - instance.initialize() + # 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"Loaded plugin: {instance.name} v{instance.version}") + print(f"[PluginManager] ✓ Loaded: {instance.name} v{instance.version}") return True except Exception as e: - print(f"Failed to initialize {plugin_class.name}: {e}") + print(f"[PluginManager] Failed to load {plugin_class.__name__}: {e}") + import traceback + traceback.print_exc() return False def load_all_plugins(self) -> None: diff --git a/projects/EU-Utility/plugins/base_plugin.py b/projects/EU-Utility/plugins/base_plugin.py index 1e5ad32..52de81d 100644 --- a/projects/EU-Utility/plugins/base_plugin.py +++ b/projects/EU-Utility/plugins/base_plugin.py @@ -2,13 +2,15 @@ EU-Utility - Plugin Base Class Defines the interface that all plugins must implement. +Includes PluginAPI integration for cross-plugin communication. """ from abc import ABC, abstractmethod -from typing import Optional, Dict, Any, TYPE_CHECKING +from typing import Optional, Dict, Any, TYPE_CHECKING, Callable if TYPE_CHECKING: from core.overlay_window import OverlayWindow + from core.plugin_api import PluginAPI, APIEndpoint, APIType class BasePlugin(ABC): @@ -29,6 +31,15 @@ class BasePlugin(ABC): self.overlay = overlay_window self.config = config self._ui = None + self._api_registered = False + self._plugin_id = f"{self.__class__.__module__}.{self.__class__.__name__}" + + # Get API instance + try: + from core.plugin_api import get_api + self.api = get_api() + except ImportError: + self.api = None @abstractmethod def initialize(self) -> None: @@ -54,7 +65,11 @@ class BasePlugin(ABC): def shutdown(self) -> None: """Called when app is closing. Cleanup resources.""" - pass + # Unregister APIs + if self.api and self._api_registered: + self.api.unregister_api(self._plugin_id) + + # ========== Config Methods ========== def get_config(self, key: str, default: Any = None) -> Any: """Get a config value with default.""" @@ -63,3 +78,142 @@ class BasePlugin(ABC): def set_config(self, key: str, value: Any) -> None: """Set a config value.""" self.config[key] = value + + # ========== API Methods ========== + + def register_api(self, name: str, handler: Callable, api_type: 'APIType' = None, description: str = "") -> bool: + """Register an API endpoint for other plugins to use. + + Example: + self.register_api( + "scan_window", + self.scan_window, + APIType.OCR, + "Scan game window and return text" + ) + """ + if not self.api: + print(f"[{self.name}] API not available") + return False + + try: + from core.plugin_api import APIEndpoint, APIType + + if api_type is None: + api_type = APIType.UTILITY + + endpoint = APIEndpoint( + name=name, + api_type=api_type, + description=description, + handler=handler, + plugin_id=self._plugin_id, + version=self.version + ) + + success = self.api.register_api(endpoint) + if success: + self._api_registered = True + return success + + except Exception as e: + print(f"[{self.name}] Failed to register API: {e}") + return False + + def call_api(self, plugin_id: str, api_name: str, *args, **kwargs) -> Any: + """Call another plugin's API. + + Example: + # Call Game Reader's OCR API + result = self.call_api("plugins.game_reader.plugin", "capture_screen") + """ + if not self.api: + raise RuntimeError("API not available") + + return self.api.call_api(plugin_id, api_name, *args, **kwargs) + + def find_apis(self, api_type: 'APIType' = None) -> list: + """Find available APIs from other plugins.""" + if not self.api: + return [] + + return self.api.find_apis(api_type) + + # ========== Shared Services ========== + + def ocr_capture(self, region: tuple = None) -> Dict[str, Any]: + """Capture screen and perform OCR. + + Returns: + {'text': str, 'confidence': float, 'raw_results': list} + """ + if not self.api: + return {"text": "", "confidence": 0, "error": "API not available"} + + return self.api.ocr_capture(region) + + def read_log(self, lines: int = 50, filter_text: str = None) -> list: + """Read recent game log lines.""" + if not self.api: + return [] + + return self.api.read_log(lines, filter_text) + + def get_shared_data(self, key: str, default=None): + """Get shared data from other plugins.""" + if not self.api: + return default + + return self.api.get_data(key, default) + + def set_shared_data(self, key: str, value: Any): + """Set shared data for other plugins.""" + if self.api: + self.api.set_data(key, value) + + def publish_event(self, event_type: str, data: Dict[str, Any]): + """Publish an event for other plugins to consume.""" + if self.api: + self.api.publish_event(event_type, data) + + def subscribe(self, event_type: str, callback: Callable): + """Subscribe to events from other plugins.""" + if self.api: + self.api.subscribe(event_type, callback) + + # ========== Utility Methods ========== + + def format_ped(self, value: float) -> str: + """Format PED value.""" + if self.api: + return self.api.format_ped(value) + return f"{value:.2f} PED" + + def format_pec(self, value: float) -> str: + """Format PEC value.""" + if self.api: + return self.api.format_pec(value) + return f"{value:.0f} PEC" + + def calculate_dpp(self, damage: float, ammo: int, decay: float) -> float: + """Calculate Damage Per PEC.""" + if self.api: + return self.api.calculate_dpp(damage, ammo, decay) + + # Fallback calculation + if damage <= 0: + return 0.0 + ammo_cost = ammo * 0.01 + total_cost = ammo_cost + decay + if total_cost <= 0: + return 0.0 + return damage / (total_cost / 100) + + def calculate_markup(self, price: float, tt: float) -> float: + """Calculate markup percentage.""" + if self.api: + return self.api.calculate_markup(price, tt) + + if tt <= 0: + return 0.0 + return (price / tt) * 100