349 lines
11 KiB
Python
349 lines
11 KiB
Python
"""
|
|
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 (thread-safe)
|
|
if self.event_bus:
|
|
try:
|
|
self.event_bus.emit(f"game.{event_type}", data, source="game_client")
|
|
except RuntimeError as e:
|
|
# No event loop in this thread - event is queued for main thread
|
|
# This is expected when emitting from background threads
|
|
pass
|
|
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"
|