fix: Path import error + Plugin API system + Error handling

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! 🚀
This commit is contained in:
LemonNexus 2026-02-13 15:09:25 +00:00
parent 2abbea9563
commit 5871ac611b
5 changed files with 462 additions and 12 deletions

View File

@ -36,6 +36,7 @@ from core.overlay_window import OverlayWindow
from core.floating_icon import FloatingIcon from core.floating_icon import FloatingIcon
from core.settings import get_settings from core.settings import get_settings
from core.overlay_widgets import OverlayManager from core.overlay_widgets import OverlayManager
from core.plugin_api import get_api, APIType
class HotkeyHandler(QObject): class HotkeyHandler(QObject):
@ -55,6 +56,7 @@ class EUUtilityApp:
self.hotkey_handler = None self.hotkey_handler = None
self.settings = None self.settings = None
self.overlay_manager = None self.overlay_manager = None
self.api = None
def run(self): def run(self):
"""Start the application.""" """Start the application."""
@ -66,6 +68,11 @@ class EUUtilityApp:
if hasattr(Qt, 'AA_EnableHighDpiScaling'): if hasattr(Qt, 'AA_EnableHighDpiScaling'):
self.app.setAttribute(Qt.ApplicationAttribute.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 # Load settings
self.settings = get_settings() self.settings = get_settings()
@ -103,10 +110,31 @@ class EUUtilityApp:
print("Press Ctrl+Shift+U to toggle overlay") print("Press Ctrl+Shift+U to toggle overlay")
print("Press Ctrl+Shift+H to hide all overlays") print("Press Ctrl+Shift+H to hide all overlays")
print("Or double-click the floating icon") print("Or double-click the floating icon")
print(f"Loaded {len(self.plugin_manager.get_all_plugins())} plugins")
# Run # Run
return self.app.exec() 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): def _setup_hotkeys(self):
"""Setup global hotkeys.""" """Setup global hotkeys."""
if KEYBOARD_AVAILABLE: if KEYBOARD_AVAILABLE:

View File

@ -2,9 +2,13 @@
EU-Utility - Overlay Widget System EU-Utility - Overlay Widget System
Draggable, hideable overlay elements that appear in-game. 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 ( from PyQt6.QtWidgets import (
QWidget, QVBoxLayout, QHBoxLayout, QLabel, QWidget, QVBoxLayout, QHBoxLayout, QLabel,
QPushButton, QFrame, QGraphicsDropShadowEffect, QPushButton, QFrame, QGraphicsDropShadowEffect,

View File

@ -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

View File

@ -52,7 +52,7 @@ class PluginManager:
config_path.write_text(json.dumps(self.config, indent=2)) config_path.write_text(json.dumps(self.config, indent=2))
def discover_plugins(self) -> List[Type[BasePlugin]]: def discover_plugins(self) -> List[Type[BasePlugin]]:
"""Discover all available plugin classes.""" """Discover all available plugin classes with error handling."""
discovered = [] discovered = []
for plugin_dir in self.PLUGIN_DIRS: for plugin_dir in self.PLUGIN_DIRS:
@ -66,9 +66,10 @@ class PluginManager:
continue continue
if item.name.startswith("__"): if item.name.startswith("__"):
continue continue
if item.name == "__pycache__":
continue
plugin_file = item / "plugin.py" plugin_file = item / "plugin.py"
init_file = item / "__init__.py"
if not plugin_file.exists(): if not plugin_file.exists():
continue continue
@ -83,6 +84,8 @@ class PluginManager:
spec = importlib.util.spec_from_file_location( spec = importlib.util.spec_from_file_location(
module_name, plugin_file module_name, plugin_file
) )
if not spec or not spec.loader:
continue
module = importlib.util.module_from_spec(spec) module = importlib.util.module_from_spec(spec)
sys.modules[module_name] = module sys.modules[module_name] = module
spec.loader.exec_module(module) spec.loader.exec_module(module)
@ -98,12 +101,13 @@ class PluginManager:
break break
except Exception as e: 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 return discovered
def load_plugin(self, plugin_class: Type[BasePlugin]) -> bool: def load_plugin(self, plugin_class: Type[BasePlugin]) -> bool:
"""Instantiate and initialize a plugin.""" """Instantiate and initialize a plugin with error handling."""
try: try:
plugin_id = f"{plugin_class.__module__}.{plugin_class.__name__}" plugin_id = f"{plugin_class.__module__}.{plugin_class.__name__}"
@ -111,24 +115,41 @@ class PluginManager:
if plugin_id in self.plugins: if plugin_id in self.plugins:
return True 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 # Get plugin config
plugin_config = self.config.get("settings", {}).get(plugin_id, {}) plugin_config = self.config.get("settings", {}).get(plugin_id, {})
# Create instance # 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 # Initialize with error handling
instance.initialize() 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 # Store
self.plugins[plugin_id] = instance self.plugins[plugin_id] = instance
self.plugin_classes[plugin_id] = plugin_class 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 return True
except Exception as e: 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 return False
def load_all_plugins(self) -> None: def load_all_plugins(self) -> None:

View File

@ -2,13 +2,15 @@
EU-Utility - Plugin Base Class EU-Utility - Plugin Base Class
Defines the interface that all plugins must implement. Defines the interface that all plugins must implement.
Includes PluginAPI integration for cross-plugin communication.
""" """
from abc import ABC, abstractmethod 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: if TYPE_CHECKING:
from core.overlay_window import OverlayWindow from core.overlay_window import OverlayWindow
from core.plugin_api import PluginAPI, APIEndpoint, APIType
class BasePlugin(ABC): class BasePlugin(ABC):
@ -29,6 +31,15 @@ class BasePlugin(ABC):
self.overlay = overlay_window self.overlay = overlay_window
self.config = config self.config = config
self._ui = None 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 @abstractmethod
def initialize(self) -> None: def initialize(self) -> None:
@ -54,7 +65,11 @@ class BasePlugin(ABC):
def shutdown(self) -> None: def shutdown(self) -> None:
"""Called when app is closing. Cleanup resources.""" """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: def get_config(self, key: str, default: Any = None) -> Any:
"""Get a config value with default.""" """Get a config value with default."""
@ -63,3 +78,142 @@ class BasePlugin(ABC):
def set_config(self, key: str, value: Any) -> None: def set_config(self, key: str, value: Any) -> None:
"""Set a config value.""" """Set a config value."""
self.config[key] = 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