EU-Utility/API.md

EU-Utility API Reference

Complete API documentation for plugin developers

Version: 2.1.0
Last Updated: 2026-02-16

---

Table of Contents

1. Getting Started
2. PluginAPI - Core services access
3. WidgetAPI - Overlay widget management
4. ExternalAPI - REST endpoints and webhooks
5. Event Bus - Pub/sub event system
6. BasePlugin Class - Plugin base class reference
7. Nexus API - Entropia Nexus integration
8. Code Examples - Practical examples
9. Best Practices

---

Getting Started

Importing the API

from core.api import get_api

class MyPlugin(BasePlugin):
    def initialize(self):
        self.api = get_api()

API Availability

Always check if a service is available before using it:

if self.api.ocr_available():
    text = self.api.recognize_text()

---

PluginAPI

The PluginAPI provides access to all core EU-Utility services.

Window Manager

# Get EU window information
window = api.get_eu_window()
if window:
    print(f"Position: {window['x']}, {window['y']}")
    print(f"Size: {window['width']}x{window['height']}")
    print(f"Focused: {window['is_focused']}")

# Check if EU is focused
if api.is_eu_focused():
    api.play_sound("alert.wav")

# Bring EU to front
api.bring_eu_to_front()

# Check if EU window is visible
is_visible = api.is_eu_visible()

OCR Service

# Check OCR availability
if api.ocr_available():
    # Read text from screen region
    text = api.recognize_text(region=(100, 100, 200, 50))
    print(f"Found: {text}")

# Alternative: Use BasePlugin helper method
result = self.ocr_capture(region=(x, y, width, height))
# Returns: {'text': str, 'confidence': float, 'error': str or None}

Screenshot Service

# Check availability
if api.screenshot_available():
    # Capture full screen
    img = api.capture_screen()
    
    # Capture specific region
    img = api.capture_screen(region=(100, 100, 400, 300))
    
    # Capture and save
    img = api.capture_screen(
        region=(100, 100, 400, 300),
        save_path="screenshot.png"
    )

# Alternative: Use BasePlugin helper
img = self.capture_screen(full_screen=True)
img = self.capture_region(x, y, width, height)

Log Reader

# Read recent log lines
lines = api.read_log_lines(count=100)

# Read logs since timestamp
from datetime import datetime, timedelta
recent = api.read_log_since(datetime.now() - timedelta(minutes=5))

# Alternative: Use BasePlugin helper
lines = self.read_log(lines=50, filter_text="loot")

Data Store

# Store data (scoped to plugin)
api.set_data("key", value)

# Retrieve with default
value = api.get_data("key", default=None)

# Delete data
api.delete_data("key")

# Alternative: Use BasePlugin helpers
self.save_data("key", value)
data = self.load_data("key", default=None)
self.delete_data("key")

HTTP Client

# GET request with caching
result = api.http_get(
    "https://api.example.com/data",
    cache=True,
    cache_duration=3600  # 1 hour
)

if result['success']:
    data = result['data']
else:
    error = result['error']

# POST request
result = api.http_post(
    "https://api.example.com/submit",
    data={"key": "value"}
)

# Alternative: Use BasePlugin helper
response = self.http_get(url, cache_ttl=300, headers={})

Audio

# Play sound file
api.play_sound("assets/sounds/alert.wav", volume=0.7)

# Simple beep
api.beep()

# Alternative: Use BasePlugin helpers
self.play_sound("hof")  # Predefined: 'hof', 'skill_gain', 'alert'
self.set_volume(0.8)
volume = self.get_volume()

Notifications

# Show toast notification
api.show_notification(
    title="Loot Alert!",
    message="You found something valuable!",
    duration=5000,  # milliseconds
    sound=True
)

# Alternative: Use BasePlugin helpers
self.notify(title, message, notification_type='info', sound=False)
self.notify_info(title, message)
self.notify_success(title, message)
self.notify_warning(title, message)
self.notify_error(title, message, sound=True)

Clipboard

# Copy to clipboard
api.copy_to_clipboard("Text to copy")

# Paste from clipboard
text = api.paste_from_clipboard()

# Alternative: Use BasePlugin helpers
self.copy_to_clipboard(text)
text = self.paste_from_clipboard()

Background Tasks

# Run function in background
def heavy_computation(data):
    # Long running task
    import time
    time.sleep(2)
    return f"Processed: {data}"

def on_complete(result):
    print(f"Done: {result}")

def on_error(error):
    print(f"Error: {error}")

task_id = api.run_task(
    heavy_computation,
    "my data",
    callback=on_complete,
    error_handler=on_error
)

# Cancel task
api.cancel_task(task_id)

# Alternative: Use BasePlugin helper
self.run_in_background(func, *args, priority='normal', 
                       on_complete=cb, on_error=err_cb)

---

WidgetAPI

Manage overlay widgets - floating UI components.

Getting Started

from core.api import get_widget_api

widget_api = get_widget_api()

# Create widget
widget = widget_api.create_widget(
    name="loot_tracker",
    title="Loot Tracker",
    size=(400, 300),
    position=(100, 100)
)

widget.show()

Widget Operations

# Show/hide
widget.show()
widget.hide()

# Position
widget.move(500, 200)
x, y = widget.position

# Size
widget.resize(400, 300)
width, height = widget.size

# Opacity (0.0 - 1.0)
widget.set_opacity(0.8)

# Lock/unlock (prevent dragging)
widget.set_locked(True)

# Minimize/restore
widget.minimize()
widget.restore()

# Close
widget.close()

Widget Management

# Get existing widget
widget = widget_api.get_widget("loot_tracker")

# Show/hide specific widget
widget_api.show_widget("loot_tracker")
widget_api.hide_widget("loot_tracker")

# Close widget
widget_api.close_widget("loot_tracker")

# Global operations
widget_api.show_all_widgets()
widget_api.hide_all_widgets()
widget_api.close_all_widgets()
widget_api.set_all_opacity(0.8)
widget_api.lock_all()
widget_api.unlock_all()

Layout Helpers

# Arrange widgets
widget_api.arrange_widgets(layout="grid", spacing=10)
widget_api.arrange_widgets(layout="horizontal")
widget_api.arrange_widgets(layout="vertical")
widget_api.arrange_widgets(layout="cascade")

# Snap to grid
widget_api.snap_to_grid(grid_size=10)

Widget Events

# Handle events
widget.on('moved', lambda data: print(f"Moved to {data['x']}, {data['y']}"))
widget.on('resized', lambda data: print(f"Sized to {data['width']}x{data['height']}"))
widget.on('closing', lambda: print("Widget closing"))
widget.on('closed', lambda: print("Widget closed"))
widget.on('update', lambda data: print(f"Update: {data}"))

---

ExternalAPI

REST endpoints, webhooks, and third-party integrations.

Getting Started

from core.api import get_external_api

ext = get_external_api()

# Start server
ext.start_server(port=8080)

# Check status
print(ext.get_status())

REST Endpoints

# Using decorator
@ext.endpoint("stats", methods=["GET"])
def get_stats():
    return {"kills": 100, "loot": "50 PED"}

@ext.endpoint("loot", methods=["POST"])
def record_loot(data):
    save_loot(data)
    return {"status": "saved"}

# Programmatic registration
def get_stats_handler(params):
    return {"kills": 100}

ext.register_endpoint("stats", get_stats_handler, methods=["GET"])

# Unregister
ext.unregister_endpoint("stats")

Incoming Webhooks

# Register webhook handler
def handle_discord(payload):
    print(f"Discord: {payload}")
    return {"status": "ok"}

ext.register_webhook(
    name="discord",
    handler=handle_discord,
    secret="my_secret"  # Optional HMAC verification
)

# POST to: http://localhost:8080/webhook/discord

Outgoing Webhooks

# POST to external webhook
result = ext.post_webhook(
    "https://discord.com/api/webhooks/...",
    {"content": "Hello from EU-Utility!"}
)

if result['success']:
    print("Sent!")
else:
    print(f"Error: {result['error']}")

---

Event Bus

Typed publish-subscribe event system for inter-plugin communication.

Event Types

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

Publishing Events

from core.event_bus import LootEvent, SkillGainEvent

# Publish loot event
self.publish_typed(LootEvent(
    mob_name="Daikiba",
    items=[{"name": "Animal Oil", "value": 0.05}],
    total_tt_value=0.05
))

# Publish skill gain
self.publish_typed(SkillGainEvent(
    skill_name="Rifle",
    skill_value=25.5,
    gain_amount=0.01
))

# Legacy event publishing
api.publish("my_plugin.event", {"data": "value"})

Subscribing to Events

from core.event_bus import LootEvent, SkillGainEvent

class MyPlugin(BasePlugin):
    def initialize(self):
        # Subscribe to loot events
        self.sub_id = self.subscribe_typed(
            LootEvent,
            self.on_loot
        )
        
        # Subscribe with filtering
        self.sub_id2 = self.subscribe_typed(
            LootEvent,
            self.on_dragon_loot,
            mob_types=["Dragon", "Drake"]
        )
        
        # Subscribe to specific skills
        self.sub_id3 = self.subscribe_typed(
            SkillGainEvent,
            self.on_combat_skill,
            skill_names=["Rifle", "Pistol", "Melee"]
        )
    
    def on_loot(self, event: LootEvent):
        print(f"Loot from {event.mob_name}: {event.items}")
    
    def on_dragon_loot(self, event: LootEvent):
        print(f"Dragon loot! Total TT: {event.total_tt_value}")
    
    def shutdown(self):
        # Clean up subscriptions
        self.unsubscribe_typed(self.sub_id)
        self.unsubscribe_typed(self.sub_id2)
        self.unsubscribe_typed(self.sub_id3)
        # Or: self.unsubscribe_all_typed()

Event Attributes

Event	Attributes
SkillGainEvent	skill_name, skill_value, gain_amount
LootEvent	mob_name, items, total_tt_value, position
DamageEvent	damage_amount, damage_type, is_critical, target_name
GlobalEvent	player_name, achievement_type, value, item_name
ChatEvent	channel, sender, message
EconomyEvent	transaction_type, amount, currency, description
SystemEvent	message, severity

Legacy Event Bus

# Subscribe
sub_id = api.subscribe("loot", on_loot_callback)

# Unsubscribe
api.unsubscribe(sub_id)

# Publish
api.publish("event_type", {"key": "value"})

# Get history
history = api.get_event_history("loot", limit=10)

---

BasePlugin Class

Complete reference for the base plugin class.

Required Attributes

class MyPlugin(BasePlugin):
    name = "My Plugin"           # Display name
    version = "1.0.0"           # Version string
    author = "Your Name"        # Author name
    description = "..."         # Short description

Optional Attributes

class MyPlugin(BasePlugin):
    icon = "path/to/icon.png"   # Icon path
    hotkey = "ctrl+shift+y"     # Legacy single hotkey
    hotkeys = [                 # Multi-hotkey format
        {
            'action': 'toggle',
            'description': 'Toggle My Plugin',
            'default': 'ctrl+shift+m',
            'config_key': 'myplugin_toggle'
        }
    ]
    enabled = True              # Start enabled
    
    dependencies = {            # Dependencies
        'pip': ['requests', 'numpy'],
        'plugins': ['other_plugin'],
        'optional': {'pillow': 'Image processing'}
    }

Lifecycle Methods

def initialize(self) -> None:
    """Called when plugin is loaded."""
    self.api = get_api()
    self.log_info("Initialized!")

def get_ui(self) -> QWidget:
    """Return the plugin's UI widget."""
    widget = QWidget()
    # ... setup UI ...
    return widget

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 hotkey is pressed."""
    pass

def shutdown(self) -> None:
    """Called when app is closing. Cleanup resources."""
    self.unsubscribe_all_typed()
    super().shutdown()

Config Methods

# Get config value
theme = self.get_config('theme', default='dark')

# Set config value
self.set_config('theme', 'light')

Logging Methods

self.log_debug("Debug message")
self.log_info("Info message")
self.log_warning("Warning message")
self.log_error("Error message")

Utility Methods

# Format currency
ped_text = self.format_ped(123.45)      # "123.45 PED"
pec_text = self.format_pec(50)          # "50 PEC"

# Calculate DPP
dpp = self.calculate_dpp(damage=45.0, ammo=100, decay=2.5)

# Calculate markup
markup = self.calculate_markup(price=150.0, tt=100.0)  # 150.0%

---

Nexus API

Entropia Nexus integration for game data.

Searching

# Search for items
results = self.nexus_search("ArMatrix", entity_type="items")

# Search for mobs
mobs = self.nexus_search("Atrox", entity_type="mobs")

# Search for locations
locations = self.nexus_search("Fort", entity_type="locations")

Entity Types

• items, weapons, armors
• mobs, pets
• blueprints, materials
• locations, teleporters, shops, vendors, planets, areas
• skills
• enhancers, medicaltools, finders, excavators, refiners
• vehicles, decorations, furniture
• storagecontainers, strongboxes

Getting Details

# Get item details
details = self.nexus_get_item_details("armatrix_lp-35")
if details:
    print(f"Name: {details['name']}")
    print(f"TT Value: {details['tt_value']} PED")

# Get market data
market = self.nexus_get_market_data("armatrix_lp-35")
if market:
    print(f"Current markup: {market['current_markup']:.1f}%")

---

Code Examples

Example 1: Simple Calculator Plugin

from plugins.base_plugin import BasePlugin
from PyQt6.QtWidgets import (
    QWidget, QVBoxLayout, QHBoxLayout,
    QLabel, QLineEdit, QPushButton
)

class MarkupCalculatorPlugin(BasePlugin):
    """Calculate markup percentages."""
    
    name = "Markup Calculator"
    version = "1.0.0"
    author = "Tutorial"
    description = "Calculate item markup"
    hotkey = "ctrl+shift+m"
    
    def initialize(self):
        self.log_info("Markup Calculator initialized!")
    
    def get_ui(self):
        widget = QWidget()
        layout = QVBoxLayout(widget)
        
        # Title
        title = QLabel("Markup Calculator")
        title.setStyleSheet("color: #4a9eff; font-size: 18px; font-weight: bold;")
        layout.addWidget(title)
        
        # TT Value input
        layout.addWidget(QLabel("TT Value:"))
        self.tt_input = QLineEdit()
        self.tt_input.setPlaceholderText("100.00")
        layout.addWidget(self.tt_input)
        
        # Market Price input
        layout.addWidget(QLabel("Market Price:"))
        self.price_input = QLineEdit()
        self.price_input.setPlaceholderText("150.00")
        layout.addWidget(self.price_input)
        
        # Calculate button
        calc_btn = QPushButton("Calculate")
        calc_btn.clicked.connect(self.calculate)
        layout.addWidget(calc_btn)
        
        # Result
        self.result_label = QLabel("Markup: -")
        self.result_label.setStyleSheet("color: #ffc107; font-size: 16px;")
        layout.addWidget(self.result_label)
        
        layout.addStretch()
        return widget
    
    def calculate(self):
        try:
            tt = float(self.tt_input.text() or 0)
            price = float(self.price_input.text() or 0)
            markup = self.calculate_markup(price, tt)
            self.result_label.setText(f"Markup: {markup:.1f}%")
        except ValueError:
            self.result_label.setText("Invalid input")

Example 2: Event-Driven Tracker

from plugins.base_plugin import BasePlugin
from PyQt6.QtWidgets import QWidget, QVBoxLayout, QLabel, QListWidget
from core.event_bus import LootEvent, SkillGainEvent

class ActivityTrackerPlugin(BasePlugin):
    """Track recent activity from events."""
    
    name = "Activity Tracker"
    version = "1.0.0"
    
    def initialize(self):
        self.subscriptions = []
        
        # Subscribe to multiple event types
        sub1 = self.subscribe_typed(LootEvent, self.on_loot)
        sub2 = self.subscribe_typed(SkillGainEvent, self.on_skill)
        
        self.subscriptions.extend([sub1, sub2])
        self.log_info("Activity Tracker initialized")
    
    def get_ui(self):
        widget = QWidget()
        layout = QVBoxLayout(widget)
        
        layout.addWidget(QLabel("Recent Activity:"))
        
        self.activity_list = QListWidget()
        layout.addWidget(self.activity_list)
        
        return widget
    
    def on_loot(self, event: LootEvent):
        text = f"Loot: {event.mob_name} - {event.total_tt_value:.2f} PED"
        self.activity_list.insertItem(0, text)
    
    def on_skill(self, event: SkillGainEvent):
        text = f"Skill: {event.skill_name} +{event.gain_amount:.4f}"
        self.activity_list.insertItem(0, text)
    
    def shutdown(self):
        for sub_id in self.subscriptions:
            self.unsubscribe_typed(sub_id)

Example 3: Background Task Plugin

from plugins.base_plugin import BasePlugin
from PyQt6.QtWidgets import (
    QWidget, QVBoxLayout, QPushButton, 
    QLabel, QProgressBar
)

class DataFetcherPlugin(BasePlugin):
    """Fetch data in background."""
    
    name = "Data Fetcher"
    version = "1.0.0"
    
    def initialize(self):
        self.log_info("Data Fetcher initialized")
    
    def get_ui(self):
        widget = QWidget()
        layout = QVBoxLayout(widget)
        
        self.status_label = QLabel("Ready")
        layout.addWidget(self.status_label)
        
        self.progress = QProgressBar()
        self.progress.setRange(0, 0)  # Indeterminate
        self.progress.hide()
        layout.addWidget(self.progress)
        
        fetch_btn = QPushButton("Fetch Data")
        fetch_btn.clicked.connect(self.fetch_data)
        layout.addWidget(fetch_btn)
        
        layout.addStretch()
        return widget
    
    def fetch_data(self):
        self.status_label.setText("Fetching...")
        self.progress.show()
        
        # Run in background
        self.run_in_background(
            self._fetch_from_api,
            priority='normal',
            on_complete=self._on_fetch_complete,
            on_error=self._on_fetch_error
        )
    
    def _fetch_from_api(self):
        # This runs in background thread
        import time
        time.sleep(2)  # Simulate API call
        return {"items": ["A", "B", "C"], "count": 3}
    
    def _on_fetch_complete(self, result):
        # This runs in main thread - safe to update UI
        self.progress.hide()
        self.status_label.setText(f"Got {result['count']} items")
        self.notify_success("Data Fetched", f"Retrieved {result['count']} items")
    
    def _on_fetch_error(self, error):
        self.progress.hide()
        self.status_label.setText(f"Error: {error}")
        self.notify_error("Fetch Failed", str(error))

---

Best Practices

1. Always Use BasePlugin Helpers

# ✅ Good - Uses helper methods
self.save_data("key", value)
data = self.load_data("key", default)

# ❌ Avoid - Direct API access when helper exists
self.api.set_data("key", value)

2. Handle Errors Gracefully

def initialize(self):
    try:
        self.api = get_api()
        self.log_info("API connected")
    except Exception as e:
        self.log_error(f"Failed to get API: {e}")
        # Continue with limited functionality

3. Clean Up Resources

def shutdown(self):
    # Unsubscribe from events
    self.unsubscribe_all_typed()
    
    # Save any pending data
    self.save_data("pending", self.pending_data)
    
    super().shutdown()

4. Don't Block the Main Thread

# ✅ Good - Use background tasks
def heavy_operation(self):
    self.run_in_background(
        self._do_heavy_work,
        on_complete=self._update_ui
    )

# ❌ Avoid - Blocks UI
def heavy_operation(self):
    result = self._do_heavy_work()  # Blocks!
    self._update_ui(result)

5. Use Type Hints

from typing import Optional, Dict, Any, List
from core.event_bus import LootEvent

def on_loot(self, event: LootEvent) -> None:
    items: List[Dict[str, Any]] = event.items
    total: float = event.total_tt_value

6. Follow Naming Conventions

# ✅ Good
class LootTrackerPlugin(BasePlugin):
    def on_loot_received(self):
        pass

# ❌ Avoid
class lootTracker(BasePlugin):
    def loot(self):
        pass

7. Document Your Plugin

class MyPlugin(BasePlugin):
    """
    Brief description of what this plugin does.
    
    Features:
    - Feature 1
    - Feature 2
    
    Usage:
    1. Step 1
    2. Step 2
    
    Hotkeys:
    - Ctrl+Shift+Y: Toggle plugin
    """

---

Error Handling

All APIs provide specific exceptions:

from core.api import (
    PluginAPIError,
    ServiceNotAvailableError,
    ExternalAPIError
)

try:
    text = api.recognize_text((0, 0, 100, 100))
except ServiceNotAvailableError:
    print("OCR not available")
except PluginAPIError as e:
    print(f"API error: {e}")

---

See Also

• Plugin Development Guide - Detailed plugin tutorial
• Architecture Overview - System architecture
• Nexus API Reference - Entropia Nexus API
• Contributing - How to contribute