18 KiB
18 KiB
EU-Utility Application Redesign Draft
Version 3.0 - Professional Architecture
🎯 Executive Summary
Current EU-Utility is feature-rich but architecture is fragmented. This redesign creates a unified, scalable, maintainable application.
Key Improvements:
- Clean Architecture (Domain-Driven Design)
- Unified API Gateway
- Reactive Event System
- Plugin Sandbox Security
- Cloud Sync Capabilities
🏗️ New Architecture
┌─────────────────────────────────────────────────────────────┐
│ PRESENTATION LAYER │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────────┐ │
│ │ Main UI │ │ Overlay │ │ System Tray │ │
│ │ (PyQt6) │ │ Manager │ │ Icon │ │
│ └──────┬──────┘ └──────┬──────┘ └──────────┬──────────┘ │
└─────────┼────────────────┼────────────────────┼─────────────┘
│ │ │
└────────────────┴────────────────────┘
│
┌────────▼────────┐
│ API GATEWAY │
│ (FastAPI/REST) │
└────────┬────────┘
│
┌────────────────────┼────────────────────┐
│ │ │
┌───────▼──────┐ ┌─────────▼─────────┐ ┌──────▼──────┐
│ CORE │ │ PLUGIN ENGINE │ │ SERVICES │
│ SERVICES │ │ │ │ │
│ │ │ ┌─────────────┐ │ │ • Nexus │
│ • Settings │ │ │ Plugin │ │ │ • OCR │
│ • DataStore │ │ │ Manager │ │ │ • Overlay │
│ • EventBus │ │ └──────┬──────┘ │ │ • Capture │
│ • Auth │ │ │ │ └─────────────┘
└──────────────┘ │ ┌──────▼──────┐ │
│ │ Sandboxed │ │
│ │ Plugins │ │
│ └─────────────┘ │
└───────────────────┘
📦 Domain Modules
1. Domain: Core
# domain/core/entities.py
from dataclasses import dataclass
from datetime import datetime
from typing import Optional, Dict, Any
from enum import Enum
class PluginState(Enum):
INSTALLED = "installed"
ACTIVE = "active"
ERROR = "error"
DISABLED = "disabled"
@dataclass
class Plugin:
"""Core plugin entity."""
id: str
name: str
version: str
author: str
description: str
state: PluginState
hotkey: Optional[str] = None
config: Dict[str, Any] = None
created_at: datetime = None
updated_at: datetime = None
@dataclass
class UserSettings:
"""User configuration entity."""
user_id: str
hotkeys: Dict[str, str]
theme: str
plugins_enabled: list
overlay_position: tuple
ocr_engine: str
api_keys: Dict[str, str]
2. Domain: Game
# domain/game/entities.py
from dataclasses import dataclass
from datetime import datetime
from typing import List, Optional
from decimal import Decimal
@dataclass
class LootItem:
"""Represents a looted item."""
name: str
quantity: int
tt_value: Decimal
market_value: Optional[Decimal] = None
@dataclass
class LootEvent:
"""Game loot event."""
id: str
timestamp: datetime
mob_name: str
items: List[LootItem]
total_tt: Decimal
location: Optional[tuple] = None
@dataclass
class SkillGain:
"""Skill increase event."""
skill_name: str
value_before: float
value_after: float
timestamp: datetime
3. Domain: Market
# domain/market/entities.py
from dataclasses import dataclass
from decimal import Decimal
from datetime import datetime
from typing import Optional
@dataclass
class MarketItem:
"""Market data for an item."""
item_id: str
name: str
category: str
current_price: Decimal
markup_percent: Decimal
volume_24h: int
last_update: datetime
@dataclass
class PriceAlert:
"""User price alert."""
item_id: str
user_id: str
target_price: Decimal
condition: str # 'above' or 'below'
is_active: bool
🔌 Plugin System Redesign
Plugin Manifest (v3.0)
{
"manifest_version": "3.0",
"plugin": {
"id": "com.example.loot_tracker",
"name": "Advanced Loot Tracker",
"version": "2.1.0",
"author": "ImpulsiveFPS",
"description": "Professional loot tracking with analytics",
"category": "tracking",
"min_api_version": "3.0"
},
"permissions": [
"log_reader:read",
"overlay:create",
"data_store:write",
"notification:send",
"ocr:read"
],
"resources": {
"memory_limit": "128MB",
"cpu_limit": "10%",
"storage_limit": "50MB"
},
"ui": {
"has_overlay": true,
"has_settings": true,
"hotkey": "ctrl+shift+l",
"icon": "treasure-chest"
},
"dependencies": {
"core": ">=3.0",
"plugins": []
}
}
Plugin Base Class
# application/plugin_base.py
from abc import ABC, abstractmethod
from typing import Dict, Any, Optional
from dataclasses import dataclass
import asyncio
@dataclass
class PluginContext:
"""Context provided to plugins."""
api: 'PluginAPI'
settings: Dict[str, Any]
storage: 'PluginStorage'
logger: 'PluginLogger'
event_bus: 'EventBus'
class BasePlugin(ABC):
"""Base class for all EU-Utility plugins."""
# Required - defined in manifest
id: str
name: str
version: str
# Optional - defined in manifest
description: str = ""
author: str = ""
category: str = "utility"
def __init__(self, context: PluginContext):
self.ctx = context
self._state = PluginState.INSTALLED
self._config: Dict[str, Any] = {}
@abstractmethod
async def initialize(self) -> bool:
"""Initialize the plugin. Return True on success."""
pass
@abstractmethod
async def activate(self) -> bool:
"""Called when plugin is activated."""
pass
@abstractmethod
async def deactivate(self) -> None:
"""Called when plugin is deactivated."""
pass
@abstractmethod
def get_ui(self) -> Optional['QWidget']:
"""Return the plugin's UI component."""
pass
async def on_event(self, event: 'Event') -> None:
"""Handle events from the event bus."""
pass
def get_config_schema(self) -> Dict[str, Any]:
"""Return JSON schema for plugin configuration."""
return {}
🔐 Security Architecture
Permission System
# security/permissions.py
from enum import Enum, auto
class PermissionScope(Enum):
"""Permission scopes for plugins."""
LOG_READER = auto()
OVERLAY = auto()
DATA_STORE = auto()
NETWORK = auto()
OCR = auto()
CLIPBOARD = auto()
FILE_SYSTEM = auto()
PROCESS = auto()
class PermissionLevel(Enum):
"""Permission levels."""
NONE = 0
READ = 1
WRITE = 2
ADMIN = 3
@dataclass
class Permission:
scope: PermissionScope
level: PermissionLevel
constraints: Dict[str, Any] = None
Plugin Sandbox
# security/sandbox.py
import subprocess
import tempfile
from pathlib import Path
class PluginSandbox:
"""Isolated execution environment for plugins."""
def __init__(self, plugin_id: str, permissions: List[Permission]):
self.plugin_id = plugin_id
self.permissions = permissions
self._temp_dir = tempfile.mkdtemp(prefix=f"eu_plugin_{plugin_id}_")
def execute(self, code: str) -> Any:
"""Execute plugin code in restricted environment."""
# Implementation uses restricted Python interpreter
# with limited builtins and custom __import__
pass
def cleanup(self) -> None:
"""Clean up sandbox resources."""
shutil.rmtree(self._temp_dir, ignore_errors=True)
🔄 Event System
Reactive Event Bus
# infrastructure/event_bus.py
from typing import Callable, TypeVar, Generic
import asyncio
from dataclasses import dataclass
from datetime import datetime
T = TypeVar('T')
@dataclass
class Event(Generic[T]):
"""Generic event container."""
type: str
payload: T
timestamp: datetime
source: str
class ReactiveEventBus:
"""Reactive event system with filtering and transformation."""
def __init__(self):
self._subscribers: Dict[str, List[Callable]] = {}
self._streams: Dict[str, 'EventStream'] = {}
def subscribe(self,
event_type: str,
handler: Callable[[Event], None],
filter_fn: Optional[Callable[[Event], bool]] = None) -> str:
"""Subscribe to events with optional filtering."""
subscription_id = str(uuid.uuid4())
# Implementation
return subscription_id
def publish(self, event: Event) -> None:
"""Publish event to all subscribers."""
# Async dispatch
asyncio.create_task(self._dispatch(event))
def stream(self, event_type: str) -> 'EventStream':
"""Get reactive stream for event type."""
if event_type not in self._streams:
self._streams[event_type] = EventStream(event_type)
return self._streams[event_type]
class EventStream:
"""Reactive stream for events."""
def __init__(self, event_type: str):
self.event_type = event_type
self._handlers = []
def filter(self, predicate: Callable[[Event], bool]) -> 'EventStream':
"""Filter events."""
return self
def map(self, transform: Callable[[Event], T]) -> 'EventStream':
"""Transform events."""
return self
def debounce(self, seconds: float) -> 'EventStream':
"""Debounce events."""
return self
def subscribe(self, handler: Callable[[T], None]) -> None:
"""Subscribe to stream."""
pass
📊 Data Layer
Repository Pattern
# infrastructure/repositories.py
from abc import ABC, abstractmethod
from typing import List, Optional, Generic, TypeVar
T = TypeVar('T')
ID = TypeVar('ID')
class Repository(ABC, Generic[T, ID]):
"""Base repository interface."""
@abstractmethod
def get(self, id: ID) -> Optional[T]:
pass
@abstractmethod
def get_all(self) -> List[T]:
pass
@abstractmethod
def create(self, entity: T) -> T:
pass
@abstractmethod
def update(self, entity: T) -> T:
pass
@abstractmethod
def delete(self, id: ID) -> bool:
pass
class SQLiteRepository(Repository[T, ID]):
"""SQLite implementation."""
def __init__(self, db_path: str, entity_class: type):
self.db_path = db_path
self.entity_class = entity_class
self._init_db()
Data Sync
# infrastructure/sync.py
from typing import Dict, Any
import aiohttp
class CloudSync:
"""Sync data with cloud backend."""
def __init__(self, api_base: str, api_key: str):
self.api_base = api_base
self.api_key = api_key
self._local_queue = []
async def sync_settings(self, settings: Dict[str, Any]) -> bool:
"""Sync user settings to cloud."""
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.api_base}/settings",
json=settings,
headers={"Authorization": f"Bearer {self.api_key}"}
) as resp:
return resp.status == 200
async def get_remote_settings(self) -> Dict[str, Any]:
"""Get settings from cloud."""
pass
🚀 API Gateway
RESTful API
# api/gateway.py
from fastapi import FastAPI, Depends, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from typing import List, Optional
app = FastAPI(
title="EU-Utility API",
version="3.0.0",
description="Unified API for EU-Utility"
)
# Authentication dependency
async def verify_token(token: str = Header(...)):
"""Verify API token."""
# Implementation
pass
@app.get("/plugins", response_model=List[PluginSchema])
async def list_plugins(
category: Optional[str] = None,
active_only: bool = True,
auth = Depends(verify_token)
):
"""List all plugins."""
pass
@app.post("/plugins/{plugin_id}/activate")
async def activate_plugin(
plugin_id: str,
auth = Depends(verify_token)
):
"""Activate a plugin."""
pass
@app.get("/market/search")
async def search_market(
query: str,
category: Optional[str] = None,
limit: int = 20
):
"""Search market items."""
pass
@app.get("/game/loot/recent")
async def get_recent_loot(
hours: int = 24,
auth = Depends(verify_token)
):
"""Get recent loot events."""
pass
@app.post("/ocr/recognize")
async def ocr_recognize(
region: tuple, # (x, y, width, height)
auth = Depends(verify_token)
):
"""Recognize text from screen region."""
pass
📱 UI/UX Design
Component Library
# ui/components.py
from PyQt6.QtWidgets import *
from PyQt6.QtCore import *
from PyQt6.QtGui import *
class EUCard(QFrame):
"""Card component for EU-Utility."""
def __init__(self, title: str, parent=None):
super().__init__(parent)
self.setObjectName("EUCard")
self.setup_ui(title)
def setup_ui(self, title: str):
layout = QVBoxLayout(self)
layout.setContentsMargins(16, 16, 16, 16)
# Header
header = QLabel(title)
header.setObjectName("card-title")
layout.addWidget(header)
# Content area
self.content = QFrame()
self.content_layout = QVBoxLayout(self.content)
layout.addWidget(self.content)
# Styling
self.setStyleSheet("""
#EUCard {
background-color: #2d2d2d;
border-radius: 12px;
border: 1px solid #3d3d3d;
}
#card-title {
color: #ffffff;
font-size: 16px;
font-weight: bold;
margin-bottom: 12px;
}
""")
class EUButton(QPushButton):
"""Primary button component."""
def __init__(self, text: str, variant: str = "primary"):
super().__init__(text)
self.variant = variant
self.setup_style()
def setup_style(self):
colors = {
"primary": ("#6366f1", "#4f46e5"),
"secondary": ("#3d3d3d", "#4d4d4d"),
"danger": ("#ef4444", "#dc2626"),
"success": ("#10b981", "#059669")
}
bg, hover = colors.get(self.variant, colors["primary"])
self.setStyleSheet(f"""
EUButton {{
background-color: {bg};
color: white;
border: none;
border-radius: 8px;
padding: 10px 20px;
font-weight: 600;
}}
EUButton:hover {{
background-color: {hover};
}}
""")
📋 Implementation Roadmap
Phase 1: Foundation (Week 1-2)
- Set up new project structure
- Implement domain entities
- Create repository layer
- Set up dependency injection
Phase 2: Core Services (Week 3-4)
- Event bus implementation
- Plugin system v3.0
- Security sandbox
- Settings management
Phase 3: API & UI (Week 5-6)
- API Gateway
- Component library
- Main UI refactoring
- Overlay system
Phase 4: Integration (Week 7-8)
- Plugin migration tool
- Data migration
- Testing suite
- Documentation
Phase 5: Release (Week 9)
- Beta testing
- Performance optimization
- Security audit
- Public release
📊 Migration Strategy
From v2.x to v3.0
# migration/v2_to_v3.py
class MigrationTool:
"""Migrate plugins and data from v2 to v3."""
def migrate_plugin(self, v2_plugin_path: str) -> str:
"""Convert v2 plugin to v3 format."""
# Read v2 plugin
# Generate v3 manifest
# Update imports
# Return new plugin path
pass
def migrate_settings(self, v2_settings: dict) -> dict:
"""Migrate user settings."""
mapping = {
"hotkeys.global_toggle": "hotkeys.overlay_toggle",
"plugins.enabled": "plugins.active",
# ...
}
# Apply mapping
return migrated_settings
🎯 Success Metrics
| Metric | Target |
|---|---|
| Startup Time | < 2 seconds |
| Memory Usage | < 200MB |
| Plugin Load | < 100ms each |
| API Response | < 50ms |
| Code Coverage | > 80% |
| Plugin Compatibility | 100% v2 plugins work |
Draft Application Design - EU-Utility v3.0 Created: 2026-02-23