From 8e38c6cb120879dcab505e2412bec9193319b977 Mon Sep 17 00:00:00 2001 From: devmatrix Date: Mon, 16 Feb 2026 22:34:51 +0000 Subject: [PATCH] Auto-sync: 2026-02-16 22:34 --- main.py | 297 +++++++++ plugins/builtin/dashboard_widget/main.py | 322 ++++++++++ plugins/builtin/dashboard_widget/plugin.json | 12 + premium/__init__.py | 173 ++++++ premium/core/event_bus.py | 599 +++++++++++++++++++ premium/eu_integration/__init__.py | 33 + premium/eu_integration/events.py | 131 ++++ premium/eu_integration/game_client.py | 344 +++++++++++ premium/eu_integration/log_parser.py | 323 ++++++++++ premium/eu_integration/window_tracker.py | 107 ++++ premium/plugins/api.py | 465 ++++++++++++++ premium/widgets/__init__.py | 378 ++++++++++++ premium/widgets/base.py | 20 + premium/widgets/chart_widget.py | 145 +++++ premium/widgets/dashboard_widget.py | 194 ++++++ premium/widgets/metrics_card.py | 121 ++++ 16 files changed, 3664 insertions(+) create mode 100644 main.py create mode 100644 plugins/builtin/dashboard_widget/main.py create mode 100644 plugins/builtin/dashboard_widget/plugin.json create mode 100644 premium/__init__.py create mode 100644 premium/core/event_bus.py create mode 100644 premium/eu_integration/__init__.py create mode 100644 premium/eu_integration/events.py create mode 100644 premium/eu_integration/game_client.py create mode 100644 premium/eu_integration/log_parser.py create mode 100644 premium/eu_integration/window_tracker.py create mode 100644 premium/plugins/api.py create mode 100644 premium/widgets/__init__.py create mode 100644 premium/widgets/base.py create mode 100644 premium/widgets/chart_widget.py create mode 100644 premium/widgets/dashboard_widget.py create mode 100644 premium/widgets/metrics_card.py diff --git a/main.py b/main.py new file mode 100644 index 0000000..c2e14ee --- /dev/null +++ b/main.py @@ -0,0 +1,297 @@ +#!/usr/bin/env python3 +""" +EU-Utility Premium - Main Entry Point +====================================== + +Launch the EU-Utility premium overlay system. + +Usage: + python main.py # Start with default config + python main.py --config path/to/config.json + python main.py --no-overlay # Run without overlay (headless) + python main.py --test # Run in test mode + +For more information, see TESTING.md +""" + +import sys +import json +import logging +import argparse +from pathlib import Path + +# Add the EU-Utility directory to path +sys.path.insert(0, str(Path(__file__).parent)) + + +def setup_logging(verbose: bool = False) -> logging.Logger: + """Setup logging configuration.""" + level = logging.DEBUG if verbose else logging.INFO + + logging.basicConfig( + level=level, + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', + handlers=[ + logging.StreamHandler(sys.stdout), + ] + ) + + return logging.getLogger("EU-Utility") + + +def load_config(config_path: Path) -> dict: + """Load configuration from file.""" + if not config_path.exists(): + # Create default config + default_config = { + "game_path": None, + "overlay_enabled": True, + "overlay_mode": "overlay_toggle", + "hotkey": "ctrl+shift+b", + "plugins": { + "enabled": [], + "disabled": [] + }, + "ui": { + "theme": "dark", + "opacity": 0.95, + "scale": 1.0 + }, + "features": { + "loot_tracker": True, + "skill_tracker": True, + "global_alerts": True, + "analytics": True + } + } + + config_path.parent.mkdir(parents=True, exist_ok=True) + with open(config_path, 'w') as f: + json.dump(default_config, f, indent=2) + + return default_config + + with open(config_path, 'r') as f: + return json.load(f) + + +def main(): + """Main entry point.""" + parser = argparse.ArgumentParser( + description="EU-Utility Premium - Entropia Universe Overlay" + ) + parser.add_argument( + '--config', + type=Path, + default=Path.home() / '.eu-utility' / 'config.json', + help='Path to configuration file' + ) + parser.add_argument( + '--no-overlay', + action='store_true', + help='Run without overlay (headless mode)' + ) + parser.add_argument( + '--test', + action='store_true', + help='Run in test mode (no game required)' + ) + parser.add_argument( + '--verbose', '-v', + action='store_true', + help='Enable verbose logging' + ) + parser.add_argument( + '--setup', + action='store_true', + help='Run first-time setup' + ) + + args = parser.parse_args() + + # Setup logging + logger = setup_logging(args.verbose) + logger.info("=" * 50) + logger.info("EU-Utility Premium Starting") + logger.info("=" * 50) + + # Load config + config = load_config(args.config) + logger.info(f"Loaded config from: {args.config}") + + # Run setup if requested + if args.setup: + run_setup(config, args.config) + return 0 + + # Check for game path + if not config.get('game_path') and not args.test: + logger.error("No game path configured. Run with --setup or edit config.json") + print("\n" + "=" * 50) + print("First-time setup required!") + print("=" * 50) + print("\nPlease run: python main.py --setup") + print("\nOr manually edit the config file:") + print(f" {args.config}") + print("\nAnd set the game_path to your Entropia Universe installation.") + print("=" * 50) + return 1 + + # Import and start the application + try: + from premium import EUUtilityApp + + logger.info("Initializing application...") + app = EUUtilityApp(config_path=args.config) + + # Override config with CLI args + if args.no_overlay: + config['overlay_enabled'] = False + + if args.test: + logger.info("Running in TEST mode") + return run_test_mode(app, config) + + # Normal operation + logger.info("Starting application...") + app.initialize() + app.run() + + except KeyboardInterrupt: + logger.info("Interrupted by user") + return 0 + + except Exception as e: + logger.exception("Fatal error") + print(f"\nError: {e}") + return 1 + + +def run_setup(config: dict, config_path: Path) -> None: + """Run first-time setup wizard.""" + print("\n" + "=" * 50) + print("EU-Utility Setup Wizard") + print("=" * 50 + "\n") + + # Detect game path + print("Looking for Entropia Universe installation...") + + possible_paths = [ + Path("C:/Program Files (x86)/Entropia Universe"), + Path("C:/Program Files/Entropia Universe"), + Path.home() / "AppData" / "Local" / "Entropia Universe", + ] + + detected = None + for path in possible_paths: + if path.exists(): + detected = path + print(f" Found: {path}") + break + + if detected: + use = input(f"\nUse this path? [Y/n]: ").strip().lower() + if use in ('', 'y', 'yes'): + config['game_path'] = str(detected) + else: + detected = None + + if not detected: + custom = input("Enter your Entropia Universe installation path: ").strip() + if custom: + config['game_path'] = custom + + # Save config + config_path.parent.mkdir(parents=True, exist_ok=True) + with open(config_path, 'w') as f: + json.dump(config, f, indent=2) + + print(f"\nConfiguration saved to: {config_path}") + print("\nYou can now run: python main.py") + print("=" * 50) + + +def run_test_mode(app, config: dict) -> int: + """Run in test mode without requiring game.""" + print("\n" + "=" * 50) + print("EU-Utility TEST MODE") + print("=" * 50 + "\n") + + try: + # Test imports + print("✓ Checking imports...") + from premium.plugins.api import PluginAPI, PluginManifest + from premium.plugins.manager import PluginManager + from premium.core.event_bus import EventBus + from premium.core.state.store import StateStore + print(" All imports successful") + + # Test plugin system + print("\n✓ Testing plugin system...") + event_bus = EventBus() + store = StateStore(reducer=lambda s, a: s or {}, initial_state={}) + + plugin_manager = PluginManager( + plugin_dirs=[], + data_dir=Path.home() / '.eu-utility' / 'data', + event_bus=event_bus, + state_store=store + ) + print(" Plugin manager initialized") + + # Test event bus + print("\n✓ Testing event bus...") + test_events = [] + + @event_bus.on('test') + def on_test(event): + test_events.append(event) + + event_bus.emit('test', {'message': 'hello'}) + import time + time.sleep(0.1) + + if len(test_events) == 1: + print(" Event bus working") + else: + print(" ⚠ Event bus may have issues") + + # Test game client (simulated) + print("\n✓ Testing game client integration...") + from premium.eu_integration import GameClient + + # Test with dummy path + client = GameClient( + game_path=Path.home(), # Dummy path for testing + event_bus=event_bus + ) + print(" Game client initialized (simulated mode)") + + # Test widget system + print("\n✓ Testing widget system...") + from premium.widgets import Widget, WidgetConfig + + class TestWidget(Widget): + def create_ui(self, parent): + return None # No UI in test mode + + widget = TestWidget(WidgetConfig(name="Test")) + print(" Widget system functional") + + print("\n" + "=" * 50) + print("All tests passed!") + print("=" * 50) + print("\nThe application is ready to use.") + print("Run 'python main.py' to start with the actual game.") + + return 0 + + except Exception as e: + print(f"\n✗ Test failed: {e}") + import traceback + traceback.print_exc() + return 1 + + +if __name__ == '__main__': + sys.exit(main()) diff --git a/plugins/builtin/dashboard_widget/main.py b/plugins/builtin/dashboard_widget/main.py new file mode 100644 index 0000000..5bbd786 --- /dev/null +++ b/plugins/builtin/dashboard_widget/main.py @@ -0,0 +1,322 @@ +""" +EU-Utility Premium - Dashboard Widget Plugin +============================================= + +A built-in plugin that provides a dashboard for player statistics. +Shows skills, loot, and game status in a convenient overlay widget. +""" + +from pathlib import Path +from typing import Optional, Dict, Any + +# Import the plugin API +from premium.plugins.api import ( + PluginAPI, PluginContext, PluginManifest, PermissionLevel +) + +# Manifest - REQUIRED +manifest = PluginManifest( + name="Player Dashboard", + version="1.0.0", + author="EU-Utility Team", + description="Player statistics dashboard with skills, loot, and game status", + entry_point="main.py", + permissions={PermissionLevel.UI, PermissionLevel.FILE_READ}, + tags=["dashboard", "stats", "built-in"], +) + + +class DashboardPlugin(PluginAPI): + """Player statistics dashboard plugin. + + This plugin displays: + - Current character name and level + - Skill gains (latest and session total) + - Loot tracker (recent items and total value) + - Game connection status + """ + + manifest = manifest + + def __init__(self): + super().__init__() + self._widget = None + self._stats = { + 'character': None, + 'session_loot_value': 0.0, + 'session_loot_count': 0, + 'session_skills': [], + 'last_loot': None, + 'last_skill': None, + } + + def on_init(self, ctx: PluginContext) -> None: + """Called when plugin is initialized.""" + self.ctx = ctx + ctx.logger.info("Dashboard plugin initialized") + + # Load saved stats if any + self._load_stats() + + # Subscribe to events + if ctx.event_bus: + ctx.event_bus.subscribe('game.loot', self._on_loot) + ctx.event_bus.subscribe('game.skill', self._on_skill) + ctx.event_bus.subscribe('game.connected', self._on_connected) + ctx.event_bus.subscribe('game.disconnected', self._on_disconnected) + + def on_activate(self) -> None: + """Called when plugin is activated.""" + self.ctx.logger.info("Dashboard plugin activated") + + def on_deactivate(self) -> None: + """Called when plugin is deactivated.""" + self._save_stats() + + def on_shutdown(self) -> None: + """Called when plugin is being unloaded.""" + self._save_stats() + + def create_widget(self) -> Optional[Any]: + """Create the dashboard UI widget.""" + try: + from PyQt6.QtWidgets import ( + QWidget, QVBoxLayout, QHBoxLayout, QLabel, + QFrame, QPushButton, QGridLayout + ) + from PyQt6.QtCore import Qt + from PyQt6.QtGui import QFont + + # Main widget + widget = QWidget() + widget.setStyleSheet(""" + QWidget { + background-color: #1e1e1e; + color: #ffffff; + font-family: 'Segoe UI', Arial, sans-serif; + } + QFrame { + background-color: #2d2d2d; + border-radius: 6px; + padding: 10px; + } + QLabel { + color: #ffffff; + } + .stat-label { + color: #888888; + font-size: 11px; + } + .stat-value { + font-size: 16px; + font-weight: bold; + } + """) + + layout = QVBoxLayout(widget) + layout.setSpacing(15) + layout.setContentsMargins(15, 15, 15, 15) + + # Header + header = QLabel("👤 Player Dashboard") + font = QFont("Segoe UI", 14, QFont.Weight.Bold) + header.setFont(font) + layout.addWidget(header) + + # Status section + status_frame = QFrame() + status_layout = QHBoxLayout(status_frame) + + self.status_label = QLabel("● Disconnected") + self.status_label.setStyleSheet("color: #f44336; font-weight: bold;") + status_layout.addWidget(self.status_label) + + self.char_label = QLabel("No character") + status_layout.addStretch() + status_layout.addWidget(self.char_label) + + layout.addWidget(status_frame) + + # Stats grid + stats_frame = QFrame() + stats_layout = QGridLayout(stats_frame) + stats_layout.setSpacing(10) + + # Session loot + loot_title = QLabel("💰 Session Loot") + loot_title.setStyleSheet("font-weight: bold;") + stats_layout.addWidget(loot_title, 0, 0) + + self.loot_value_label = QLabel("0.00 PED") + self.loot_value_label.setStyleSheet("font-size: 18px; font-weight: bold; color: #4caf50;") + stats_layout.addWidget(self.loot_value_label, 1, 0) + + self.loot_count_label = QLabel("0 items") + self.loot_count_label.setStyleSheet("color: #888888;") + stats_layout.addWidget(self.loot_count_label, 2, 0) + + # Skills + skills_title = QLabel("📈 Skills") + skills_title.setStyleSheet("font-weight: bold;") + stats_layout.addWidget(skills_title, 0, 1) + + self.skills_count_label = QLabel("0 gains") + self.skills_count_label.setStyleSheet("font-size: 18px; font-weight: bold; color: #2196f3;") + stats_layout.addWidget(self.skills_count_label, 1, 1) + + self.last_skill_label = QLabel("-") + self.last_skill_label.setStyleSheet("color: #888888;") + stats_layout.addWidget(self.last_skill_label, 2, 1) + + layout.addWidget(stats_frame) + + # Recent activity + activity_title = QLabel("📝 Recent Activity") + activity_title.setStyleSheet("font-weight: bold; margin-top: 10px;") + layout.addWidget(activity_title) + + self.activity_label = QLabel("No activity yet...") + self.activity_label.setWordWrap(True) + self.activity_label.setStyleSheet("color: #aaaaaa; padding: 5px;") + layout.addWidget(self.activity_label) + + layout.addStretch() + + # Save reference for updates + self._widget = widget + self._update_display() + + return widget + + except ImportError: + self.ctx.logger.warning("PyQt6 not available, cannot create widget") + return None + + # ========== Event Handlers ========== + + def _on_loot(self, event: Dict[str, Any]) -> None: + """Handle loot event.""" + item = event.get('item', 'Unknown') + value = event.get('value', 0.0) + quantity = event.get('quantity', 1) + + self._stats['session_loot_value'] += value + self._stats['session_loot_count'] += quantity + self._stats['last_loot'] = { + 'item': item, + 'value': value, + 'quantity': quantity, + } + + self.ctx.logger.info(f"Loot: {item} x{quantity} ({value:.2f} PED)") + self._update_display() + + def _on_skill(self, event: Dict[str, Any]) -> None: + """Handle skill gain event.""" + skill = event.get('skill', 'Unknown') + gain = event.get('gain', 0.0) + + self._stats['session_skills'].append({ + 'skill': skill, + 'gain': gain, + }) + self._stats['last_skill'] = { + 'skill': skill, + 'gain': gain, + } + + self.ctx.logger.info(f"Skill: {skill} +{gain:.4f}") + self._update_display() + + def _on_connected(self, event: Dict[str, Any]) -> None: + """Handle game connection.""" + self.ctx.logger.info("Game connected") + self._update_status(True) + + def _on_disconnected(self, event: Dict[str, Any]) -> None: + """Handle game disconnection.""" + self.ctx.logger.info("Game disconnected") + self._update_status(False) + + # ========== Display Updates ========== + + def _update_display(self) -> None: + """Update the widget display.""" + if not self._widget: + return + + try: + # Update loot stats + self.loot_value_label.setText( + f"{self._stats['session_loot_value']:.2f} PED" + ) + self.loot_count_label.setText( + f"{self._stats['session_loot_count']} items" + ) + + # Update skill stats + self.skills_count_label.setText( + f"{len(self._stats['session_skills'])} gains" + ) + + if self._stats['last_skill']: + skill = self._stats['last_skill'] + self.last_skill_label.setText( + f"{skill['skill']} +{skill['gain']:.4f}" + ) + + # Update activity + activity_text = "" + if self._stats['last_loot']: + loot = self._stats['last_loot'] + activity_text += f"Loot: {loot['item']} ({loot['value']:.2f} PED)\n" + + if self._stats['last_skill']: + skill = self._stats['last_skill'] + activity_text += f"Skill: {skill['skill']} +{skill['gain']:.4f}" + + if activity_text: + self.activity_label.setText(activity_text) + + except Exception as e: + self.ctx.logger.error(f"Error updating display: {e}") + + def _update_status(self, connected: bool) -> None: + """Update connection status display.""" + if not self._widget: + return + + if connected: + self.status_label.setText("● Connected") + self.status_label.setStyleSheet("color: #4caf50; font-weight: bold;") + else: + self.status_label.setText("● Disconnected") + self.status_label.setStyleSheet("color: #f44336; font-weight: bold;") + + # ========== Persistence ========== + + def _load_stats(self) -> None: + """Load stats from disk.""" + try: + import json + stats_path = self.ctx.data_dir / "session_stats.json" + if stats_path.exists(): + with open(stats_path, 'r') as f: + saved = json.load(f) + self._stats.update(saved) + except Exception as e: + self.ctx.logger.error(f"Failed to load stats: {e}") + + def _save_stats(self) -> None: + """Save stats to disk.""" + try: + import json + stats_path = self.ctx.data_dir / "session_stats.json" + with open(stats_path, 'w') as f: + json.dump(self._stats, f, indent=2) + except Exception as e: + self.ctx.logger.error(f"Failed to save stats: {e}") + + +# Export the plugin class +Plugin = DashboardPlugin diff --git a/plugins/builtin/dashboard_widget/plugin.json b/plugins/builtin/dashboard_widget/plugin.json new file mode 100644 index 0000000..68a3a5b --- /dev/null +++ b/plugins/builtin/dashboard_widget/plugin.json @@ -0,0 +1,12 @@ +{ + "name": "Player Dashboard", + "version": "1.0.0", + "author": "EU-Utility Team", + "description": "Player statistics dashboard with skills, loot, and game status", + "entry_point": "main.py", + "permissions": ["ui", "file_read"], + "dependencies": {}, + "min_api_version": "3.0.0", + "tags": ["dashboard", "stats", "built-in"], + "icon": "📊" +} diff --git a/premium/__init__.py b/premium/__init__.py new file mode 100644 index 0000000..c8d3829 --- /dev/null +++ b/premium/__init__.py @@ -0,0 +1,173 @@ +""" +EU-Utility Premium +================== + +Enterprise-grade modular overlay system for Entropia Universe. + +This package provides the premium/enterprise layer with: +- Advanced plugin system with sandboxing +- Redux-inspired state management +- High-performance event bus +- Widget system for dashboard +- Entropia Universe game integration + +Quick Start: + from premium import EUUtilityApp + + app = EUUtilityApp() + app.run() + +Modules: + plugins - Plugin management system + core - Core infrastructure (state, events) + widgets - Dashboard widget system + eu_integration - Entropia Universe integration +""" + +__version__ = "3.0.0" +__author__ = "EU-Utility Team" + +# Core exports +from premium.core.state.store import StateStore, ActionBase, create_store, get_store +from premium.core.event_bus import EventBus, Event, EventPriority +from premium.plugins.api import ( + PluginAPI, PluginManifest, PluginContext, PluginInstance, + PluginState, PermissionLevel, + PluginError, PluginLoadError, PluginInitError +) +from premium.plugins.manager import PluginManager + +# Widget system +from premium.widgets.base import Widget, WidgetConfig +from premium.widgets.dashboard import Dashboard + +# EU Integration +from premium.eu_integration.game_client import GameClient +from premium.eu_integration.log_parser import LogParser +from premium.eu_integration.events import GameEvent, LootEvent, SkillEvent + + +class EUUtilityApp: + """Main application class for EU-Utility Premium. + + This is the high-level API for the entire premium system. + + Example: + app = EUUtilityApp() + app.initialize() + app.run() + """ + + def __init__(self, config_path=None): + self.config_path = config_path + self.plugin_manager = None + self.state_store = None + self.event_bus = None + self.game_client = None + self.dashboard = None + self._initialized = False + + def initialize(self): + """Initialize all subsystems.""" + from pathlib import Path + + # Create event bus + self.event_bus = EventBus() + + # Create state store + self.state_store = StateStore( + reducer=self._root_reducer, + initial_state=self._get_initial_state() + ) + + # Create plugin manager + plugin_dirs = [ + Path(__file__).parent.parent / "plugins" / "builtin", + Path(__file__).parent.parent / "plugins" / "user", + ] + data_dir = Path.home() / ".eu-utility" / "data" + + self.plugin_manager = PluginManager( + plugin_dirs=plugin_dirs, + data_dir=data_dir, + event_bus=self.event_bus, + state_store=self.state_store + ) + + # Create game client + self.game_client = GameClient(event_bus=self.event_bus) + + self._initialized = True + + def run(self): + """Run the application.""" + if not self._initialized: + self.initialize() + + # Discover and load plugins + self.plugin_manager.discover_all() + self.plugin_manager.load_all(auto_activate=True) + + # Start game client + self.game_client.start() + + # Run main loop + try: + self._main_loop() + except KeyboardInterrupt: + self.shutdown() + + def shutdown(self): + """Shutdown the application gracefully.""" + if self.plugin_manager: + self.plugin_manager.shutdown() + if self.game_client: + self.game_client.stop() + + def _root_reducer(self, state, action): + """Root state reducer.""" + # Default reducer - just returns state + # Plugins can register their own reducers + return state or {} + + def _get_initial_state(self): + """Get initial application state.""" + return { + 'app': { + 'version': __version__, + 'initialized': False, + 'game_connected': False, + }, + 'player': { + 'name': None, + 'skills': {}, + 'loot': [], + }, + 'plugins': {}, + } + + def _main_loop(self): + """Main application loop.""" + import time + while True: + time.sleep(0.1) + + +__all__ = [ + # Version + '__version__', + # Core classes + 'EUUtilityApp', + # State management + 'StateStore', 'ActionBase', 'create_store', 'get_store', + # Event system + 'EventBus', 'Event', 'EventPriority', + # Plugin system + 'PluginManager', 'PluginAPI', 'PluginManifest', 'PluginContext', + 'PluginInstance', 'PluginState', 'PermissionLevel', + 'PluginError', 'PluginLoadError', 'PluginInitError', + # Widgets + 'Widget', 'WidgetConfig', 'Dashboard', + # EU Integration + 'GameClient', 'LogParser', 'GameEvent', 'LootEvent', 'SkillEvent', +] diff --git a/premium/core/event_bus.py b/premium/core/event_bus.py new file mode 100644 index 0000000..0f3bf7a --- /dev/null +++ b/premium/core/event_bus.py @@ -0,0 +1,599 @@ +""" +EU-Utility Premium - Event Bus +=============================== + +High-performance async event system for inter-plugin communication. + +Features: +- Async/await support +- Event prioritization +- Event filtering and routing +- Buffered events for high-throughput scenarios +- Type-safe event definitions + +Example: + from premium.core.event_bus import EventBus, Event, EventPriority + + bus = EventBus() + + # Subscribe to events + @bus.on('game.loot') + async def handle_loot(event: Event): + print(f"Got loot: {event.data}") + + # Emit events + bus.emit('game.loot', {'item': 'Angel Scales', 'value': 150.50}) +""" + +from __future__ import annotations + +import asyncio +import inspect +import logging +import time +import traceback +import weakref +from abc import ABC, abstractmethod +from collections import defaultdict, deque +from dataclasses import dataclass, field +from enum import Enum, auto +from typing import ( + Any, Callable, Coroutine, Dict, List, Optional, Set, Type, TypeVar, + Union, Generic, Protocol, runtime_checkable +) + + +# ============================================================================= +# EVENT PRIORITY +# ============================================================================= + +class EventPriority(Enum): + """Priority levels for event handlers. + + Higher priority handlers are called first. + """ + CRITICAL = 100 # System-critical events (error handling, shutdown) + HIGH = 75 # Important game events (loot, globals) + NORMAL = 50 # Standard events + LOW = 25 # Background tasks + BACKGROUND = 10 # Analytics, logging + + +# ============================================================================= +# EVENT CLASS +# ============================================================================= + +@dataclass +class Event: + """Represents an event in the system. + + Attributes: + type: Event type/category + data: Event payload data + source: Source of the event (plugin_id or system) + timestamp: When the event was created + priority: Event priority + id: Unique event identifier + """ + type: str + data: Dict[str, Any] = field(default_factory=dict) + source: str = "system" + timestamp: float = field(default_factory=time.time) + priority: EventPriority = EventPriority.NORMAL + id: str = field(default_factory=lambda: f"evt_{int(time.time()*1000)}_{id(Event) % 10000}") + + def get(self, key: str, default: Any = None) -> Any: + """Get a value from event data.""" + return self.data.get(key, default) + + def to_dict(self) -> Dict[str, Any]: + """Convert event to dictionary.""" + return { + 'id': self.id, + 'type': self.type, + 'data': self.data, + 'source': self.source, + 'timestamp': self.timestamp, + 'priority': self.priority.name, + } + + @classmethod + def create( + cls, + event_type: str, + data: Dict[str, Any], + source: str = "system", + priority: EventPriority = EventPriority.NORMAL + ) -> Event: + """Create a new event.""" + return cls( + type=event_type, + data=data, + source=source, + priority=priority + ) + + +# ============================================================================= +# EVENT FILTER +# ============================================================================= + +@dataclass +class EventFilter: + """Filter for event subscription. + + Allows subscribers to receive only events matching specific criteria. + + Example: + filter = EventFilter( + event_types=['game.loot', 'game.skill'], + sources=['game_client'], + min_priority=EventPriority.HIGH + ) + """ + event_types: Optional[Set[str]] = None + sources: Optional[Set[str]] = None + min_priority: Optional[EventPriority] = None + data_filter: Optional[Callable[[Dict[str, Any]], bool]] = None + + def matches(self, event: Event) -> bool: + """Check if an event matches this filter.""" + if self.event_types and event.type not in self.event_types: + return False + + if self.sources and event.source not in self.sources: + return False + + if self.min_priority and event.priority.value < self.min_priority.value: + return False + + if self.data_filter and not self.data_filter(event.data): + return False + + return True + + +# ============================================================================= +# SUBSCRIPTION +# ============================================================================= + +@dataclass +class Subscription: + """Represents an event subscription.""" + id: str + event_types: Set[str] + callback: Callable[[Event], Any] + filter: Optional[EventFilter] + priority: EventPriority + once: bool + + def matches(self, event: Event) -> bool: + """Check if this subscription matches an event.""" + if self.event_types and event.type not in self.event_types: + return False + + if self.filter and not self.filter.matches(event): + return False + + return True + + +# ============================================================================= +# EVENT BUS +# ============================================================================= + +HandlerType = Callable[[Event], Any] + + +class EventBus: + """High-performance async event bus. + + Features: + - Async and sync handler support + - Priority-based execution order + - Event filtering + - Buffered event queues + - Wildcard subscriptions + + Example: + bus = EventBus() + + # Sync handler + @bus.on('game.loot') + def handle_loot(event): + print(f"Loot: {event.data}") + + # Async handler + @bus.on('game.global', priority=EventPriority.HIGH) + async def handle_global(event): + await notify_discord(event.data) + + # Emit event + bus.emit('game.loot', {'item': 'Uber Item', 'value': 1000}) + """ + + def __init__(self, max_queue_size: int = 10000): + """Initialize event bus. + + Args: + max_queue_size: Maximum events in queue before dropping + """ + self._subscribers: Dict[str, Set[Subscription]] = defaultdict(set) + self._wildcard_subscribers: Set[Subscription] = set() + self._queue: deque = deque(maxlen=max_queue_size) + self._logger = logging.getLogger("EventBus") + self._sub_counter = 0 + self._running = False + self._lock = asyncio.Lock() + self._event_count = 0 + self._dropped_count = 0 + + # ========== Subscription Methods ========== + + def on( + self, + event_type: Union[str, List[str]], + priority: EventPriority = EventPriority.NORMAL, + filter: Optional[EventFilter] = None, + once: bool = False + ) -> Callable[[HandlerType], HandlerType]: + """Decorator to subscribe to events. + + Args: + event_type: Event type(s) to subscribe to + priority: Handler priority + filter: Optional event filter + once: Remove after first matching event + + Example: + @bus.on('game.loot') + def handle_loot(event): + print(event.data) + """ + def decorator(handler: HandlerType) -> HandlerType: + self.subscribe(event_type, handler, priority, filter, once) + return handler + return decorator + + def subscribe( + self, + event_type: Union[str, List[str]], + handler: HandlerType, + priority: EventPriority = EventPriority.NORMAL, + filter: Optional[EventFilter] = None, + once: bool = False + ) -> str: + """Subscribe to events. + + Args: + event_type: Event type(s) to subscribe to, or '*' for all + handler: Callback function + priority: Handler priority + filter: Optional event filter + once: Remove after first matching event + + Returns: + Subscription ID + """ + self._sub_counter += 1 + sub_id = f"sub_{self._sub_counter}" + + # Handle wildcard subscription + if event_type == '*': + sub = Subscription( + id=sub_id, + event_types=set(), + callback=handler, + filter=filter, + priority=priority, + once=once + ) + self._wildcard_subscribers.add(sub) + return sub_id + + # Handle single or multiple event types + if isinstance(event_type, str): + event_types = {event_type} + else: + event_types = set(event_type) + + sub = Subscription( + id=sub_id, + event_types=event_types, + callback=handler, + filter=filter, + priority=priority, + once=once + ) + + for et in event_types: + self._subscribers[et].add(sub) + + return sub_id + + def unsubscribe(self, subscription_id: str) -> bool: + """Unsubscribe from events. + + Args: + subscription_id: ID returned by subscribe() + + Returns: + True if subscription was found and removed + """ + # Check wildcard subscribers + for sub in self._wildcard_subscribers: + if sub.id == subscription_id: + self._wildcard_subscribers.remove(sub) + return True + + # Check typed subscribers + for event_type, subs in self._subscribers.items(): + for sub in list(subs): + if sub.id == subscription_id: + subs.remove(sub) + return True + + return False + + def once( + self, + event_type: str, + handler: HandlerType, + priority: EventPriority = EventPriority.NORMAL, + timeout: Optional[float] = None + ) -> asyncio.Future: + """Subscribe to a single event. + + Args: + event_type: Event type to wait for + handler: Optional handler callback + priority: Handler priority + timeout: Optional timeout in seconds + + Returns: + Future that resolves with the event + """ + future = asyncio.Future() + + def wrapper(event: Event): + if handler: + handler(event) + if not future.done(): + future.set_result(event) + + self.subscribe(event_type, wrapper, priority, once=True) + + if timeout: + async def timeout_handler(): + await asyncio.sleep(timeout) + if not future.done(): + future.set_exception(TimeoutError(f"Event {event_type} timed out")) + + asyncio.create_task(timeout_handler()) + + return future + + # ========== Event Emission ========== + + def emit( + self, + event_type: str, + data: Dict[str, Any], + source: str = "system", + priority: EventPriority = EventPriority.NORMAL + ) -> Event: + """Emit an event. + + Args: + event_type: Type of event + data: Event data + source: Event source + priority: Event priority + + Returns: + The created event + """ + event = Event.create( + event_type=event_type, + data=data, + source=source, + priority=priority + ) + + self._queue.append(event) + self._event_count += 1 + + # Process immediately in async context + if asyncio.get_event_loop().is_running(): + asyncio.create_task(self._process_event(event)) + + return event + + async def emit_async( + self, + event_type: str, + data: Dict[str, Any], + source: str = "system", + priority: EventPriority = EventPriority.NORMAL + ) -> Event: + """Emit an event asynchronously.""" + event = Event.create( + event_type=event_type, + data=data, + source=source, + priority=priority + ) + + await self._process_event(event) + return event + + async def _process_event(self, event: Event) -> None: + """Process a single event.""" + # Collect all matching handlers + handlers: List[tuple] = [] + + # Get handlers for specific event type + for sub in self._subscribers.get(event.type, set()): + if sub.matches(event): + handlers.append((sub.priority.value, sub)) + + # Get wildcard handlers + for sub in self._wildcard_subscribers: + if sub.filter is None or sub.filter.matches(event): + handlers.append((sub.priority.value, sub)) + + # Sort by priority (highest first) + handlers.sort(key=lambda x: -x[0]) + + # Execute handlers + once_subs = [] + + for _, sub in handlers: + try: + if inspect.iscoroutinefunction(sub.callback): + await sub.callback(event) + else: + sub.callback(event) + + if sub.once: + once_subs.append(sub) + + except Exception as e: + self._logger.error(f"Error in event handler: {e}") + traceback.print_exc() + + # Remove once subscriptions + for sub in once_subs: + self.unsubscribe(sub.id) + + # ========== Utility Methods ========== + + def clear(self) -> None: + """Clear all subscriptions and queued events.""" + self._subscribers.clear() + self._wildcard_subscribers.clear() + self._queue.clear() + + def get_stats(self) -> Dict[str, Any]: + """Get event bus statistics.""" + return { + 'total_events': self._event_count, + 'dropped_events': self._dropped_count, + 'queue_size': len(self._queue), + 'subscription_count': sum( + len(subs) for subs in self._subscribers.values() + ), + 'wildcard_subscriptions': len(self._wildcard_subscribers), + } + + def wait_for( + self, + event_type: str, + condition: Optional[Callable[[Event], bool]] = None, + timeout: Optional[float] = None + ) -> asyncio.Future: + """Wait for a specific event. + + Args: + event_type: Event type to wait for + condition: Optional condition function + timeout: Optional timeout in seconds + + Returns: + Future that resolves with the matching event + """ + future = asyncio.Future() + + def handler(event: Event): + if condition and not condition(event): + return + if not future.done(): + future.set_result(event) + return True # Unsubscribe + + self.subscribe(event_type, handler, once=True) + + if timeout: + async def timeout_coro(): + await asyncio.sleep(timeout) + if not future.done(): + future.set_exception(TimeoutError(f"Timeout waiting for {event_type}")) + + asyncio.create_task(timeout_coro()) + + return future + + +# ============================================================================= +# TYPED EVENTS +# ============================================================================= + +class TypedEventBus(EventBus): + """Type-safe event bus with predefined event types.""" + + # Game events + GAME_CONNECTED = "game.connected" + GAME_DISCONNECTED = "game.disconnected" + GAME_FOCUS_CHANGED = "game.focus_changed" + + # Loot events + LOOT_RECEIVED = "game.loot" + GLOBAL_HOF = "game.global" + + # Skill events + SKILL_GAIN = "game.skill" + PROFESSION_GAIN = "game.profession" + + # Chat events + CHAT_MESSAGE = "game.chat" + + # System events + PLUGIN_LOADED = "system.plugin.loaded" + PLUGIN_UNLOADED = "system.plugin.unloaded" + ERROR_OCCURRED = "system.error" + + def emit_game_connected(self, character_name: str) -> Event: + """Emit game connected event.""" + return self.emit(self.GAME_CONNECTED, { + 'character_name': character_name + }) + + def emit_loot(self, item: str, value: float, quantity: int = 1) -> Event: + """Emit loot received event.""" + return self.emit(self.LOOT_RECEIVED, { + 'item': item, + 'value': value, + 'quantity': quantity, + 'timestamp': time.time(), + }) + + def emit_skill(self, skill_name: str, value: float, gain: float) -> Event: + """Emit skill gain event.""" + return self.emit(self.SKILL_GAIN, { + 'skill': skill_name, + 'value': value, + 'gain': gain, + }) + + def emit_global(self, mob_name: str, value: float, player: str) -> Event: + """Emit global/HoF event.""" + return self.emit(self.GLOBAL_HOF, { + 'mob': mob_name, + 'value': value, + 'player': player, + 'is_hof': value >= 1000, + }) + + +# ============================================================================= +# EXPORTS +# ============================================================================= + +__all__ = [ + 'EventPriority', + 'Event', + 'EventFilter', + 'Subscription', + 'EventBus', + 'TypedEventBus', +] diff --git a/premium/eu_integration/__init__.py b/premium/eu_integration/__init__.py new file mode 100644 index 0000000..f527861 --- /dev/null +++ b/premium/eu_integration/__init__.py @@ -0,0 +1,33 @@ +""" +EU-Utility Premium - Entropia Universe Integration +=================================================== + +Modules for integrating with the Entropia Universe game client. + +Example: + from premium.eu_integration import GameClient + + client = GameClient() + client.start() + + @client.on_loot + def handle_loot(event): + print(f"Got loot: {event.item}") +""" + +from .game_client import GameClient, GameState +from .log_parser import LogParser, LogEvent +from .window_tracker import WindowTracker +from .events import GameEvent, LootEvent, SkillEvent, ChatEvent + +__all__ = [ + 'GameClient', + 'GameState', + 'LogParser', + 'LogEvent', + 'WindowTracker', + 'GameEvent', + 'LootEvent', + 'SkillEvent', + 'ChatEvent', +] diff --git a/premium/eu_integration/events.py b/premium/eu_integration/events.py new file mode 100644 index 0000000..a1a0e68 --- /dev/null +++ b/premium/eu_integration/events.py @@ -0,0 +1,131 @@ +""" +Typed event definitions for Entropia Universe. +""" + +from dataclasses import dataclass +from datetime import datetime +from typing import Optional, Dict, Any + + +@dataclass +class GameEvent: + """Base game event.""" + timestamp: datetime + type: str + data: Dict[str, Any] + + +@dataclass +class LootEvent(GameEvent): + """Loot received event.""" + item: str + quantity: int + value: float + + @classmethod + def from_data(cls, data: Dict[str, Any]) -> 'LootEvent': + return cls( + timestamp=datetime.now(), + type='loot', + data=data, + item=data.get('item', 'Unknown'), + quantity=data.get('quantity', 1), + value=data.get('value', 0.0), + ) + + +@dataclass +class SkillEvent(GameEvent): + """Skill gain event.""" + skill: str + gain: float + new_value: Optional[float] = None + + @classmethod + def from_data(cls, data: Dict[str, Any]) -> 'SkillEvent': + return cls( + timestamp=datetime.now(), + type='skill', + data=data, + skill=data.get('skill', 'Unknown'), + gain=data.get('gain', 0.0), + new_value=data.get('new_value'), + ) + + +@dataclass +class GlobalEvent(GameEvent): + """Global/HoF event.""" + player: str + mob: str + item: str + value: float + is_hof: bool + + @classmethod + def from_data(cls, data: Dict[str, Any]) -> 'GlobalEvent': + return cls( + timestamp=datetime.now(), + type='global', + data=data, + player=data.get('player', 'Unknown'), + mob=data.get('mob', 'Unknown'), + item=data.get('item', 'Unknown'), + value=data.get('value', 0.0), + is_hof=data.get('is_hof', False), + ) + + +@dataclass +class ChatEvent(GameEvent): + """Chat message event.""" + channel: str + player: str + message: str + + @classmethod + def from_data(cls, data: Dict[str, Any]) -> 'ChatEvent': + return cls( + timestamp=datetime.now(), + type='chat', + data=data, + channel=data.get('channel', 'Unknown'), + player=data.get('player', 'Unknown'), + message=data.get('message', ''), + ) + + +@dataclass +class DamageEvent(GameEvent): + """Damage dealt/received event.""" + damage: float + target: Optional[str] = None + is_critical: bool = False + + @classmethod + def from_data(cls, data: Dict[str, Any], dealt: bool = True) -> 'DamageEvent': + return cls( + timestamp=datetime.now(), + type='damage_dealt' if dealt else 'damage_received', + data=data, + damage=data.get('damage', 0.0), + target=data.get('target'), + is_critical=data.get('is_critical', False), + ) + + +@dataclass +class KillEvent(GameEvent): + """Creature killed event.""" + creature: str + experience: Optional[float] = None + + @classmethod + def from_data(cls, data: Dict[str, Any]) -> 'KillEvent': + return cls( + timestamp=datetime.now(), + type='kill', + data=data, + creature=data.get('creature', 'Unknown'), + experience=data.get('experience'), + ) diff --git a/premium/eu_integration/game_client.py b/premium/eu_integration/game_client.py new file mode 100644 index 0000000..80766fd --- /dev/null +++ b/premium/eu_integration/game_client.py @@ -0,0 +1,344 @@ +""" +Entropia Universe Game Client Integration +========================================== + +Provides integration with the EU game client: +- Process detection +- Window tracking +- Log file monitoring +- Event emission +""" + +import os +import re +import time +import logging +import threading +from dataclasses import dataclass, field +from enum import Enum, auto +from pathlib import Path +from typing import Optional, Callable, Dict, List, Any, Set + + +class GameState(Enum): + """Game client connection state.""" + DISCONNECTED = auto() + DETECTED = auto() # Process found + CONNECTED = auto() # Log file accessible + PLAYING = auto() # Character active + + +@dataclass +class CharacterInfo: + """Information about the current character.""" + name: Optional[str] = None + level: Optional[int] = None + profession: Optional[str] = None + health: Optional[float] = None + position: Optional[tuple] = None + + +class GameClient: + """Main interface for Entropia Universe game client. + + This class monitors the game client process, tracks window state, + parses log files, and emits events for game actions. + + Example: + client = GameClient() + + @client.on('loot') + def handle_loot(event): + print(f"Got: {event['item']}") + + client.start() + """ + + # Default paths + DEFAULT_INSTALL_PATHS = [ + Path("C:/Program Files (x86)/Entropia Universe"), + Path("C:/Program Files/Entropia Universe"), + Path.home() / "AppData" / "Local" / "Entropia Universe", + ] + + def __init__( + self, + game_path: Optional[Path] = None, + event_bus: Optional[Any] = None, + poll_interval: float = 1.0 + ): + """Initialize game client. + + Args: + game_path: Path to EU installation (auto-detected if None) + event_bus: Event bus for emitting events + poll_interval: Seconds between process checks + """ + self.game_path = game_path + self.event_bus = event_bus + self.poll_interval = poll_interval + + self._state = GameState.DISCONNECTED + self._character = CharacterInfo() + self._process_id: Optional[int] = None + self._window_handle: Optional[int] = None + self._log_path: Optional[Path] = None + + self._running = False + self._monitor_thread: Optional[threading.Thread] = None + self._callbacks: Dict[str, List[Callable]] = {} + self._logger = logging.getLogger("GameClient") + + # Sub-components + self._log_parser: Optional[Any] = None + self._window_tracker: Optional[Any] = None + + # ========== Properties ========== + + @property + def state(self) -> GameState: + """Current game connection state.""" + return self._state + + @property + def is_connected(self) -> bool: + """Whether game client is connected.""" + return self._state in (GameState.CONNECTED, GameState.PLAYING) + + @property + def is_playing(self) -> bool: + """Whether character is active in game.""" + return self._state == GameState.PLAYING + + @property + def character(self) -> CharacterInfo: + """Current character information.""" + return self._character + + # ========== Lifecycle ========== + + def start(self) -> bool: + """Start monitoring the game client. + + Returns: + True if started successfully + """ + if self._running: + return True + + self._running = True + + # Auto-detect game path if not provided + if not self.game_path: + self.game_path = self._detect_game_path() + + if self.game_path: + self._log_path = self.game_path / "chat.log" + + # Start monitoring thread + self._monitor_thread = threading.Thread(target=self._monitor_loop, daemon=True) + self._monitor_thread.start() + + self._logger.info("Game client monitor started") + return True + + def stop(self) -> None: + """Stop monitoring the game client.""" + self._running = False + + if self._monitor_thread: + self._monitor_thread.join(timeout=2.0) + + if self._log_parser: + self._log_parser.stop() + + self._logger.info("Game client monitor stopped") + + # ========== Event Handling ========== + + def on(self, event_type: str, callback: Callable) -> Callable: + """Register event callback. + + Args: + event_type: Event type ('loot', 'skill', 'global', etc.) + callback: Function to call when event occurs + + Returns: + The callback function (for use as decorator) + """ + if event_type not in self._callbacks: + self._callbacks[event_type] = [] + self._callbacks[event_type].append(callback) + return callback + + def off(self, event_type: str, callback: Callable) -> bool: + """Unregister event callback.""" + if event_type in self._callbacks: + if callback in self._callbacks[event_type]: + self._callbacks[event_type].remove(callback) + return True + return False + + def _emit(self, event_type: str, data: Dict[str, Any]) -> None: + """Emit event to callbacks and event bus.""" + # Call local callbacks + for callback in self._callbacks.get(event_type, []): + try: + callback(data) + except Exception as e: + self._logger.error(f"Error in callback: {e}") + + # Emit to event bus + if self.event_bus: + try: + self.event_bus.emit(f"game.{event_type}", data, source="game_client") + except Exception as e: + self._logger.error(f"Error emitting to event bus: {e}") + + # ========== Detection ========== + + def _detect_game_path(self) -> Optional[Path]: + """Auto-detect EU installation path.""" + # Check default paths + for path in self.DEFAULT_INSTALL_PATHS: + if path.exists() and (path / "Entropia.exe").exists(): + self._logger.info(f"Found EU at: {path}") + return path + + # Check registry on Windows + try: + import winreg + with winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, + r"SOFTWARE\Microsoft\Windows\CurrentVersion\Uninstall") as key: + for i in range(winreg.QueryInfoKey(key)[0]): + try: + subkey_name = winreg.EnumKey(key, i) + with winreg.OpenKey(key, subkey_name) as subkey: + name = winreg.QueryValueEx(subkey, "DisplayName")[0] + if "Entropia" in name: + path = winreg.QueryValueEx(subkey, "InstallLocation")[0] + return Path(path) + except: + continue + except Exception as e: + self._logger.debug(f"Registry search failed: {e}") + + return None + + def _find_process(self) -> Optional[int]: + """Find EU game process ID.""" + try: + import psutil + for proc in psutil.process_iter(['pid', 'name']): + try: + if proc.info['name'] and 'entropia' in proc.info['name'].lower(): + return proc.info['pid'] + except (psutil.NoSuchProcess, psutil.AccessDenied): + continue + except ImportError: + pass + + # Fallback: check if log file exists and is being written + if self._log_path and self._log_path.exists(): + # Check if file was modified recently + mtime = self._log_path.stat().st_mtime + if time.time() - mtime < 60: # Modified in last minute + return -1 # Unknown PID but likely running + + return None + + def _is_window_focused(self) -> bool: + """Check if EU window is focused.""" + try: + import win32gui + import win32process + + hwnd = win32gui.GetForegroundWindow() + _, pid = win32process.GetWindowThreadProcessId(hwnd) + + return pid == self._process_id + except ImportError: + pass + + return False + + # ========== Monitoring Loop ========== + + def _monitor_loop(self) -> None: + """Main monitoring loop.""" + last_log_position = 0 + + while self._running: + try: + # Check process + pid = self._find_process() + + if pid and self._state == GameState.DISCONNECTED: + self._process_id = pid + self._state = GameState.DETECTED + self._emit('connected', {'pid': pid}) + self._start_log_parsing() + + elif not pid and self._state != GameState.DISCONNECTED: + self._state = GameState.DISCONNECTED + self._process_id = None + self._emit('disconnected', {}) + self._stop_log_parsing() + + # Check window focus + if self._state != GameState.DISCONNECTED: + focused = self._is_window_focused() + self._emit('focus_changed', {'focused': focused}) + + time.sleep(self.poll_interval) + + except Exception as e: + self._logger.error(f"Monitor error: {e}") + time.sleep(self.poll_interval) + + def _start_log_parsing(self) -> None: + """Start parsing log files.""" + if not self._log_path or not self._log_path.exists(): + return + + from .log_parser import LogParser + + self._log_parser = LogParser(self._log_path) + + @self._log_parser.on_event + def handle_log_event(event_type, data): + self._handle_log_event(event_type, data) + + self._log_parser.start() + self._state = GameState.CONNECTED + + def _stop_log_parsing(self) -> None: + """Stop parsing log files.""" + if self._log_parser: + self._log_parser.stop() + self._log_parser = None + + def _handle_log_event(self, event_type: str, data: Dict[str, Any]) -> None: + """Handle events from log parser.""" + # Update character info + if event_type == 'system_message' and 'character' in data: + self._character.name = data.get('character') + self._state = GameState.PLAYING + + # Emit to listeners + self._emit(event_type, data) + + # ========== Public API ========== + + def get_game_path(self) -> Optional[Path]: + """Get detected game path.""" + return self.game_path + + def get_log_path(self) -> Optional[Path]: + """Get path to chat.log.""" + return self._log_path + + def set_game_path(self, path: Path) -> None: + """Manually set game path.""" + self.game_path = Path(path) + self._log_path = self.game_path / "chat.log" diff --git a/premium/eu_integration/log_parser.py b/premium/eu_integration/log_parser.py new file mode 100644 index 0000000..b828e06 --- /dev/null +++ b/premium/eu_integration/log_parser.py @@ -0,0 +1,323 @@ +""" +Log file parser for Entropia Universe chat.log +=============================================== + +Parses EU chat.log file to extract: +- Loot events +- Skill gains +- Chat messages +- System messages +- Globals/HoFs +""" + +import re +import time +import logging +import threading +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path +from typing import Optional, Callable, Dict, List, Any, Pattern + + +@dataclass +class LogEvent: + """Represents a parsed log event.""" + timestamp: datetime + type: str + data: Dict[str, Any] + raw: str + + +class LogParser: + """Parser for Entropia Universe chat.log file. + + Monitors the log file and emits events for interesting game actions. + + Example: + parser = LogParser(Path("C:/.../chat.log")) + + @parser.on_event + def handle(event_type, data): + if event_type == 'loot': + print(f"Loot: {data['item']}") + + parser.start() + """ + + # Regex patterns for different event types + PATTERNS = { + # Loot: "2024-01-15 14:30:25 [System] [] [Player] You received Angel Scales x1 Value: 150 PED" + 'loot': re.compile( + r'You received\s+(.+?)\s+x(\d+)\s+Value:\s+([\d.]+)' + ), + + # Skill gain: "Skill increase: Rifle (Gain: 0.5234)" + 'skill': re.compile( + r'Skill increase:\s+(.+?)\s+\(Gain:\s+([\d.]+)\)' + ), + + # Global/HoF: "[Global] [Player] found something rare! Angel Scales (Uber)" + 'global': re.compile( + r'\[Global\].*?found something rare!\s+(.+?)\s+\((\d+)\s*PED\)' + ), + + # HoF (Hall of Fame): "[Hall of Fame] [Player] found something extraordinary!" + 'hof': re.compile( + r'\[Hall of Fame\].*?found something extraordinary!\s+(.+?)\s+\((\d+)\s*PED\)' + ), + + # Chat message: "[Trade] [PlayerName]: Message text" + 'chat': re.compile( + r'\[(\w+)\]\s+\[(.+?)\]:\s+(.*)' + ), + + # Damage dealt: "You inflicted 45.2 points of damage" + 'damage_dealt': re.compile( + r'You inflicted\s+([\d.]+)\s+points of damage' + ), + + # Damage received: "You took 12.3 points of damage" + 'damage_received': re.compile( + r'You took\s+([\d.]+)\s+points of damage' + ), + + # Enemy killed: "You killed [Creature Name]" + 'kill': re.compile( + r'You killed\s+\[(.+?)\]' + ), + + # Item crafted: "You have successfully crafted [Item Name]" + 'craft': re.compile( + r'You have successfully crafted\s+\[(.+?)\]' + ), + + # Mission completed: "Mission "Mission Name" completed!" + 'mission_complete': re.compile( + r'Mission\s+"(.+?)"\s+completed!' + ), + + # Experience gain: "You gained 0.2345 experience in your Rifle skill" + 'experience': re.compile( + r'You gained\s+([\d.]+)\s+experience in your\s+(.+?)\s+skill' + ), + + # Item looted (alt format): "You looted [Item Name]" + 'loot_alt': re.compile( + r'You looted\s+\[(.+?)\]\s*(?:x(\d+))?' + ), + } + + def __init__( + self, + log_path: Path, + poll_interval: float = 0.5, + encoding: str = 'utf-8' + ): + """Initialize log parser. + + Args: + log_path: Path to chat.log file + poll_interval: Seconds between file checks + encoding: File encoding + """ + self.log_path = Path(log_path) + self.poll_interval = poll_interval + self.encoding = encoding + + self._running = False + self._thread: Optional[threading.Thread] = None + self._file_position = 0 + self._callbacks: List[Callable] = [] + self._logger = logging.getLogger("LogParser") + + # Track last position to detect new content + self._last_size = 0 + self._last_modified = 0 + + def start(self) -> bool: + """Start parsing log file. + + Returns: + True if started successfully + """ + if self._running: + return True + + if not self.log_path.exists(): + self._logger.error(f"Log file not found: {self.log_path}") + return False + + self._running = True + + # Start at end of file (only read new content) + try: + self._last_size = self.log_path.stat().st_size + self._file_position = self._last_size + except Exception as e: + self._logger.error(f"Failed to get file size: {e}") + return False + + self._thread = threading.Thread(target=self._parse_loop, daemon=True) + self._thread.start() + + self._logger.info(f"Started parsing {self.log_path}") + return True + + def stop(self) -> None: + """Stop parsing log file.""" + self._running = False + + if self._thread: + self._thread.join(timeout=2.0) + + self._logger.info("Stopped parsing") + + def on_event(self, callback: Callable[[str, Dict[str, Any]], None]) -> Callable: + """Register event callback. + + Args: + callback: Function(event_type, data) called for each event + + Returns: + The callback function (for use as decorator) + """ + self._callbacks.append(callback) + return callback + + def _emit(self, event_type: str, data: Dict[str, Any]) -> None: + """Emit event to all callbacks.""" + for callback in self._callbacks: + try: + callback(event_type, data) + except Exception as e: + self._logger.error(f"Error in callback: {e}") + + def _parse_loop(self) -> None: + """Main parsing loop.""" + while self._running: + try: + self._check_file() + time.sleep(self.poll_interval) + except Exception as e: + self._logger.error(f"Parse error: {e}") + time.sleep(self.poll_interval) + + def _check_file(self) -> None: + """Check log file for new content.""" + try: + stat = self.log_path.stat() + current_size = stat.st_size + current_modified = stat.st_mtime + + # Check if file has new content + if current_size == self._last_size: + return + + if current_size < self._last_size: + # File was truncated or rotated, start from beginning + self._file_position = 0 + + self._last_size = current_size + self._last_modified = current_modified + + # Read new content + with open(self.log_path, 'r', encoding=self.encoding, errors='ignore') as f: + f.seek(self._file_position) + new_lines = f.readlines() + self._file_position = f.tell() + + # Parse each new line + for line in new_lines: + self._parse_line(line.strip()) + + except Exception as e: + self._logger.error(f"Error checking file: {e}") + + def _parse_line(self, line: str) -> Optional[LogEvent]: + """Parse a single log line.""" + if not line: + return None + + # Extract timestamp if present + timestamp = datetime.now() + ts_match = re.match(r'(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})', line) + if ts_match: + try: + timestamp = datetime.strptime(ts_match.group(1), '%Y-%m-%d %H:%M:%S') + except ValueError: + pass + + # Try each pattern + for event_type, pattern in self.PATTERNS.items(): + match = pattern.search(line) + if match: + data = self._extract_data(event_type, match, line) + event = LogEvent( + timestamp=timestamp, + type=event_type, + data=data, + raw=line + ) + self._emit(event_type, data) + return event + + return None + + def _extract_data( + self, + event_type: str, + match: re.Match, + raw_line: str + ) -> Dict[str, Any]: + """Extract data from regex match.""" + groups = match.groups() + + data = { + 'raw': raw_line, + 'timestamp': datetime.now().isoformat(), + } + + if event_type == 'loot': + data['item'] = groups[0].strip() + data['quantity'] = int(groups[1]) if len(groups) > 1 else 1 + data['value'] = float(groups[2]) if len(groups) > 2 else 0.0 + + elif event_type == 'skill': + data['skill'] = groups[0].strip() + data['gain'] = float(groups[1]) if len(groups) > 1 else 0.0 + + elif event_type in ('global', 'hof'): + data['item'] = groups[0].strip() if groups else 'Unknown' + data['value'] = float(groups[1]) if len(groups) > 1 else 0.0 + data['is_hof'] = event_type == 'hof' + + elif event_type == 'chat': + data['channel'] = groups[0] if groups else 'Unknown' + data['player'] = groups[1] if len(groups) > 1 else 'Unknown' + data['message'] = groups[2] if len(groups) > 2 else '' + + elif event_type == 'damage_dealt': + data['damage'] = float(groups[0]) if groups else 0.0 + + elif event_type == 'damage_received': + data['damage'] = float(groups[0]) if groups else 0.0 + + elif event_type == 'kill': + data['creature'] = groups[0] if groups else 'Unknown' + + elif event_type == 'craft': + data['item'] = groups[0] if groups else 'Unknown' + + elif event_type == 'mission_complete': + data['mission'] = groups[0] if groups else 'Unknown' + + elif event_type == 'experience': + data['amount'] = float(groups[0]) if groups else 0.0 + data['skill'] = groups[1].strip() if len(groups) > 1 else 'Unknown' + + elif event_type == 'loot_alt': + data['item'] = groups[0] if groups else 'Unknown' + data['quantity'] = int(groups[1]) if len(groups) > 1 and groups[1] else 1 + + return data diff --git a/premium/eu_integration/window_tracker.py b/premium/eu_integration/window_tracker.py new file mode 100644 index 0000000..0723056 --- /dev/null +++ b/premium/eu_integration/window_tracker.py @@ -0,0 +1,107 @@ +""" +Window tracker for monitoring Entropia Universe window state. +""" + +import time +import logging +import threading +from typing import Optional, Callable, List + + +class WindowTracker: + """Tracks the EU game window state. + + Monitors window focus, visibility, and position without + interfering with the game. + """ + + def __init__(self, poll_interval: float = 0.5): + """Initialize window tracker. + + Args: + poll_interval: Seconds between window checks + """ + self.poll_interval = poll_interval + + self._running = False + self._thread: Optional[threading.Thread] = None + self._callbacks: List[Callable] = [] + self._logger = logging.getLogger("WindowTracker") + + self._is_focused = False + self._window_handle: Optional[int] = None + self._process_id: Optional[int] = None + + def start(self, process_id: Optional[int] = None) -> bool: + """Start tracking window.""" + if self._running: + return True + + self._process_id = process_id + self._running = True + + self._thread = threading.Thread(target=self._track_loop, daemon=True) + self._thread.start() + + return True + + def stop(self) -> None: + """Stop tracking window.""" + self._running = False + + if self._thread: + self._thread.join(timeout=1.0) + + def on_change(self, callback: Callable[[bool], None]) -> Callable: + """Register focus change callback.""" + self._callbacks.append(callback) + return callback + + def is_focused(self) -> bool: + """Check if game window is focused.""" + return self._is_focused + + def _track_loop(self) -> None: + """Main tracking loop.""" + while self._running: + try: + focused = self._check_focus() + + if focused != self._is_focused: + self._is_focused = focused + self._notify_change(focused) + + time.sleep(self.poll_interval) + + except Exception as e: + self._logger.error(f"Tracking error: {e}") + time.sleep(self.poll_interval) + + def _check_focus(self) -> bool: + """Check if game window is focused.""" + try: + import win32gui + import win32process + + hwnd = win32gui.GetForegroundWindow() + + if self._process_id: + _, pid = win32process.GetWindowThreadProcessId(hwnd) + return pid == self._process_id + else: + # Check window title + title = win32gui.GetWindowText(hwnd) + return 'entropia' in title.lower() + + except ImportError: + pass + + return False + + def _notify_change(self, focused: bool) -> None: + """Notify callbacks of focus change.""" + for callback in self._callbacks: + try: + callback(focused) + except Exception as e: + self._logger.error(f"Callback error: {e}") diff --git a/premium/plugins/api.py b/premium/plugins/api.py new file mode 100644 index 0000000..d2f5454 --- /dev/null +++ b/premium/plugins/api.py @@ -0,0 +1,465 @@ +""" +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', +] diff --git a/premium/widgets/__init__.py b/premium/widgets/__init__.py new file mode 100644 index 0000000..b2ff94d --- /dev/null +++ b/premium/widgets/__init__.py @@ -0,0 +1,378 @@ +""" +EU-Utility Premium - Widget System +=================================== + +Dashboard widget system for creating plugin UIs. + +Example: + from premium.widgets import Widget, WidgetConfig + + class MyWidget(Widget): + def __init__(self): + super().__init__(WidgetConfig( + name="My Widget", + icon="📊", + size=(300, 200) + )) + + def create_ui(self, parent): + # Create and return your widget + label = QLabel("Hello World", parent) + return label +""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from dataclasses import dataclass, field +from enum import Enum, auto +from typing import Any, Callable, Dict, List, Optional, Tuple, Union + + +# ============================================================================= +# WIDGET CONFIG +# ============================================================================= + +class WidgetSize(Enum): + """Standard widget sizes.""" + SMALL = (150, 100) + MEDIUM = (300, 200) + LARGE = (450, 300) + WIDE = (600, 200) + TALL = (300, 400) + FULL = (600, 400) + + +@dataclass +class WidgetConfig: + """Configuration for a widget. + + Attributes: + name: Display name of the widget + icon: Emoji or icon name + size: Widget size tuple (width, height) + resizable: Whether widget can be resized + collapsible: Whether widget can be collapsed + refresh_interval: Auto-refresh interval in seconds (0 = disabled) + settings: Widget-specific settings + """ + name: str = "Widget" + icon: str = "📦" + size: Tuple[int, int] = (300, 200) + resizable: bool = True + collapsible: bool = True + refresh_interval: float = 0.0 + settings: Dict[str, Any] = field(default_factory=dict) + category: str = "General" + description: str = "" + author: str = "" + version: str = "1.0.0" + + +# ============================================================================= +# WIDGET BASE CLASS +# ============================================================================= + +class Widget(ABC): + """Base class for dashboard widgets. + + Widgets are the UI components that plugins can create. + They are displayed in the dashboard overlay. + + Example: + class StatsWidget(Widget): + def __init__(self): + super().__init__(WidgetConfig( + name="Player Stats", + icon="👤", + size=WidgetSize.MEDIUM.value + )) + + def create_ui(self, parent): + self.widget = QWidget(parent) + layout = QVBoxLayout(self.widget) + + self.label = QLabel("Loading...") + layout.addWidget(self.label) + + return self.widget + + def update(self, data): + self.label.setText(f"Level: {data['level']}") + """ + + def __init__(self, config: WidgetConfig): + self.config = config + self._ui: Optional[Any] = None + self._visible = True + self._collapsed = False + self._data: Dict[str, Any] = {} + self._refresh_timer = None + self._callbacks: Dict[str, List[Callable]] = { + 'update': [], + 'resize': [], + 'collapse': [], + 'close': [], + } + + # ========== Abstract Methods ========== + + @abstractmethod + def create_ui(self, parent: Any) -> Any: + """Create the widget UI. + + Args: + parent: Parent widget (QWidget) + + Returns: + The created widget + """ + pass + + # ========== Lifecycle Methods ========== + + def initialize(self, parent: Any) -> Any: + """Initialize the widget with a parent.""" + self._ui = self.create_ui(parent) + + if self.config.refresh_interval > 0: + self._start_refresh_timer() + + return self._ui + + def destroy(self) -> None: + """Destroy the widget and clean up resources.""" + self._stop_refresh_timer() + self._emit('close') + + if self._ui: + self._ui.deleteLater() + self._ui = None + + # ========== Update Methods ========== + + def update(self, data: Dict[str, Any]) -> None: + """Update widget with new data. + + Override this to update your widget's display. + """ + self._data.update(data) + self._emit('update', data) + self.on_update(data) + + def on_update(self, data: Dict[str, Any]) -> None: + """Called when widget receives new data. + + Override this instead of update() for custom behavior. + """ + pass + + def refresh(self) -> None: + """Manually trigger a refresh.""" + self.on_refresh() + + def on_refresh(self) -> None: + """Called when widget should refresh its display.""" + pass + + # ========== State Methods ========== + + def show(self) -> None: + """Show the widget.""" + self._visible = True + if self._ui: + self._ui.show() + + def hide(self) -> None: + """Hide the widget.""" + self._visible = False + if self._ui: + self._ui.hide() + + def collapse(self) -> None: + """Collapse the widget.""" + self._collapsed = True + self._emit('collapse', True) + if self._ui: + self._ui.setMaximumHeight(40) + + def expand(self) -> None: + """Expand the widget.""" + self._collapsed = False + self._emit('collapse', False) + if self._ui: + self._ui.setMaximumHeight(self.config.size[1]) + + def toggle_collapse(self) -> None: + """Toggle collapsed state.""" + if self._collapsed: + self.expand() + else: + self.collapse() + + # ========== Property Methods ========== + + @property + def visible(self) -> bool: + """Whether widget is visible.""" + return self._visible + + @property + def collapsed(self) -> bool: + """Whether widget is collapsed.""" + return self._collapsed + + @property + def data(self) -> Dict[str, Any]: + """Current widget data.""" + return self._data.copy() + + @property + def ui(self) -> Optional[Any]: + """The UI widget.""" + return self._ui + + # ========== Event Handling ========== + + def on(self, event: str, callback: Callable) -> Callable: + """Subscribe to a widget event.""" + if event in self._callbacks: + self._callbacks[event].append(callback) + return callback + + def _emit(self, event: str, *args, **kwargs) -> None: + """Emit an event to subscribers.""" + for callback in self._callbacks.get(event, []): + try: + callback(*args, **kwargs) + except Exception as e: + print(f"Error in widget callback: {e}") + + # ========== Timer Methods ========== + + def _start_refresh_timer(self) -> None: + """Start auto-refresh timer.""" + if self.config.refresh_interval <= 0: + return + + try: + from PyQt6.QtCore import QTimer + self._refresh_timer = QTimer() + self._refresh_timer.timeout.connect(self.refresh) + self._refresh_timer.start(int(self.config.refresh_interval * 1000)) + except ImportError: + pass + + def _stop_refresh_timer(self) -> None: + """Stop auto-refresh timer.""" + if self._refresh_timer: + self._refresh_timer.stop() + self._refresh_timer = None + + +# ============================================================================= +# DASHBOARD +# ============================================================================= + +class Dashboard: + """Dashboard for managing widgets. + + The dashboard is the container for all plugin widgets. + It manages layout, visibility, and widget lifecycle. + """ + + def __init__(self, parent: Any = None): + self.parent = parent + self._widgets: Dict[str, Widget] = {} + self._layout: Optional[Any] = None + self._visible = False + self._position: Tuple[int, int] = (100, 100) + + def initialize(self, parent: Any) -> Any: + """Initialize the dashboard.""" + try: + from PyQt6.QtWidgets import QWidget, QVBoxLayout, QScrollArea + + self.container = QWidget(parent) + self.container.setWindowTitle("EU-Utility Dashboard") + self.container.setGeometry(*self._position, 350, 600) + + scroll = QScrollArea(self.container) + scroll.setWidgetResizable(True) + + self.widget_container = QWidget() + self._layout = QVBoxLayout(self.widget_container) + self._layout.setSpacing(10) + self._layout.addStretch() + + scroll.setWidget(self.widget_container) + + layout = QVBoxLayout(self.container) + layout.addWidget(scroll) + + return self.container + + except ImportError: + return None + + def add_widget(self, widget_id: str, widget: Widget) -> bool: + """Add a widget to the dashboard.""" + if widget_id in self._widgets: + return False + + self._widgets[widget_id] = widget + + if self._layout: + ui = widget.initialize(self.widget_container) + if ui: + # Insert before the stretch + self._layout.insertWidget(self._layout.count() - 1, ui) + + return True + + def remove_widget(self, widget_id: str) -> bool: + """Remove a widget from the dashboard.""" + widget = self._widgets.pop(widget_id, None) + if widget: + widget.destroy() + return True + return False + + def get_widget(self, widget_id: str) -> Optional[Widget]: + """Get a widget by ID.""" + return self._widgets.get(widget_id) + + def show(self) -> None: + """Show the dashboard.""" + self._visible = True + if hasattr(self, 'container') and self.container: + self.container.show() + + def hide(self) -> None: + """Hide the dashboard.""" + self._visible = False + if hasattr(self, 'container') and self.container: + self.container.hide() + + def toggle(self) -> None: + """Toggle dashboard visibility.""" + if self._visible: + self.hide() + else: + self.show() + + def get_all_widgets(self) -> Dict[str, Widget]: + """Get all widgets.""" + return self._widgets.copy() + + +# ============================================================================= +# EXPORTS +# ============================================================================= + +__all__ = [ + 'WidgetSize', + 'WidgetConfig', + 'Widget', + 'Dashboard', +] diff --git a/premium/widgets/base.py b/premium/widgets/base.py new file mode 100644 index 0000000..bd75559 --- /dev/null +++ b/premium/widgets/base.py @@ -0,0 +1,20 @@ +""" +EU-Utility Premium - Base Widget Components +============================================ + +Base widget classes and components for building custom widgets. +""" + +from .dashboard_widget import DashboardWidget, WidgetHeader, WidgetContent +from .metrics_card import MetricsCard, StatItem +from .chart_widget import ChartWidget, ChartType + +__all__ = [ + 'DashboardWidget', + 'WidgetHeader', + 'WidgetContent', + 'MetricsCard', + 'StatItem', + 'ChartWidget', + 'ChartType', +] diff --git a/premium/widgets/chart_widget.py b/premium/widgets/chart_widget.py new file mode 100644 index 0000000..618fd81 --- /dev/null +++ b/premium/widgets/chart_widget.py @@ -0,0 +1,145 @@ +""" +Chart widget for displaying data visualizations. +""" + +from enum import Enum +from typing import List, Dict, Any, Optional + +try: + from PyQt6.QtWidgets import QWidget, QVBoxLayout + from PyQt6.QtCore import Qt + from PyQt6.QtGui import QPainter, QColor, QPen, QBrush, QFont + HAS_QT = True +except ImportError: + HAS_QT = False + QWidget = object + + +class ChartType(Enum): + LINE = "line" + BAR = "bar" + PIE = "pie" + + +class ChartWidget(QWidget if HAS_QT else object): + """Simple chart widget.""" + + def __init__( + self, + chart_type: ChartType = ChartType.LINE, + title: str = "Chart", + parent=None + ): + if not HAS_QT: + return + super().__init__(parent) + self.chart_type = chart_type + self.title = title + self._data: List[Dict[str, Any]] = [] + self._max_points = 50 + self._y_max = 100 + + self.setMinimumHeight(150) + self.setStyleSheet("background-color: #2d2d2d; border-radius: 8px;") + + def add_point(self, x: Any, y: float, label: Optional[str] = None): + """Add a data point.""" + self._data.append({'x': x, 'y': y, 'label': label}) + + # Limit points + if len(self._data) > self._max_points: + self._data = self._data[-self._max_points:] + + # Update scale + if y > self._y_max: + self._y_max = y * 1.1 + + if HAS_QT: + self.update() + + def set_data(self, data: List[Dict[str, Any]]): + """Set all data points.""" + self._data = data + if data: + self._y_max = max(d['y'] for d in data) * 1.1 + if HAS_QT: + self.update() + + def clear(self): + """Clear all data.""" + self._data = [] + if HAS_QT: + self.update() + + def paintEvent(self, event): + if not HAS_QT or not self._data: + return + + painter = QPainter(self) + painter.setRenderHint(QPainter.RenderHint.Antialiasing) + + width = self.width() + height = self.height() + padding = 40 + + chart_width = width - padding * 2 + chart_height = height - padding * 2 + + # Background + painter.fillRect(self.rect(), QColor(45, 45, 45)) + + # Draw title + painter.setPen(QColor(255, 255, 255)) + painter.setFont(QFont("Arial", 10, QFont.Weight.Bold)) + painter.drawText(10, 20, self.title) + + if len(self._data) < 2: + return + + # Draw chart based on type + if self.chart_type == ChartType.LINE: + self._draw_line_chart(painter, padding, chart_width, chart_height) + elif self.chart_type == ChartType.BAR: + self._draw_bar_chart(painter, padding, chart_width, chart_height) + + def _draw_line_chart(self, painter, padding, width, height): + """Draw a line chart.""" + if len(self._data) < 2: + return + + pen = QPen(QColor(76, 175, 80)) + pen.setWidth(2) + painter.setPen(pen) + + x_step = width / max(len(self._data) - 1, 1) + + # Draw line + points = [] + for i, point in enumerate(self._data): + x = padding + i * x_step + y = padding + height - (point['y'] / self._y_max * height) + points.append((x, y)) + + for i in range(len(points) - 1): + painter.drawLine(int(points[i][0]), int(points[i][1]), + int(points[i+1][0]), int(points[i+1][1])) + + # Draw points + painter.setBrush(QColor(76, 175, 80)) + for x, y in points: + painter.drawEllipse(int(x - 3), int(y - 3), 6, 6) + + def _draw_bar_chart(self, painter, padding, width, height): + """Draw a bar chart.""" + bar_width = width / len(self._data) * 0.8 + gap = width / len(self._data) * 0.2 + + painter.setBrush(QColor(76, 175, 80)) + painter.setPen(Qt.PenStyle.NoPen) + + for i, point in enumerate(self._data): + bar_height = point['y'] / self._y_max * height + x = padding + i * (bar_width + gap) + gap / 2 + y = padding + height - bar_height + + painter.drawRect(int(x), int(y), int(bar_width), int(bar_height)) diff --git a/premium/widgets/dashboard_widget.py b/premium/widgets/dashboard_widget.py new file mode 100644 index 0000000..b271811 --- /dev/null +++ b/premium/widgets/dashboard_widget.py @@ -0,0 +1,194 @@ +""" +Base widget components for the dashboard. +""" + +from typing import Any, Optional, Callable + +try: + from PyQt6.QtWidgets import ( + QWidget, QVBoxLayout, QHBoxLayout, QLabel, QPushButton, + QFrame, QSizePolicy + ) + from PyQt6.QtCore import Qt, pyqtSignal + from PyQt6.QtGui import QFont + HAS_QT = True +except ImportError: + HAS_QT = False + QWidget = object + pyqtSignal = lambda *a, **k: None + + +class WidgetHeader(QFrame if HAS_QT else object): + """Header component for widgets with title and controls.""" + + collapsed_changed = pyqtSignal(bool) if HAS_QT else None + close_requested = pyqtSignal() if HAS_QT else None + + def __init__(self, title: str = "Widget", icon: str = "📦", parent=None): + if not HAS_QT: + return + super().__init__(parent) + self._collapsed = False + self._setup_ui(title, icon) + + def _setup_ui(self, title: str, icon: str): + self.setFrameShape(QFrame.Shape.StyledPanel) + self.setStyleSheet(""" + WidgetHeader { + background-color: #2d2d2d; + border-radius: 8px 8px 0 0; + padding: 8px; + } + """) + + layout = QHBoxLayout(self) + layout.setContentsMargins(10, 5, 10, 5) + layout.setSpacing(10) + + # Icon + self.icon_label = QLabel(icon) + self.icon_label.setStyleSheet("font-size: 16px;") + layout.addWidget(self.icon_label) + + # Title + self.title_label = QLabel(title) + font = QFont() + font.setBold(True) + self.title_label.setFont(font) + self.title_label.setStyleSheet("color: #ffffff;") + layout.addWidget(self.title_label) + + layout.addStretch() + + # Collapse button + self.collapse_btn = QPushButton("▼") + self.collapse_btn.setFixedSize(24, 24) + self.collapse_btn.setStyleSheet(""" + QPushButton { + background-color: #3d3d3d; + color: #ffffff; + border: none; + border-radius: 4px; + } + QPushButton:hover { + background-color: #4d4d4d; + } + """) + self.collapse_btn.clicked.connect(self._toggle_collapse) + layout.addWidget(self.collapse_btn) + + # Close button + self.close_btn = QPushButton("×") + self.close_btn.setFixedSize(24, 24) + self.close_btn.setStyleSheet(""" + QPushButton { + background-color: #3d3d3d; + color: #ff6b6b; + border: none; + border-radius: 4px; + font-size: 16px; + font-weight: bold; + } + QPushButton:hover { + background-color: #ff6b6b; + color: #ffffff; + } + """) + self.close_btn.clicked.connect(lambda: self.close_requested.emit()) + layout.addWidget(self.close_btn) + + def _toggle_collapse(self): + self._collapsed = not self._collapsed + self.collapse_btn.setText("▶" if self._collapsed else "▼") + self.collapsed_changed.emit(self._collapsed) + + def set_collapsed(self, collapsed: bool): + self._collapsed = collapsed + self.collapse_btn.setText("▶" if collapsed else "▼") + + +class WidgetContent(QFrame if HAS_QT else object): + """Content container for widgets.""" + + def __init__(self, parent=None): + if not HAS_QT: + return + super().__init__(parent) + self.setFrameShape(QFrame.Shape.StyledPanel) + self.setStyleSheet(""" + WidgetContent { + background-color: #1e1e1e; + border-radius: 0 0 8px 8px; + padding: 10px; + } + """) + + self._layout = QVBoxLayout(self) + self._layout.setContentsMargins(10, 10, 10, 10) + self._layout.setSpacing(10) + + def add_widget(self, widget): + if HAS_QT: + self._layout.addWidget(widget) + + def layout(self): + return self._layout if HAS_QT else None + + +class DashboardWidget(QFrame if HAS_QT else object): + """Complete dashboard widget with header and content.""" + + def __init__( + self, + title: str = "Widget", + icon: str = "📦", + size: tuple = (300, 200), + parent=None + ): + if not HAS_QT: + return + super().__init__(parent) + self._size = size + self._setup_ui(title, icon) + + def _setup_ui(self, title: str, icon: str): + self.setStyleSheet(""" + DashboardWidget { + background-color: transparent; + border: 1px solid #3d3d3d; + border-radius: 8px; + } + """) + + layout = QVBoxLayout(self) + layout.setContentsMargins(0, 0, 0, 0) + layout.setSpacing(0) + + # Header + self.header = WidgetHeader(title, icon, self) + self.header.collapsed_changed.connect(self._on_collapse_changed) + layout.addWidget(self.header) + + # Content + self.content = WidgetContent(self) + layout.addWidget(self.content) + + self.setFixedSize(*self._size) + + def _on_collapse_changed(self, collapsed: bool): + self.content.setVisible(not collapsed) + if collapsed: + self.setFixedHeight(self.header.height()) + else: + self.setFixedHeight(self._size[1]) + + def set_content_widget(self, widget: QWidget): + """Set the main content widget.""" + if HAS_QT: + # Clear existing + while self.content.layout().count(): + item = self.content.layout().takeAt(0) + if item.widget(): + item.widget().deleteLater() + + self.content.layout().addWidget(widget) diff --git a/premium/widgets/metrics_card.py b/premium/widgets/metrics_card.py new file mode 100644 index 0000000..7e5325a --- /dev/null +++ b/premium/widgets/metrics_card.py @@ -0,0 +1,121 @@ +""" +Metrics card widget for displaying statistics. +""" + +from typing import List, Optional +from dataclasses import dataclass + +try: + from PyQt6.QtWidgets import ( + QWidget, QVBoxLayout, QHBoxLayout, QLabel, QGridLayout + ) + from PyQt6.QtCore import Qt + from PyQt6.QtGui import QFont + HAS_QT = True +except ImportError: + HAS_QT = False + QWidget = object + + +@dataclass +class StatItem: + """Single statistic item.""" + label: str + value: str + change: Optional[str] = None + positive: bool = True + + +class MetricsCard(QWidget if HAS_QT else object): + """Card displaying multiple metrics.""" + + def __init__(self, title: str = "Metrics", parent=None): + if not HAS_QT: + return + super().__init__(parent) + self._stats: List[StatItem] = [] + self._setup_ui(title) + + def _setup_ui(self, title: str): + self.setStyleSheet(""" + MetricsCard { + background-color: #2d2d2d; + border-radius: 8px; + padding: 15px; + } + QLabel { + color: #ffffff; + } + .metric-label { + color: #888888; + font-size: 12px; + } + .metric-value { + font-size: 18px; + font-weight: bold; + } + .metric-positive { + color: #4caf50; + } + .metric-negative { + color: #f44336; + } + """) + + layout = QVBoxLayout(self) + layout.setContentsMargins(15, 15, 15, 15) + layout.setSpacing(15) + + # Title + self.title_label = QLabel(title) + font = QFont() + font.setBold(True) + font.setPointSize(14) + self.title_label.setFont(font) + layout.addWidget(self.title_label) + + # Stats grid + self.stats_layout = QGridLayout() + self.stats_layout.setSpacing(10) + layout.addLayout(self.stats_layout) + + layout.addStretch() + + def set_stats(self, stats: List[StatItem]): + """Update displayed statistics.""" + if not HAS_QT: + return + + self._stats = stats + + # Clear existing + while self.stats_layout.count(): + item = self.stats_layout.takeAt(0) + if item.widget(): + item.widget().deleteLater() + + # Add stats + for i, stat in enumerate(stats): + row = i // 2 + col = (i % 2) * 2 + + # Label + label = QLabel(stat.label) + label.setStyleSheet("color: #888888; font-size: 11px;") + self.stats_layout.addWidget(label, row * 2, col) + + # Value with optional change + value_text = stat.value + if stat.change: + value_text += f" {stat.change}" + + value_label = QLabel(value_text) + value_label.setStyleSheet("font-size: 16px; font-weight: bold; color: #ffffff;") + self.stats_layout.addWidget(value_label, row * 2 + 1, col) + + def update_stat(self, index: int, value: str, change: Optional[str] = None): + """Update a single statistic.""" + if 0 <= index < len(self._stats): + self._stats[index].value = value + self._stats[index].change = change + self.set_stats(self._stats)