impulsivefps
/
EU-Utility
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Auto-sync: 2026-02-16 22:34

This commit is contained in:
devmatrix 2026-02-16 22:34:51 +00:00
parent d1510dfff8
commit 8e38c6cb12
16 changed files with 3664 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

297
main.py Normal file
View File

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

322
plugins/builtin/dashboard_widget/main.py Normal file
View File

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

12
plugins/builtin/dashboard_widget/plugin.json Normal file
View File

@ -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": "📊"
}

173
premium/__init__.py Normal file
View File

@ -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',
]

599
premium/core/event_bus.py Normal file
View File

@ -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',
]

33
premium/eu_integration/__init__.py Normal file
View File

@ -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',
]

131
premium/eu_integration/events.py Normal file
View File

@ -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'),
)

344
premium/eu_integration/game_client.py Normal file
View File

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

323
premium/eu_integration/log_parser.py Normal file
View File

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

107
premium/eu_integration/window_tracker.py Normal file
View File

@ -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}")

465
premium/plugins/api.py Normal file
View File

@ -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',
]

378
premium/widgets/__init__.py Normal file
View File

@ -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',
]

20
premium/widgets/base.py Normal file
View File

@ -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',
]

145
premium/widgets/chart_widget.py Normal file
View File

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

194
premium/widgets/dashboard_widget.py Normal file
View File

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

121
premium/widgets/metrics_card.py Normal file
View File

@ -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" <span style='color: {'#4caf50' if stat.positive else '#f44336'}'>{stat.change}</span>"
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)