600 lines
12 KiB
Markdown
600 lines
12 KiB
Markdown
# EU-Utility API Documentation
|
|
|
|
Complete API reference for EU-Utility core services and plugin development.
|
|
|
|
## Table of Contents
|
|
|
|
1. [Plugin API](#plugin-api)
|
|
2. [Window Manager](#window-manager)
|
|
3. [Event Bus](#event-bus)
|
|
4. [Data Store](#data-store)
|
|
5. [Nexus API](#nexus-api)
|
|
6. [HTTP Client](#http-client)
|
|
7. [Creating Plugins](#creating-plugins)
|
|
|
|
---
|
|
|
|
## Plugin API
|
|
|
|
The Plugin API is the primary interface for plugin developers.
|
|
|
|
### Getting the API
|
|
|
|
```python
|
|
from core.plugin_api import get_api
|
|
|
|
class MyPlugin(BasePlugin):
|
|
def initialize(self):
|
|
self.api = get_api()
|
|
```
|
|
|
|
### Log Reader
|
|
|
|
```python
|
|
# Read recent log lines
|
|
lines = self.api.read_log_lines(count=100)
|
|
|
|
# Read logs since timestamp
|
|
lines = self.api.read_log_since(timestamp)
|
|
```
|
|
|
|
### Window Manager
|
|
|
|
```python
|
|
# Get EU window info
|
|
window = self.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 EU focus
|
|
if self.api.is_eu_focused():
|
|
self.api.play_sound("alert.wav")
|
|
|
|
# Bring EU to front
|
|
self.api.bring_eu_to_front()
|
|
```
|
|
|
|
### OCR
|
|
|
|
```python
|
|
# Check OCR availability
|
|
if self.api.ocr_available():
|
|
# Recognize text from screen region
|
|
text = self.api.recognize_text(region=(100, 100, 200, 50))
|
|
|
|
# Recognize from image file
|
|
text = self.api.recognize_text(image_path="screenshot.png")
|
|
```
|
|
|
|
### Screenshot
|
|
|
|
```python
|
|
# Check screenshot availability
|
|
if self.api.screenshot_available():
|
|
# Capture screen region
|
|
img = self.api.capture_screen(region=(0, 0, 1920, 1080))
|
|
|
|
# Capture and save
|
|
img = self.api.capture_screen(
|
|
region=(100, 100, 200, 200),
|
|
save_path="screenshot.png"
|
|
)
|
|
```
|
|
|
|
### Nexus API
|
|
|
|
```python
|
|
# Search for items
|
|
items = self.api.search_items("omegaton", limit=10)
|
|
for item in items:
|
|
print(f"{item['Name']}: {item['Value']} PED")
|
|
|
|
# Get item details
|
|
details = self.api.get_item_details(item_id=12345)
|
|
```
|
|
|
|
### HTTP Client
|
|
|
|
```python
|
|
# GET request with caching
|
|
result = self.api.http_get(
|
|
"https://api.example.com/data",
|
|
cache=True,
|
|
cache_duration=3600
|
|
)
|
|
|
|
if result['success']:
|
|
data = result['data']
|
|
else:
|
|
error = result['error']
|
|
|
|
# POST request
|
|
result = self.api.http_post(
|
|
"https://api.example.com/submit",
|
|
data={"key": "value"}
|
|
)
|
|
```
|
|
|
|
### Audio
|
|
|
|
```python
|
|
# Play sound
|
|
self.api.play_sound("assets/sounds/alert.wav", volume=0.7)
|
|
|
|
# Simple beep
|
|
self.api.beep()
|
|
```
|
|
|
|
### Notifications
|
|
|
|
```python
|
|
# Show notification
|
|
self.api.show_notification(
|
|
title="Loot Alert!",
|
|
message="You found something valuable!",
|
|
duration=5000, # milliseconds
|
|
sound=True
|
|
)
|
|
```
|
|
|
|
### Clipboard
|
|
|
|
```python
|
|
# Copy to clipboard
|
|
self.api.copy_to_clipboard("Text to copy")
|
|
|
|
# Paste from clipboard
|
|
text = self.api.paste_from_clipboard()
|
|
```
|
|
|
|
### Event Bus
|
|
|
|
```python
|
|
# Subscribe to events
|
|
self.sub_id = self.api.subscribe("loot", self.on_loot)
|
|
|
|
# Unsubscribe
|
|
self.api.unsubscribe(self.sub_id)
|
|
|
|
# Publish event
|
|
self.api.publish("my_plugin.event", {"data": "value"})
|
|
```
|
|
|
|
### Data Store
|
|
|
|
```python
|
|
# Store data
|
|
self.api.set_data("key", value)
|
|
|
|
# Retrieve data
|
|
value = self.api.get_data("key", default="default")
|
|
|
|
# Delete data
|
|
self.api.delete_data("key")
|
|
```
|
|
|
|
### Background Tasks
|
|
|
|
```python
|
|
# Run function in background
|
|
def heavy_computation(data):
|
|
# Long running task
|
|
return result
|
|
|
|
def on_complete(result):
|
|
print(f"Done: {result}")
|
|
|
|
task_id = self.api.run_task(
|
|
heavy_computation,
|
|
my_data,
|
|
callback=on_complete,
|
|
error_handler=lambda e: print(f"Error: {e}")
|
|
)
|
|
|
|
# Cancel task
|
|
self.api.cancel_task(task_id)
|
|
```
|
|
|
|
---
|
|
|
|
## Window Manager
|
|
|
|
Direct access to window management functionality.
|
|
|
|
```python
|
|
from core.window_manager import get_window_manager, is_eu_running
|
|
|
|
wm = get_window_manager()
|
|
|
|
# Check availability
|
|
if wm.is_available():
|
|
# Find EU window
|
|
window = wm.find_eu_window()
|
|
|
|
if window:
|
|
print(f"Title: {window.title}")
|
|
print(f"PID: {window.pid}")
|
|
print(f"Rect: {window.rect}")
|
|
|
|
# Check focus
|
|
if wm.is_window_focused():
|
|
print("EU is focused")
|
|
|
|
# Bring to front
|
|
wm.bring_to_front()
|
|
|
|
# Get window rectangle
|
|
rect = wm.get_window_rect()
|
|
|
|
# Get process info
|
|
process = wm.get_eu_process_info()
|
|
|
|
# Quick check
|
|
if is_eu_running():
|
|
print("EU is running!")
|
|
```
|
|
|
|
---
|
|
|
|
## Event Bus
|
|
|
|
Publish-subscribe event system for inter-plugin communication.
|
|
|
|
```python
|
|
from core.event_bus import EventBus, get_event_bus
|
|
|
|
# Get global event bus
|
|
event_bus = get_event_bus()
|
|
|
|
# Subscribe to events
|
|
def on_loot(event):
|
|
print(f"Loot: {event.data}")
|
|
print(f"Type: {event.type}")
|
|
print(f"Timestamp: {event.timestamp}")
|
|
|
|
sub_id = event_bus.subscribe("loot", on_loot)
|
|
|
|
# Publish event
|
|
event_bus.publish("loot", {
|
|
"item": "Shrapnel",
|
|
"amount": 50
|
|
})
|
|
|
|
# Unsubscribe
|
|
event_bus.unsubscribe(sub_id)
|
|
|
|
# Get event history
|
|
history = event_bus.get_event_history("loot", limit=10)
|
|
|
|
# Clear history
|
|
event_bus.clear_history()
|
|
```
|
|
|
|
---
|
|
|
|
## Data Store
|
|
|
|
Persistent key-value storage for plugins.
|
|
|
|
```python
|
|
from core.data_store import DataStore
|
|
|
|
# Create store
|
|
store = DataStore("path/to/data.json")
|
|
|
|
# Basic operations
|
|
store.set("key", "value")
|
|
value = store.get("key")
|
|
store.delete("key")
|
|
exists = store.has("key")
|
|
|
|
# Complex data
|
|
store.set("player", {
|
|
"name": "Avatar Name",
|
|
"level": 45,
|
|
"skills": {
|
|
"rifle": 2500,
|
|
"pistol": 1800
|
|
}
|
|
})
|
|
|
|
player = store.get("player")
|
|
print(player["skills"]["rifle"]) # 2500
|
|
|
|
# Batch operations
|
|
store.set_multi({
|
|
"key1": "value1",
|
|
"key2": "value2"
|
|
})
|
|
|
|
all_data = store.get_all()
|
|
|
|
# Clear all
|
|
store.clear()
|
|
|
|
# Persistence
|
|
store.save() # Save to disk
|
|
store.load() # Load from disk
|
|
```
|
|
|
|
---
|
|
|
|
## Nexus API
|
|
|
|
Interface to Entropia Nexus data.
|
|
|
|
```python
|
|
from core.nexus_api import NexusAPI
|
|
|
|
nexus = NexusAPI()
|
|
|
|
# Search items
|
|
items = nexus.search_items("omegaton", limit=10)
|
|
for item in items:
|
|
print(f"{item['Name']}: {item['Value']} PED")
|
|
|
|
# Get item details
|
|
details = nexus.get_item(item_id=12345)
|
|
print(details["Name"])
|
|
print(details["Value"])
|
|
print(details["Markup"])
|
|
|
|
# Get creature info
|
|
creature = nexus.get_creature("Feffoid")
|
|
print(creature["Name"])
|
|
print(creature["Health"])
|
|
print(creature["Damage"])
|
|
|
|
# Get location info
|
|
location = nexus.get_location("Port Atlantis")
|
|
print(location["Coordinates"])
|
|
print(location["Teleporters"])
|
|
```
|
|
|
|
---
|
|
|
|
## HTTP Client
|
|
|
|
Web requests with built-in caching.
|
|
|
|
```python
|
|
from core.http_client import HTTPClient
|
|
|
|
client = HTTPClient()
|
|
|
|
# GET request
|
|
response = client.get("https://api.example.com/data")
|
|
if response['success']:
|
|
data = response['data']
|
|
|
|
# With caching
|
|
response = client.get(
|
|
"https://api.example.com/data",
|
|
cache=True,
|
|
cache_duration=3600 # 1 hour
|
|
)
|
|
|
|
# POST request
|
|
response = client.post(
|
|
"https://api.example.com/submit",
|
|
data={"key": "value"}
|
|
)
|
|
|
|
# Clear cache
|
|
client.clear_cache()
|
|
|
|
# Get cache info
|
|
info = client.get_cache_info()
|
|
print(f"Entries: {info['entries']}")
|
|
print(f"Size: {info['size_mb']} MB")
|
|
```
|
|
|
|
---
|
|
|
|
## Creating Plugins
|
|
|
|
### Basic Plugin Structure
|
|
|
|
```python
|
|
from plugins.base_plugin import BasePlugin
|
|
from PyQt6.QtWidgets import QWidget, QVBoxLayout, QLabel, QPushButton
|
|
|
|
class MyPlugin(BasePlugin):
|
|
"""My custom EU-Utility plugin."""
|
|
|
|
# Plugin metadata
|
|
name = "My Plugin"
|
|
version = "1.0.0"
|
|
author = "Your Name"
|
|
description = "What my plugin does"
|
|
hotkey = "ctrl+shift+y"
|
|
|
|
# Dependencies (optional)
|
|
requirements = ["requests", "numpy"]
|
|
|
|
def initialize(self):
|
|
"""Called when plugin is loaded."""
|
|
self.api = get_api()
|
|
self.data = DataStore("data/my_plugin.json")
|
|
self.log_info("My Plugin initialized!")
|
|
|
|
def shutdown(self):
|
|
"""Called when plugin is unloaded."""
|
|
self.data.save()
|
|
self.log_info("My Plugin shutdown")
|
|
|
|
def get_ui(self):
|
|
"""Return the plugin's UI widget."""
|
|
widget = QWidget()
|
|
layout = QVBoxLayout(widget)
|
|
|
|
label = QLabel("Hello from My Plugin!")
|
|
layout.addWidget(label)
|
|
|
|
button = QPushButton("Click Me")
|
|
button.clicked.connect(self.on_click)
|
|
layout.addWidget(button)
|
|
|
|
return widget
|
|
|
|
def on_hotkey(self):
|
|
"""Called when hotkey is pressed."""
|
|
self.api.show_notification("My Plugin", "Hotkey pressed!")
|
|
|
|
def on_click(self):
|
|
"""Handle button click."""
|
|
self.log_info("Button clicked")
|
|
```
|
|
|
|
### Plugin with Background Tasks
|
|
|
|
```python
|
|
from plugins.base_plugin import BasePlugin
|
|
from PyQt6.QtWidgets import QWidget, QVBoxLayout, QPushButton, QLabel
|
|
|
|
class AsyncPlugin(BasePlugin):
|
|
name = "Async Plugin"
|
|
|
|
def initialize(self):
|
|
self.api = get_api()
|
|
self.result_label = None
|
|
|
|
def get_ui(self):
|
|
widget = QWidget()
|
|
layout = QVBoxLayout(widget)
|
|
|
|
self.result_label = QLabel("Ready")
|
|
layout.addWidget(self.result_label)
|
|
|
|
btn = QPushButton("Start Task")
|
|
btn.clicked.connect(self.start_task)
|
|
layout.addWidget(btn)
|
|
|
|
return widget
|
|
|
|
def start_task(self):
|
|
self.result_label.setText("Working...")
|
|
|
|
# Run in background
|
|
self.api.run_task(
|
|
self.heavy_work,
|
|
"input data",
|
|
callback=self.on_complete,
|
|
error_handler=self.on_error
|
|
)
|
|
|
|
def heavy_work(self, data):
|
|
# This runs in background thread
|
|
import time
|
|
time.sleep(2)
|
|
return f"Result: {data.upper()}"
|
|
|
|
def on_complete(self, result):
|
|
# This runs in main thread
|
|
self.result_label.setText(result)
|
|
|
|
def on_error(self, error):
|
|
self.result_label.setText(f"Error: {error}")
|
|
```
|
|
|
|
### Plugin with Event Subscription
|
|
|
|
```python
|
|
from plugins.base_plugin import BasePlugin
|
|
from PyQt6.QtWidgets import QWidget, QVBoxLayout, QLabel
|
|
|
|
class EventPlugin(BasePlugin):
|
|
name = "Event Plugin"
|
|
|
|
def initialize(self):
|
|
self.api = get_api()
|
|
self.subscriptions = []
|
|
|
|
# Subscribe to events
|
|
sub_id = self.api.subscribe("loot", self.on_loot)
|
|
self.subscriptions.append(sub_id)
|
|
|
|
sub_id = self.api.subscribe("skill_gain", self.on_skill)
|
|
self.subscriptions.append(sub_id)
|
|
|
|
def shutdown(self):
|
|
# Unsubscribe from all
|
|
for sub_id in self.subscriptions:
|
|
self.api.unsubscribe(sub_id)
|
|
|
|
def get_ui(self):
|
|
widget = QWidget()
|
|
layout = QVBoxLayout(widget)
|
|
|
|
self.status_label = QLabel("Listening for events...")
|
|
layout.addWidget(self.status_label)
|
|
|
|
return widget
|
|
|
|
def on_loot(self, event):
|
|
data = event.data
|
|
self.status_label.setText(f"Got loot: {data}")
|
|
|
|
def on_skill(self, event):
|
|
data = event.data
|
|
self.status_label.setText(f"Skill gain: {data}")
|
|
```
|
|
|
|
### Plugin Configuration
|
|
|
|
```python
|
|
from plugins.base_plugin import BasePlugin
|
|
|
|
class ConfigurablePlugin(BasePlugin):
|
|
name = "Configurable Plugin"
|
|
|
|
def initialize(self):
|
|
# Access config passed to constructor
|
|
self.api = get_api()
|
|
|
|
# Get config with defaults
|
|
self.update_interval = self.config.get("update_interval", 1000)
|
|
self.auto_start = self.config.get("auto_start", True)
|
|
self.threshold = self.config.get("threshold", 0.5)
|
|
|
|
self.log_info(f"Update interval: {self.update_interval}ms")
|
|
|
|
def on_config_changed(self, key, value):
|
|
"""Called when config is updated."""
|
|
self.log_info(f"Config changed: {key} = {value}")
|
|
|
|
if key == "update_interval":
|
|
self.update_interval = value
|
|
```
|
|
|
|
### File Structure
|
|
|
|
```
|
|
plugins/
|
|
└── my_plugin/
|
|
├── __init__.py
|
|
├── plugin.py # Main plugin file
|
|
├── ui.py # UI components (optional)
|
|
├── utils.py # Helper functions (optional)
|
|
└── assets/ # Plugin assets (optional)
|
|
└── icon.png
|
|
```
|
|
|
|
---
|
|
|
|
## Best Practices
|
|
|
|
1. **Always use get_api()** - Don't create API instances directly
|
|
2. **Handle errors gracefully** - Use try/except for API calls
|
|
3. **Clean up in shutdown()** - Unsubscribe from events, save data
|
|
4. **Use background tasks** - Don't block the UI thread
|
|
5. **Log appropriately** - Use self.log_* methods
|
|
6. **Follow naming conventions** - Use descriptive names
|
|
7. **Document your plugin** - Add docstrings and README
|
|
|
|
---
|
|
|
|
For more examples, see `docs/API_COOKBOOK.md`
|