""" EU-Utility - Plugin API ================ The PluginAPI provides plugins with access to core services and functionality. This is the primary interface for plugin developers. Quick Start: ----------- ```python from core.plugin_api import get_api class MyPlugin(BasePlugin): def initialize(self): self.api = get_api() # Access services log_lines = self.api.read_log_lines(100) window_info = self.api.get_eu_window() # Show notifications self.api.show_notification("Hello", "Plugin started!") ``` Services Available: ------------------ - Log Reader - Read game chat.log - Window Manager - Get EU window info, focus control - OCR Service - Screen text recognition - Screenshot - Capture screen regions - Nexus API - Item database queries - HTTP Client - Web requests with caching - Audio - Play sounds - Notifications - Toast notifications - Clipboard - Copy/paste operations - Event Bus - Pub/sub events - Data Store - Key-value storage - Tasks - Background task execution For full documentation, see: docs/API_COOKBOOK.md """ import json import time from pathlib import Path from typing import Optional, Dict, List, Callable, Any, Tuple, Union from functools import wraps from datetime import datetime from core.logger import get_logger logger = get_logger(__name__) class PluginAPIError(Exception): """Base exception for PluginAPI errors.""" pass class ServiceNotAvailableError(PluginAPIError): """Raised when a requested service is not available.""" pass class PluginAPI: """ PluginAPI - Core API for EU-Utility plugins. Provides access to all core services in a unified, documented interface. Plugins should obtain the API via get_api() and store it for reuse. Thread Safety: -------------- Most methods are thread-safe. UI-related methods (notifications, etc.) automatically marshal to the main thread via Qt signals. Example: -------- ```python class MyPlugin(BasePlugin): def initialize(self): self.api = get_api() def on_event(self, event): # Safe to call from any thread self.api.show_notification("Event", str(event)) ``` """ # Service registry _services: Dict[str, Any] = {} _instance: Optional['PluginAPI'] = None def __new__(cls): if cls._instance is None: cls._instance = super().__new__(cls) return cls._instance # ========================================================================= # SERVICE REGISTRATION (Internal Use) # ========================================================================= def register_log_service(self, read_lines_func: Callable) -> None: """Register the log reader service.""" self._services['log_reader'] = read_lines_func logger.debug("[PluginAPI] Log service registered") def register_window_service(self, window_manager) -> None: """Register the window manager service.""" self._services['window_manager'] = window_manager logger.debug("[PluginAPI] Window service registered") def register_ocr_service(self, recognize_func: Callable) -> None: """Register the OCR service.""" self._services['ocr'] = recognize_func logger.debug("[PluginAPI] OCR service registered") def register_screenshot_service(self, screenshot_service) -> None: """Register the screenshot service.""" self._services['screenshot'] = screenshot_service logger.debug("[PluginAPI] Screenshot service registered") def register_nexus_service(self, nexus_api) -> None: """Register the Nexus API service.""" self._services['nexus'] = nexus_api logger.debug("[PluginAPI] Nexus service registered") def register_http_service(self, http_client) -> None: """Register the HTTP client service.""" self._services['http'] = http_client logger.debug("[PluginAPI] HTTP service registered") def register_audio_service(self, audio_manager) -> None: """Register the audio service.""" self._services['audio'] = audio_manager logger.debug("[PluginAPI] Audio service registered") def register_notification_service(self, notification_manager) -> None: """Register the notification service.""" self._services['notification'] = notification_manager logger.debug("[PluginAPI] Notification service registered") def register_clipboard_service(self, clipboard_manager) -> None: """Register the clipboard service.""" self._services['clipboard'] = clipboard_manager logger.debug("[PluginAPI] Clipboard service registered") def register_event_bus(self, event_bus) -> None: """Register the event bus service.""" self._services['event_bus'] = event_bus logger.debug("[PluginAPI] Event bus registered") def register_data_service(self, data_store) -> None: """Register the data store service.""" self._services['data_store'] = data_store logger.debug("[PluginAPI] Data service registered") def register_task_service(self, task_manager) -> None: """Register the task manager service.""" self._services['tasks'] = task_manager logger.debug("[PluginAPI] Task service registered") # ========================================================================= # LOG READER API # ========================================================================= def read_log_lines(self, count: int = 100) -> List[str]: """ Read recent lines from the game chat.log. Args: count: Number of lines to read (default: 100) Returns: List of log line strings Example: >>> lines = api.read_log_lines(50) >>> for line in lines: ... if 'Loot:' in line: ... print(line) """ service = self._services.get('log_reader') if service: return service(count) return [] def read_log_since(self, timestamp: datetime) -> List[str]: """ Read log lines since a specific timestamp. Args: timestamp: Datetime to read from Returns: List of log lines after the timestamp """ lines = self.read_log_lines(1000) result = [] for line in lines: # Parse timestamp from line (format varies) try: # Attempt to extract timestamp if '[' in line and ']' in line: result.append(line) except: continue return result # ========================================================================= # WINDOW MANAGER API # ========================================================================= def get_eu_window(self) -> Optional[Dict[str, Any]]: """ Get information about the EU game window. Returns: Dict with window info or None if not found: { 'title': str, 'hwnd': int, 'x': int, 'y': int, 'width': int, 'height': int, 'is_focused': bool, 'is_visible': bool } Example: >>> window = api.get_eu_window() >>> if window: ... print(f"EU is at {window['x']}, {window['y']}") """ wm = self._services.get('window_manager') if wm and wm.is_available(): window = wm.find_eu_window() if window: return { 'title': window.title, 'hwnd': window.hwnd, 'x': window.x, 'y': window.y, 'width': window.width, 'height': window.height, 'is_focused': window.is_focused(), 'is_visible': window.is_visible() } return None def is_eu_focused(self) -> bool: """ Check if the EU window is currently focused. Returns: True if EU is the active window Example: >>> if api.is_eu_focused(): ... api.play_sound("alert.wav") """ window = self.get_eu_window() return window['is_focused'] if window else False def is_eu_visible(self) -> bool: """ Check if the EU window is visible. Returns: True if EU window is visible (not minimized) """ window = self.get_eu_window() return window['is_visible'] if window else False def bring_eu_to_front(self) -> bool: """ Bring the EU window to the foreground. Returns: True if successful Warning: Use sparingly - can be intrusive to user """ wm = self._services.get('window_manager') if wm and wm.is_available(): window = wm.find_eu_window() if window: return window.focus() return False # ========================================================================= # OCR API # ========================================================================= def recognize_text(self, region: Optional[Tuple[int, int, int, int]] = None, image_path: Optional[str] = None) -> str: """ Perform OCR on screen region or image. Args: region: (x, y, width, height) tuple for screen region image_path: Path to image file (alternative to region) Returns: Recognized text string Raises: ServiceNotAvailableError: If OCR service not initialized Example: >>> # Read text from screen region >>> text = api.recognize_text((100, 100, 200, 50)) >>> print(f"Found: {text}") """ service = self._services.get('ocr') if service: return service(region=region, image_path=image_path) raise ServiceNotAvailableError("OCR service not available") def ocr_available(self) -> bool: """Check if OCR service is available.""" return 'ocr' in self._services # ========================================================================= # SCREENSHOT API # ========================================================================= def capture_screen(self, region: Optional[Tuple[int, int, int, int]] = None, save_path: Optional[str] = None) -> Optional[Any]: """ Capture a screenshot. Args: region: (x, y, width, height) tuple for specific region save_path: Optional path to save image Returns: PIL Image object or None if failed Example: >>> img = api.capture_screen((0, 0, 1920, 1080), "screenshot.png") >>> if img: ... print(f"Captured {img.size}") """ service = self._services.get('screenshot') if service and service.is_available(): return service.capture(region=region, save_path=save_path) return None def screenshot_available(self) -> bool: """Check if screenshot service is available.""" service = self._services.get('screenshot') return service.is_available() if service else False # ========================================================================= # NEXUS API (Item Database) # ========================================================================= def search_items(self, query: str, limit: int = 10) -> List[Dict]: """ Search for items in the Entropia Nexus database. Args: query: Search query string limit: Maximum results (default: 10) Returns: List of item dictionaries Example: >>> items = api.search_items("omegaton", limit=5) >>> for item in items: ... print(f"{item['Name']}: {item['Value']} PED") """ service = self._services.get('nexus') if service: try: return service.search_items(query, limit=limit) except Exception as e: logger.error(f"[PluginAPI] Nexus search failed: {e}") return [] def get_item_details(self, item_id: int) -> Optional[Dict]: """ Get detailed information about an item. Args: item_id: Nexus item ID Returns: Item details dict or None if not found """ service = self._services.get('nexus') if service: try: return service.get_item(item_id) except Exception as e: logger.error(f"[PluginAPI] Get item failed: {e}") return None # ========================================================================= # HTTP CLIENT API # ========================================================================= def http_get(self, url: str, cache: bool = True, cache_duration: int = 3600) -> Dict[str, Any]: """ Perform HTTP GET request with optional caching. Args: url: URL to fetch cache: Whether to use cache (default: True) cache_duration: Cache TTL in seconds (default: 1 hour) Returns: Response dict: {'success': bool, 'data': Any, 'error': str} Example: >>> result = api.http_get("https://api.example.com/data") >>> if result['success']: ... data = result['data'] """ service = self._services.get('http') if service: try: return service.get(url, cache=cache, cache_duration=cache_duration) except Exception as e: return {'success': False, 'error': str(e)} return {'success': False, 'error': 'HTTP service not available'} def http_post(self, url: str, data: Dict, cache: bool = False) -> Dict[str, Any]: """ Perform HTTP POST request. Args: url: URL to post to data: POST data dictionary cache: Whether to cache response Returns: Response dict: {'success': bool, 'data': Any, 'error': str} """ service = self._services.get('http') if service: try: return service.post(url, data=data, cache=cache) except Exception as e: return {'success': False, 'error': str(e)} return {'success': False, 'error': 'HTTP service not available'} # ========================================================================= # AUDIO API # ========================================================================= def play_sound(self, sound_path: str, volume: float = 1.0) -> bool: """ Play an audio file. Args: sound_path: Path to audio file (.wav, .mp3, etc.) volume: Volume level 0.0 to 1.0 Returns: True if playback started Example: >>> api.play_sound("assets/sounds/alert.wav", volume=0.7) """ service = self._services.get('audio') if service: try: service.play(sound_path, volume=volume) return True except Exception as e: logger.error(f"[PluginAPI] Play sound failed: {e}") return False def beep(self) -> bool: """Play a simple beep sound.""" service = self._services.get('audio') if service: try: service.beep() return True except: pass return False # ========================================================================= # NOTIFICATION API # ========================================================================= def show_notification(self, title: str, message: str, duration: int = 5000, sound: bool = False) -> bool: """ Show a toast notification. Args: title: Notification title message: Notification body duration: Duration in milliseconds (default: 5000) sound: Play sound with notification Returns: True if notification shown Example: >>> api.show_notification("Loot Alert", ... "Found something valuable!", ... duration=3000, sound=True) """ service = self._services.get('notification') if service: try: service.show(title, message, duration=duration, sound=sound) return True except Exception as e: logger.error(f"[PluginAPI] Notification failed: {e}") return False # ========================================================================= # CLIPBOARD API # ========================================================================= def copy_to_clipboard(self, text: str) -> bool: """ Copy text to system clipboard. Args: text: Text to copy Returns: True if successful Example: >>> api.copy_to_clipboard("TT: 100 PED") """ service = self._services.get('clipboard') if service: try: service.copy(text) return True except Exception as e: logger.error(f"[PluginAPI] Clipboard copy failed: {e}") return False def paste_from_clipboard(self) -> str: """ Get text from system clipboard. Returns: Clipboard text or empty string """ service = self._services.get('clipboard') if service: try: return service.paste() except: pass return "" # ========================================================================= # EVENT BUS API (Pub/Sub) # ========================================================================= def subscribe(self, event_type: str, callback: Callable) -> str: """ Subscribe to an event type. Args: event_type: Event type string (e.g., "loot", "skill_gain") callback: Function to call when event occurs Returns: Subscription ID (use to unsubscribe) Example: >>> def on_loot(event): ... print(f"Got loot: {event.data}") >>> sub_id = api.subscribe("loot", on_loot) """ service = self._services.get('event_bus') if service: try: return service.subscribe(event_type, callback) except Exception as e: logger.error(f"[PluginAPI] Subscribe failed: {e}") return "" def unsubscribe(self, subscription_id: str) -> bool: """ Unsubscribe from events. Args: subscription_id: ID returned by subscribe() Returns: True if unsubscribed """ service = self._services.get('event_bus') if service: try: service.unsubscribe(subscription_id) return True except Exception as e: logger.error(f"[PluginAPI] Unsubscribe failed: {e}") return False def publish(self, event_type: str, data: Any) -> bool: """ Publish an event. Args: event_type: Event type string data: Event data (any type) Returns: True if published Example: >>> api.publish("my_plugin.event", {"key": "value"}) """ service = self._services.get('event_bus') if service: try: service.publish(event_type, data) return True except Exception as e: logger.error(f"[PluginAPI] Publish failed: {e}") return False # ========================================================================= # DATA STORE API (Key-Value Storage) # ========================================================================= def get_data(self, key: str, default: Any = None) -> Any: """ Get data from plugin data store. Args: key: Data key default: Default value if key not found Returns: Stored value or default Example: >>> count = api.get_data("kill_count", 0) >>> api.set_data("kill_count", count + 1) """ service = self._services.get('data_store') if service: try: return service.get(key, default) except: pass return default def set_data(self, key: str, value: Any) -> bool: """ Store data in plugin data store. Args: key: Data key value: Value to store (must be JSON serializable) Returns: True if stored """ service = self._services.get('data_store') if service: try: service.set(key, value) return True except Exception as e: logger.error(f"[PluginAPI] Set data failed: {e}") return False def delete_data(self, key: str) -> bool: """Delete data from store.""" service = self._services.get('data_store') if service: try: service.delete(key) return True except: pass return False # ========================================================================= # TASK API (Background Execution) # ========================================================================= def run_task(self, task_func: Callable, *args, callback: Optional[Callable] = None, error_handler: Optional[Callable] = None) -> str: """ Run a function in background thread. Args: task_func: Function to execute *args: Arguments for function callback: Called with result on success error_handler: Called with exception on error Returns: Task ID Example: >>> def heavy_work(data): ... return process(data) >>> >>> def on_done(result): ... print(f"Done: {result}") >>> >>> task_id = api.run_task(heavy_work, my_data, callback=on_done) """ service = self._services.get('tasks') if service: try: return service.submit(task_func, *args, callback=callback, error_handler=error_handler) except Exception as e: logger.error(f"[PluginAPI] Run task failed: {e}") return "" def cancel_task(self, task_id: str) -> bool: """Cancel a running task.""" service = self._services.get('tasks') if service: try: return service.cancel(task_id) except: pass return False # Global API instance _api_instance: Optional[PluginAPI] = None def get_api() -> PluginAPI: """ Get the global PluginAPI instance. This is the entry point for all plugin API access. Returns: PluginAPI singleton instance Example: >>> from core.plugin_api import get_api >>> api = get_api() >>> api.show_notification("Hello", "World!") """ global _api_instance if _api_instance is None: _api_instance = PluginAPI() return _api_instance # Convenience exports __all__ = [ 'PluginAPI', 'get_api', 'PluginAPIError', 'ServiceNotAvailableError' ]