EU-Utility/docs/API_REFERENCE.md

EU-Utility API Reference

Complete API reference for EU-Utility core services and plugin development

Version: 2.0
Last Updated: 2025-02-14

---

Table of Contents

1. Core Services Overview
2. PluginAPI
3. BasePlugin
4. Event Bus
5. Task Manager
6. Nexus API
7. Settings
8. Data Store
9. Audio Manager
10. HTTP Client
11. OCR Service
12. Screenshot Service
13. Log Reader
14. Window Manager
15. Notification Manager
16. Clipboard Manager

---

Core Services Overview

EU-Utility provides a comprehensive set of core services accessible through the PluginAPI:

┌─────────────────────────────────────────────────────────────┐
│                      PluginAPI (Singleton)                   │
├─────────────────────────────────────────────────────────────┤
│  OCR Service    │  Screenshot  │  Log Reader  │  Nexus API  │
├─────────────────────────────────────────────────────────────┤
│  Task Manager   │  Event Bus   │  Audio       │  HTTP Client│
├─────────────────────────────────────────────────────────────┤
│  Window Manager │  Data Store  │  Clipboard   │  Settings   │
└─────────────────────────────────────────────────────────────┘

Accessing Services

from core.plugin_api import get_api

api = get_api()

# Use services
result = api.ocr_capture()

---

PluginAPI

Class: PluginAPI

Central API registry and shared services.

Singleton Access

from core.plugin_api import get_api

api = get_api()

API Registration

from core.plugin_api import APIEndpoint, APIType

# Register an API endpoint
endpoint = APIEndpoint(
    name="scan_window",
    api_type=APIType.OCR,
    description="Scan game window and return text",
    handler=self.scan_window,
    plugin_id=self._plugin_id,
    version="1.0.0"
)

success = api.register_api(endpoint)

# Unregister
api.unregister_api(plugin_id, name)

Calling APIs

# Call another plugin's API
result = api.call_api("plugin_id", "api_name", arg1, arg2)

# Find available APIs
ocr_apis = api.find_apis(APIType.OCR)

OCR Service

# Capture and OCR
result = api.ocr_capture(region=(x, y, width, height))
# Returns: {'text': str, 'confidence': float, 'raw_results': list}

Screenshot Service

# Capture screen
image = api.capture_screen(full_screen=True)

# Capture region
image = api.capture_region(x, y, width, height)

# Get last screenshot
image = api.get_last_screenshot()

# Save screenshot
path = api.save_screenshot(image, filename="screenshot.png")

Log Service

# Read log lines
lines = api.read_log(lines=50, filter_text="loot")
# Returns: List[str]

Shared Data

# Set data
api.set_data("key", value)

# Get data
value = api.get_data("key", default=None)

Audio Service

# Play sounds
api.play_sound("hof")                    # Predefined
api.play_sound("/path/to/custom.wav")    # Custom file
api.play_sound("alert", blocking=True)   # Wait for completion

# Volume control
api.set_volume(0.8)           # 0.0 to 1.0
volume = api.get_volume()     # Get current volume

# Mute
api.mute_audio()
api.unmute_audio()
is_muted = api.is_audio_muted()

# Check availability
if api.is_audio_available():
    api.play_sound("skill_gain")

Background Tasks

# Run in background
task_id = api.run_in_background(
    func=my_function,
    arg1, arg2,
    priority='normal',        # 'high', 'normal', 'low'
    on_complete=on_done,      # Callback with result
    on_error=on_error         # Callback with exception
)

# Schedule task
task_id = api.schedule_task(
    delay_ms=5000,            # 5 seconds
    func=my_function,
    priority='normal',
    on_complete=on_done
)

# Periodic task
task_id = api.schedule_task(
    delay_ms=0,
    func=refresh_data,
    periodic=True,
    interval_ms=30000,        # Every 30 seconds
    on_complete=update_ui
)

# Cancel task
api.cancel_task(task_id)

# Get status
status = api.get_task_status(task_id)  # 'pending', 'running', 'completed', 'failed', 'cancelled'

# Wait for completion
completed = api.wait_for_task(task_id, timeout=10.0)

# Connect signals
api.connect_task_signal('completed', on_task_complete)
api.connect_task_signal('failed', on_task_error)

Nexus API

# Search
results = api.nexus_search("ArMatrix", entity_type="items", limit=20)
results = api.nexus_search("Atrox", entity_type="mobs")

# Get details
details = api.nexus_get_item_details("armatrix_lp-35")

# Get market data
market = api.nexus_get_market_data("armatrix_lp-35")

# Check availability
if api.nexus_is_available():
    results = api.nexus_search("query")

Event Bus (Typed)

from core.event_bus import SkillGainEvent, LootEvent

# Publish event
api.publish_typed(SkillGainEvent(
    skill_name="Rifle",
    skill_value=25.5,
    gain_amount=0.01
))

# Publish synchronously
count = api.publish_typed_sync(event)

# Subscribe
sub_id = api.subscribe_typed(
    SkillGainEvent,
    callback_function,
    min_damage=100,           # Filter options
    replay_last=10            # Replay recent events
)

# Unsubscribe
api.unsubscribe_typed(sub_id)

# Get recent events
recent = api.get_recent_events(SkillGainEvent, count=50)

# Get stats
stats = api.get_event_stats()

Utility Methods

# Currency formatting
api.format_ped(123.45)       # "123.45 PED"
api.format_pec(50)           # "50 PEC"

# Calculations
api.calculate_dpp(damage=45.0, ammo=100, decay=2.5)
api.calculate_markup(price=150.0, tt=100.0)  # 150.0%

---

BasePlugin

Class: BasePlugin

Abstract base class for all plugins.

Attributes

Attribute	Type	Required	Description
name	str	Yes	Display name
version	str	Yes	Version string
author	str	Yes	Creator name
description	str	Yes	Short description
hotkey	str	No	Global hotkey
enabled	bool	No	Start enabled
icon	str	No	Icon path/emoji

Constructor

def __init__(self, overlay_window: OverlayWindow, config: Dict[str, Any])

Abstract Methods

def initialize(self) -> None:
    """Called when plugin is loaded. Setup API connections."""
    pass

def get_ui(self) -> Any:
    """Return the plugin's UI widget (QWidget)."""
    return None

Lifecycle Methods

def on_show(self) -> None:
    """Called when overlay becomes visible."""
    pass

def on_hide(self) -> None:
    """Called when overlay is hidden."""
    pass

def on_hotkey(self) -> None:
    """Called when plugin's hotkey is pressed."""
    pass

def shutdown(self) -> None:
    """Called when app is closing. Cleanup resources."""
    # Unregister APIs
    if self.api and self._api_registered:
        self.api.unregister_api(self._plugin_id)
    
    # Unsubscribe from events
    self.unsubscribe_all_typed()

Configuration Methods

# Get config with default
value = self.get_config('key', default_value)

# Set config
self.set_config('key', value)

API Methods

# Register API endpoint
self.register_api(
    name="scan_window",
    handler=self.scan_window,
    api_type=APIType.OCR,
    description="Scan game window"
)

# Call another plugin's API
result = self.call_api("plugin_id", "api_name", *args, **kwargs)

# Find APIs
apis = self.find_apis(APIType.OCR)

Service Methods

# OCR
result = self.ocr_capture(region=(x, y, w, h))

# Screenshots
image = self.capture_screen(full_screen=True)
image = self.capture_region(x, y, w, h)
last = self.get_last_screenshot()

# Log
lines = self.read_log(lines=50, filter_text="keyword")

# Shared data
self.set_shared_data('key', value)
data = self.get_shared_data('key', default)

Event Methods

# Legacy events
self.publish_event('event_type', {'key': 'value'})
self.subscribe('event_type', callback)

# Typed events
self.publish_typed(SkillGainEvent(...))

sub_id = self.subscribe_typed(
    SkillGainEvent,
    callback,
    min_damage=100,
    replay_last=10
)

self.unsubscribe_typed(sub_id)
self.unsubscribe_all_typed()

# Get recent events
recent = self.get_recent_events(SkillGainEvent, 50)
stats = self.get_event_stats()

Utility Methods

# Currency
self.format_ped(123.45)      # "123.45 PED"
self.format_pec(50)          # "50 PEC"

# Calculations
self.calculate_dpp(damage, ammo, decay)
self.calculate_markup(price, tt)

Audio Methods

self.play_sound('hof')
self.set_volume(0.8)
self.get_volume()
self.mute()
self.unmute()
self.toggle_mute()
self.is_muted()
self.is_audio_available()

Background Task Methods

# Run in background
task_id = self.run_in_background(
    func, *args,
    priority='normal',
    on_complete=callback,
    on_error=error_callback,
    **kwargs
)

# Schedule
task_id = self.schedule_task(
    delay_ms, func, *args,
    priority='normal',
    on_complete=callback,
    periodic=False,
    interval_ms=None,
    **kwargs
)

# Control
self.cancel_task(task_id)
self.connect_task_signals(
    on_completed=callback,
    on_failed=error_callback,
    on_started=callback,
    on_cancelled=callback
)

Nexus API Methods

results = self.nexus_search(query, entity_type="items", limit=20)
details = self.nexus_get_item_details(item_id)
market = self.nexus_get_market_data(item_id)
available = self.nexus_is_available()

---

Event Bus

Class: EventBus

Typed event system with filtering and persistence.

Singleton Access

from core.event_bus import get_event_bus

event_bus = get_event_bus()

Event Types

from core.event_bus import (
    BaseEvent,
    SkillGainEvent,
    LootEvent,
    DamageEvent,
    GlobalEvent,
    ChatEvent,
    EconomyEvent,
    SystemEvent,
    EventCategory
)

Publishing Events

# Async (non-blocking)
event_bus.publish(event)

# Sync (blocking, returns count)
count = event_bus.publish_sync(event)

Subscribing

# Basic subscription
sub_id = event_bus.subscribe(
    callback=my_callback,
    event_filter=filter_obj,
    replay_history=True,
    replay_count=100
)

# Typed subscription
sub_id = event_bus.subscribe_typed(
    SkillGainEvent,
    callback,
    skill_names=["Rifle", "Pistol"],
    replay_last=10
)

# Unsubscribe
event_bus.unsubscribe(sub_id)

Event Filters

from core.event_bus import EventFilter

# Create filter
filter_obj = EventFilter(
    event_types=[SkillGainEvent, LootEvent],
    categories=[EventCategory.SKILL, EventCategory.LOOT],
    min_damage=100,
    max_damage=500,
    mob_types=["Dragon", "Drake"],
    skill_names=["Rifle"],
    sources=["my_plugin"],
    custom_predicate=lambda e: e.value > 100
)

# Check match
if filter_obj.matches(event):
    print("Event matches filter!")

Getting Events

# Recent events
recent = event_bus.get_recent_events(
    event_type=SkillGainEvent,
    count=100,
    category=EventCategory.SKILL
)

# By time range
events = event_bus.get_events_by_time_range(
    start=datetime_obj,
    end=datetime_obj
)

Statistics

stats = event_bus.get_stats()
# Returns: {
#     'total_published': int,
#     'total_delivered': int,
#     'active_subscriptions': int,
#     'events_per_minute': float,
#     'avg_delivery_ms': float,
#     'errors': int,
#     'uptime_seconds': float,
#     'top_event_types': dict
# }

Event Classes

@dataclass(frozen=True)
class SkillGainEvent(BaseEvent):
    skill_name: str = ""
    skill_value: float = 0.0
    gain_amount: float = 0.0

@dataclass(frozen=True)
class LootEvent(BaseEvent):
    mob_name: str = ""
    items: List[Dict[str, Any]] = field(default_factory=list)
    total_tt_value: float = 0.0
    position: Optional[tuple] = None

@dataclass(frozen=True)
class DamageEvent(BaseEvent):
    damage_amount: float = 0.0
    damage_type: str = ""
    is_critical: bool = False
    target_name: str = ""
    attacker_name: str = ""
    is_outgoing: bool = True

@dataclass(frozen=True)
class GlobalEvent(BaseEvent):
    player_name: str = ""
    achievement_type: str = ""  # "hof", "ath", "discovery"
    value: float = 0.0
    item_name: Optional[str] = None

@dataclass(frozen=True)
class ChatEvent(BaseEvent):
    channel: str = ""  # "main", "team", "society"
    sender: str = ""
    message: str = ""

@dataclass(frozen=True)
class EconomyEvent(BaseEvent):
    transaction_type: str = ""  # "sale", "purchase"
    amount: float = 0.0
    currency: str = "PED"
    description: str = ""

@dataclass(frozen=True)
class SystemEvent(BaseEvent):
    message: str = ""
    severity: str = "info"  # "debug", "info", "warning", "error", "critical"

---

Task Manager

Class: TaskManager

Thread pool-based background task execution.

Singleton Access

from core.tasks import get_task_manager

task_manager = get_task_manager(max_workers=4)

Task Priorities

from core.tasks import TaskPriority

TaskPriority.HIGH     # 3
TaskPriority.NORMAL   # 2
TaskPriority.LOW      # 1

Running Tasks

# Run in thread
task_id = task_manager.run_in_thread(
    func=my_function,
    arg1, arg2,
    priority=TaskPriority.NORMAL,
    on_complete=callback,
    on_error=error_callback
)

# Run later
task_id = task_manager.run_later(
    delay_ms=5000,
    func=my_function,
    *args,
    priority=TaskPriority.NORMAL,
    on_complete=callback
)

# Run periodic
task_id = task_manager.run_periodic(
    interval_ms=30000,
    func=my_function,
    *args,
    priority=TaskPriority.NORMAL,
    on_complete=callback,
    run_immediately=True
)

Task Control

# Cancel
task_manager.cancel_task(task_id)

# Get status
status = task_manager.get_task_status(task_id)
# Returns: TaskStatus.PENDING, RUNNING, COMPLETED, FAILED, CANCELLED

# Wait for completion
task_manager.wait_for_task(task_id, timeout=10.0)

# Get all tasks
tasks = task_manager.get_all_tasks()

# Get task by ID
task = task_manager.get_task(task_id)

Signals

# Connect to signals
task_manager.connect_signal('completed', on_completed)
task_manager.connect_signal('failed', on_failed)
task_manager.connect_signal('started', on_started)
task_manager.connect_signal('cancelled', on_cancelled)
task_manager.connect_signal('periodic', on_periodic)

Shutdown

# Graceful shutdown
task_manager.shutdown(wait=True, timeout=30.0)

---

Nexus API

Class: NexusAPI

Client for Entropia Nexus API.

Singleton Access

from core.nexus_api import get_nexus_api

nexus = get_nexus_api()

Search Methods

# Search items
results = nexus.search_items("ArMatrix", limit=20)

# Search mobs
results = nexus.search_mobs("Atrox", limit=20)

# Search all types
results = nexus.search_all("Calypso", limit=30)

# Search by entity type
results = nexus.search_by_type("ArMatrix", entity_type="weapons", limit=20)

Detail Methods

# Get item details
details = nexus.get_item_details("armatrix_lp-35")

# Get entity details
details = nexus.get_entity_details("atrox", entity_type="mobs")

# Get market data
market = nexus.get_market_data("armatrix_lp-35")

Batch Methods

# Get multiple items
results = nexus.get_items_batch(["id1", "id2", "id3"])

# Get multiple market data
results = nexus.get_market_batch(["id1", "id2"])

Utility Methods

# Clear cache
nexus.clear_cache()

# Check availability
if nexus.is_available():
    results = nexus.search_items("query")

Data Classes

@dataclass
class SearchResult:
    id: str
    name: str
    type: str
    category: Optional[str]
    icon_url: Optional[str]
    data: Dict[str, Any]

@dataclass
class ItemDetails:
    id: str
    name: str
    description: Optional[str]
    category: Optional[str]
    weight: Optional[float]
    tt_value: Optional[float]
    decay: Optional[float]
    ammo_consumption: Optional[int]
    damage: Optional[float]
    range: Optional[float]
    accuracy: Optional[float]
    durability: Optional[int]
    requirements: Dict[str, Any]
    materials: List[Dict[str, Any]]
    raw_data: Dict[str, Any]

@dataclass
class MarketData:
    item_id: str
    item_name: str
    current_markup: Optional[float]
    avg_markup_7d: Optional[float]
    avg_markup_30d: Optional[float]
    volume_24h: Optional[int]
    volume_7d: Optional[int]
    buy_orders: List[Dict[str, Any]]
    sell_orders: List[Dict[str, Any]]
    last_updated: Optional[datetime]
    raw_data: Dict[str, Any]

---

Settings

Class: Settings

User preferences and configuration management.

Singleton Access

from core.settings import get_settings

settings = get_settings()

Default Settings

DEFAULTS = {
    'overlay_enabled': True,
    'overlay_opacity': 0.9,
    'overlay_theme': 'dark',
    'hotkey_toggle': 'ctrl+shift+u',
    'hotkey_search': 'ctrl+shift+f',
    'hotkey_calculator': 'ctrl+shift+c',
    'enabled_plugins': [...],
    'icon_size': 24,
    'accent_color': '#4a9eff',
    'check_updates': True,
    'data_retention_days': 30,
}

Methods

# Get setting
value = settings.get('key', default_value)

# Set setting
settings.set('key', value)

# Reset setting
settings.reset('key')  # Single key
settings.reset()       # All to defaults

# Plugin management
if settings.is_plugin_enabled('plugin_id'):
    settings.enable_plugin('plugin_id')
    settings.disable_plugin('plugin_id')

# Get all settings
all_settings = settings.all_settings()

# Save to disk
settings.save()

Signals

# Connect to changes
settings.setting_changed.connect(on_setting_changed)

def on_setting_changed(key, value):
    print(f"Setting {key} changed to {value}")

---

Data Store

Class: DataStore

Persistent key-value storage for plugins.

Singleton Access

from core.data_store import get_data_store

data_store = get_data_store()

Methods

# Store data
data_store.set('my_plugin.key', value)
data_store.set('my_plugin.nested', {'a': 1, 'b': 2})

# Retrieve data
value = data_store.get('my_plugin.key', default=None)

# Check existence
if data_store.has('my_plugin.key'):
    print("Key exists!")

# Delete
data_store.delete('my_plugin.key')

# Get all keys for plugin
keys = data_store.get_plugin_keys('my_plugin')

# Get all data for plugin
data = data_store.get_plugin_data('my_plugin')

# Clear plugin data
data_store.clear_plugin_data('my_plugin')

# Save to disk
data_store.save()

---

Audio Manager

Class: AudioManager

Sound playback and volume control.

Singleton Access

from core.audio import get_audio_manager

audio = get_audio_manager()

Methods

# Play sound
audio.play_sound('hof')                    # Predefined
audio.play_sound('/path/to/custom.wav')    # Custom
audio.play_sound('alert', blocking=True)   # Wait for completion

# Predefined sounds: 'global', 'hof', 'skill_gain', 'alert', 'error'

# Volume control
audio.set_volume(0.8)           # 0.0 to 1.0
volume = audio.get_volume()     # Get current

# Mute
audio.mute()
audio.unmute()
is_muted = audio.toggle_mute()
is_muted = audio.is_muted()

# Check availability
if audio.is_available():
    audio.play_sound('skill_gain')

# Get backend info
backend = audio.get_backend()   # 'pygame', 'pyaudio', etc.

# Shutdown
audio.shutdown()

---

HTTP Client

Class: HTTPClient

Cached HTTP requests with rate limiting.

Singleton Access

from core.http_client import get_http_client

http = get_http_client(
    cache_dir="cache/http",
    default_cache_ttl=3600,
    rate_limit_delay=0.1,
    max_retries=3,
    backoff_factor=0.5
)

Methods

# GET request
response = http.get(
    url,
    cache_ttl=300,              # Cache for 5 minutes
    headers={'Accept': 'application/json'},
    timeout=30
)

# Response format
{
    'status_code': 200,
    'headers': {...},
    'text': 'response body',
    'json': {...},              # Parsed JSON if applicable
    'cached': False,
    'cache_time': None
}

# Clear cache
http.clear_cache()

# Get cache stats
stats = http.get_cache_stats()

---

OCR Service

Class: OCRService

Text recognition from screen captures.

Singleton Access

from core.ocr_service import get_ocr_service

ocr = get_ocr_service()

Methods

# Initialize (lazy, called automatically)
ocr.initialize()

# Recognize text
result = ocr.recognize(region=(x, y, width, height))
# Returns: {
#     'text': 'extracted text',
#     'confidence': 0.95,
#     'raw_results': [...],
#     'error': None
# }

# Check availability
if ocr.is_available():
    result = ocr.recognize()

# Get available backends
backends = ocr.get_available_backends()  # ['easyocr', 'pytesseract']

# Shutdown
ocr.shutdown()

---

Screenshot Service

Class: ScreenshotService

Screen capture functionality.

Singleton Access

from core.screenshot import get_screenshot_service

screenshot = get_screenshot_service()

Methods

# Capture full screen
image = screenshot.capture(full_screen=True)

# Capture region
image = screenshot.capture_region(x, y, width, height)

# Get last screenshot
image = screenshot.get_last_screenshot()

# Save screenshot
path = screenshot.save_screenshot(image, filename="screenshot.png")

# Check availability
if screenshot.is_available():
    image = screenshot.capture()

# Get available backends
backends = screenshot.get_available_backends()  # ['pil', 'pyautogui']

---

Log Reader

Class: LogReader

Read Entropia Universe chat logs.

Singleton Access

from core.log_reader import get_log_reader

log_reader = get_log_reader()

Methods

# Start reading
log_reader.start()

# Read lines
lines = log_reader.read_lines(count=50, filter_text="loot")

# Check availability
if log_reader.is_available():
    lines = log_reader.read_lines()

# Get log path
path = log_reader.get_log_path()

# Stop reading
log_reader.stop()

---

Window Manager

Class: WindowManager

Windows-specific window management.

Singleton Access

from core.window_manager import get_window_manager

wm = get_window_manager()

Methods

# Check availability (Windows only)
if wm.is_available():
    # Find EU window
    eu_window = wm.find_eu_window()
    
    if eu_window:
        print(f"Title: {eu_window.title}")
        print(f"Size: {eu_window.width}x{eu_window.height}")
        print(f"Position: ({eu_window.x}, {eu_window.y})")
    
    # Check if focused
    if wm.is_eu_focused():
        print("EU is active!")
    
    # Get all windows
    windows = wm.get_all_windows()
    
    # Find by title
    window = wm.find_window_by_title("Entropia Universe")
    
    # Get foreground window
    active = wm.get_foreground_window()

---

Notification Manager

Class: NotificationManager

Desktop notifications.

Singleton Access

from core.notifications import get_notification_manager

notifications = get_notification_manager()

Methods

# Initialize
notifications.initialize(app)

# Show notification
notifications.show(
    title="Skill Gain!",
    message="Rifle increased to 25.5",
    duration_ms=3000
)

# Show with actions
notifications.show(
    title="Global!",
    message="You got a global!",
    actions=[
        {"label": "View", "callback": view_callback},
        {"label": "Dismiss", "callback": dismiss_callback}
    ]
)

# Close all
notifications.close_all()

---

Clipboard Manager

Class: ClipboardManager

Clipboard operations.

Singleton Access

from core.clipboard import get_clipboard_manager

clipboard = get_clipboard_manager()

Methods

# Copy to clipboard
clipboard.copy("text to copy")

# Paste from clipboard
text = clipboard.paste()

# Check availability
if clipboard.is_available():
    clipboard.copy("Hello!")

---

API Reference Complete 📚