EU-Utility/core/main.py

622 lines
26 KiB
Python

"""
EU-Utility - Main Entry Point
Launch the overlay, floating icon, dashboard, and plugin system.
"""
import sys
import os
import time
from pathlib import Path
# Add project root to path
project_root = Path(__file__).parent.parent
if str(project_root) not in sys.path:
sys.path.insert(0, str(project_root))
# Import debug logger FIRST (before anything else)
from core.debug_logger import debug_logger, timed, log_call
debug_logger.info("MAIN", "Starting EU-Utility...")
start_time = time.perf_counter()
try:
from PyQt6.QtWidgets import QApplication
from PyQt6.QtCore import Qt, QObject, pyqtSignal
PYQT_AVAILABLE = True
debug_logger.info("MAIN", "PyQt6 imported successfully")
except ImportError as e:
PYQT_AVAILABLE = False
debug_logger.error("MAIN", f"PyQt6 import failed: {e}")
print("Error: PyQt6 is required.")
print("Install with: pip install PyQt6")
sys.exit(1)
try:
import keyboard
KEYBOARD_AVAILABLE = True
debug_logger.info("MAIN", "keyboard library imported")
except ImportError:
KEYBOARD_AVAILABLE = False
debug_logger.warn("MAIN", "keyboard library not available")
print("Warning: 'keyboard' library not installed.")
print("Global hotkeys won't work. Install with: pip install keyboard")
debug_logger.start_timer("imports")
from core.plugin_manager import PluginManager
from core.perfect_ux import create_perfect_window
from core.tray_icon import TrayIcon
from core.settings import get_settings
from core.overlay_widgets import OverlayManager
from core.api import get_api, get_widget_api, get_external_api
from core.log_reader import get_log_reader
from core.ocr_service import get_ocr_service
from core.screenshot import get_screenshot_service
from core.notifications import get_notification_manager, NotificationManager
from core.nexus_api import get_nexus_api
from core.http_client import get_http_client
from core.window_manager import get_window_manager
from core.event_bus import get_event_bus
from core.tasks import get_task_manager
from core.audio import get_audio_manager
from core.clipboard import get_clipboard_manager
debug_logger.end_timer("imports")
from core.data_store import get_data_store
class HotkeyHandler(QObject):
"""Signal bridge for thread-safe hotkey handling."""
toggle_signal = pyqtSignal()
hide_overlays_signal = pyqtSignal()
class EUUtilityApp:
"""Main application controller."""
def __init__(self):
self.app = None
self.overlay = None
self.floating_icon = None
self.plugin_manager = None
self.hotkey_handler = None
self.settings = None
self.overlay_manager = None
self.api = None
self.notification_manager = None
def run(self):
"""Start the application."""
debug_logger.start_timer("MAIN_run_total")
# Create Qt Application
debug_logger.start_timer("MAIN_create_app")
self.app = QApplication(sys.argv)
self.app.setQuitOnLastWindowClosed(False)
debug_logger.end_timer("MAIN_create_app")
# Enable high DPI scaling (Qt6 has this enabled by default)
if hasattr(Qt, 'AA_EnableHighDpiScaling'):
try:
self.app.setAttribute(Qt.ApplicationAttribute.AA_EnableHighDpiScaling)
except (AttributeError, TypeError):
pass
# Initialize Plugin API
debug_logger.start_timer("MAIN_init_api")
print("Initializing Plugin API...")
self.api = get_api()
self._setup_api_services()
debug_logger.end_timer("MAIN_init_api")
# Initialize Event Bus
debug_logger.start_timer("MAIN_init_eventbus")
print("Initializing Event Bus...")
self.event_bus = get_event_bus()
self._print_event_bus_stats()
debug_logger.end_timer("MAIN_init_eventbus")
# Initialize Notification Manager
debug_logger.start_timer("MAIN_init_notifications")
print("Initializing Notification Manager...")
self.notification_manager = get_notification_manager()
self.notification_manager.initialize(self.app)
self.api.register_notification_service(self.notification_manager)
print("[Core] Notification service registered")
debug_logger.end_timer("MAIN_init_notifications")
# Load settings
debug_logger.start_timer("MAIN_load_settings")
self.settings = get_settings()
debug_logger.end_timer("MAIN_load_settings")
# Create hotkey handler (must be in main thread)
debug_logger.start_timer("MAIN_init_hotkeys")
self.hotkey_handler = HotkeyHandler()
debug_logger.end_timer("MAIN_init_hotkeys")
# Initialize plugin manager
debug_logger.start_timer("MAIN_load_plugins")
print("Loading plugins...")
self.plugin_manager = PluginManager(None)
self.plugin_manager.load_all_plugins()
debug_logger.end_timer("MAIN_load_plugins")
# Create overlay manager
debug_logger.start_timer("MAIN_init_overlay_mgr")
self.overlay_manager = OverlayManager(self.app)
debug_logger.end_timer("MAIN_init_overlay_mgr")
# Create perfect UX main window
debug_logger.start_timer("MAIN_create_window")
print("Creating main window with perfect UX...")
self.dashboard = create_perfect_window(self.plugin_manager)
self.plugin_manager.overlay = self.dashboard
self.dashboard.show()
debug_logger.end_timer("MAIN_create_window")
# Create system tray icon (replaces floating icon)
debug_logger.start_timer("MAIN_create_tray")
print("Creating system tray icon...")
self.tray_icon = TrayIcon(self.app)
self.tray_icon.show_dashboard.connect(self._toggle_overlay)
self.tray_icon.toggle_activity_bar.connect(self._toggle_activity_bar)
self.tray_icon.quit_app.connect(self.quit)
debug_logger.end_timer("MAIN_create_tray")
# Create Activity Bar (in-game overlay) - hidden by default
debug_logger.start_timer("MAIN_create_activitybar")
print("Creating Activity Bar...")
try:
from core.activity_bar import get_activity_bar
self.activity_bar = get_activity_bar(self.plugin_manager)
if self.activity_bar:
if self.activity_bar.config.enabled:
print("[Core] Activity Bar created (will show when EU is focused)")
# Connect signals
self.activity_bar.widget_requested.connect(self._on_activity_bar_widget)
# Start EU focus detection
self._start_eu_focus_detection()
else:
print("[Core] Activity Bar disabled in config")
else:
print("[Core] Activity Bar not available")
self.activity_bar = None
except Exception as e:
print(f"[Core] Failed to create Activity Bar: {e}")
self.activity_bar = None
debug_logger.end_timer("MAIN_create_activitybar")
# Connect hotkey signals
self.hotkey_handler.toggle_signal.connect(self._on_toggle_signal)
# Setup global hotkeys
debug_logger.start_timer("MAIN_setup_hotkeys")
self._setup_hotkeys()
debug_logger.end_timer("MAIN_setup_hotkeys")
# Load saved overlay widgets
debug_logger.start_timer("MAIN_load_widgets")
self._load_overlay_widgets()
debug_logger.end_timer("MAIN_load_widgets")
# Log total startup time
total_startup = debug_logger.end_timer("MAIN_run_total")
debug_logger.info("MAIN", f"=== TOTAL STARTUP TIME: {total_startup:.2f}ms ===")
print("EU-Utility started!")
print("Dashboard window is open")
print("Press Ctrl+Shift+U to toggle dashboard")
print("Press Ctrl+Shift+H to hide all overlays")
print("Press Ctrl+Shift+B to toggle activity bar")
print(f"Loaded {len(self.plugin_manager.get_all_plugins())} plugins")
# Show Event Bus stats
self._print_event_bus_stats()
# Run
debug_logger.info("MAIN", "Starting Qt event loop...")
return self.app.exec()
def _setup_api_services(self):
"""Setup shared API services - Window, OCR and Log are core services."""
# Initialize and start Log Reader
print("[Core] Initializing Log Reader...")
self.log_reader = get_log_reader()
if self.log_reader.start():
print(f"[Core] Log Reader started - watching: {self.log_reader.log_path}")
else:
print("[Core] Log Reader not available - log file not found")
# Register Log service with API
self.api.register_log_service(self.log_reader.read_lines)
# Initialize Window Manager (Windows-only, gracefully handles Linux)
print("[Core] Initializing Window Manager...")
self.window_manager = get_window_manager()
if self.window_manager.is_available():
# Try to find EU window on startup
eu_window = self.window_manager.find_eu_window()
if eu_window:
print(f"[Core] Found EU window: {eu_window.title} ({eu_window.width}x{eu_window.height})")
else:
print("[Core] EU window not found - will retry when needed")
# Register Window service with API
self.api.register_window_service(self.window_manager)
else:
print("[Core] Window Manager not available (Windows only)")
# Screenshot Service - Initialize on startup (lightweight)
print("[Core] Initializing Screenshot Service...")
self.screenshot_service = get_screenshot_service()
if self.screenshot_service.is_available():
self.api.register_screenshot_service(self.screenshot_service)
backends = self.screenshot_service.get_available_backends()
print(f"[Core] Screenshot Service ready (backends: {', '.join(backends)})")
else:
print("[Core] Screenshot Service not available - install pillow or pyautogui")
# OCR Service - LAZY INITIALIZATION (don't init on startup)
# It will initialize on first use
print("[Core] OCR Service configured (lazy init)")
self.ocr_service = get_ocr_service()
# Register OCR service with API (lazy - will init on first call)
self.api.register_ocr_service(self._lazy_ocr_handler)
# Initialize Nexus API Service
print("[Core] Initializing Nexus API Service...")
self.nexus_api = get_nexus_api()
self.api.register_nexus_service(self.nexus_api)
# HTTP Client - Initialize on startup
print("[Core] Initializing HTTP Client...")
try:
self.http_client = get_http_client(
cache_dir="cache/http",
default_cache_ttl=3600,
rate_limit_delay=0.1, # Small delay between requests
max_retries=3,
backoff_factor=0.5,
respect_cache_control=True
)
self.api.register_http_service(self.http_client)
print("[Core] HTTP Client initialized with caching")
except Exception as e:
print(f"[Core] HTTP Client initialization failed: {e}")
# Initialize Audio Service
print("[Core] Initializing Audio Service...")
self.audio_manager = get_audio_manager()
if self.audio_manager.is_available():
self.api.register_audio_service(self.audio_manager)
backend = self.audio_manager.get_backend()
volume = int(self.audio_manager.get_volume() * 100)
print(f"[Core] Audio Service ready (backend: {backend}, volume: {volume}%)")
else:
print("[Core] Audio Service not available - no audio backend found")
# Initialize Task Manager
print("[Core] Initializing Task Manager...")
try:
self.task_manager = get_task_manager(max_workers=4)
self.task_manager.initialize()
self.api.register_task_service(self.task_manager)
print("[Core] Task Manager initialized with 4 workers")
except Exception as e:
print(f"[Core] Task Manager initialization failed: {e}")
# Initialize Clipboard Manager
print("[Core] Initializing Clipboard Manager...")
self.clipboard_manager = get_clipboard_manager()
if self.clipboard_manager.is_available():
self.api.register_clipboard_service(self.clipboard_manager)
print("[Core] Clipboard Service ready")
else:
print("[Core] Clipboard Service not available - install pyperclip")
# Initialize Data Store
print("[Core] Initializing Data Store...")
self.data_store = get_data_store()
self.api.register_data_service(self.data_store)
print("[Core] Data Store ready")
print("[Core] API services registered: Window, Screenshot, OCR (lazy), Log, Nexus, HTTP, Audio, Tasks, Clipboard, Data")
def _lazy_ocr_handler(self, region=None):
"""Lazy OCR handler - triggers init on first use."""
return self.ocr_service.recognize(region=region)
def _print_event_bus_stats(self):
"""Print Event Bus statistics on startup."""
if not hasattr(self, 'event_bus') or not self.event_bus:
return
stats = self.event_bus.get_stats()
print("\n" + "=" * 50)
print("📊 Event Bus Statistics")
print("=" * 50)
print(f" Total Events Published: {stats.get('total_published', 0)}")
print(f" Total Events Delivered: {stats.get('total_delivered', 0)}")
print(f" Active Subscriptions: {stats.get('active_subscriptions', 0)}")
print(f" Events Per Minute: {stats.get('events_per_minute', 0)}")
print(f" Avg Delivery Time: {stats.get('avg_delivery_ms', 0)} ms")
print(f" Errors: {stats.get('errors', 0)}")
top_types = stats.get('top_event_types', {})
if top_types:
print(f"\n Top Event Types:")
for event_type, count in list(top_types.items())[:5]:
print(f"{event_type}: {count}")
print("=" * 50 + "\n")
def _setup_hotkeys(self):
"""Setup global hotkeys."""
if KEYBOARD_AVAILABLE:
try:
# Toggle main overlay
keyboard.add_hotkey('ctrl+shift+u', self._on_hotkey_pressed)
# Hide all overlays
keyboard.add_hotkey('ctrl+shift+h', self._on_hide_overlays_pressed)
# Toggle activity bar
keyboard.add_hotkey('ctrl+shift+b', self._on_activity_bar_hotkey)
except Exception as e:
print(f"Failed to register hotkey: {e}")
def _on_hotkey_pressed(self):
"""Called when toggle hotkey is pressed."""
if self.hotkey_handler:
self.hotkey_handler.toggle_signal.emit()
def _on_hide_overlays_pressed(self):
"""Called when hide hotkey is pressed."""
if self.overlay_manager:
self.overlay_manager.hide_all()
def _on_activity_bar_hotkey(self):
"""Called when activity bar hotkey is pressed."""
self._toggle_activity_bar()
def _on_toggle_signal(self):
"""Handle toggle signal in main thread."""
self._toggle_overlay()
def _on_activity_bar_widget(self, plugin_id: str):
"""Handle activity bar widget request."""
print(f"[Main] Activity bar requested plugin: {plugin_id}")
# Get plugin class
all_plugins = self.plugin_manager.get_all_discovered_plugins()
if plugin_id not in all_plugins:
print(f"[Main] Plugin not found: {plugin_id}")
return
plugin_class = all_plugins[plugin_id]
# Check if plugin has a mini widget
if hasattr(plugin_class, 'get_mini_widget'):
try:
# Create mini widget
widget = plugin_class.get_mini_widget()
if widget:
widget.show()
print(f"[Main] Created mini widget for {plugin_class.name}")
except Exception as e:
print(f"[Main] Error creating mini widget: {e}")
else:
# No mini widget, try to show main dashboard
self._toggle_overlay()
# Switch to this plugin
if self.dashboard:
self.dashboard.show_plugin(plugin_id)
def _toggle_overlay(self):
"""Toggle dashboard visibility."""
debug_logger.start_timer("MAIN_toggle_overlay")
debug_logger.debug("MAIN", f"_toggle_overlay called, dashboard exists: {self.dashboard is not None}")
if self.dashboard:
if self.dashboard.isVisible():
debug_logger.debug("MAIN", "Hiding dashboard...")
self.dashboard.hide()
debug_logger.debug("MAIN", "Dashboard hidden")
else:
debug_logger.debug("MAIN", "Showing dashboard...")
self.dashboard.show()
debug_logger.debug("MAIN", "Dashboard shown, calling raise_...")
self.dashboard.raise_()
debug_logger.debug("MAIN", "Calling activateWindow...")
self.dashboard.activateWindow()
debug_logger.debug("MAIN", "Dashboard activation complete")
else:
debug_logger.warn("MAIN", "Dashboard is None!")
debug_logger.end_timer("MAIN_toggle_overlay")
def _toggle_activity_bar(self):
"""Toggle activity bar visibility."""
debug_logger.start_timer("MAIN_toggle_activity_bar")
debug_logger.debug("MAIN", f"_toggle_activity_bar called, activity_bar exists: {hasattr(self, 'activity_bar')}")
if hasattr(self, 'activity_bar') and self.activity_bar:
try:
if self.activity_bar.isVisible():
debug_logger.debug("MAIN", "Hiding activity bar...")
self.activity_bar.hide()
if hasattr(self, 'tray_icon') and self.tray_icon:
self.tray_icon.set_activity_bar_checked(False)
debug_logger.debug("MAIN", "Activity bar hidden")
else:
debug_logger.debug("MAIN", "Showing activity bar...")
self.activity_bar.show()
if hasattr(self, 'tray_icon') and self.tray_icon:
self.tray_icon.set_activity_bar_checked(True)
debug_logger.debug("MAIN", "Activity bar shown")
except Exception as e:
debug_logger.error("MAIN", f"Error toggling activity bar: {e}")
else:
debug_logger.warn("MAIN", "Activity Bar not available")
debug_logger.end_timer("MAIN_toggle_activity_bar")
def _start_eu_focus_detection(self):
"""Start timer to detect EU window focus and show/hide activity bar."""
from PyQt6.QtCore import QTimer
debug_logger.info("MAIN", "Starting EU focus detection...")
debug_logger.start_timer("MAIN_start_eu_focus")
self.eu_focus_timer = QTimer(self.app)
self.eu_focus_timer.timeout.connect(self._check_eu_focus)
self.eu_focus_timer.start(5000) # Check every 5 seconds (reduced from 2s for debugging)
self._last_eu_focused = False
debug_logger.end_timer("MAIN_start_eu_focus")
debug_logger.info("MAIN", "EU focus detection started (5s interval)")
def _check_eu_focus(self):
"""Check if EU window is focused and show/hide activity bar."""
debug_logger.start_timer("MAIN_check_eu_focus")
try:
if not hasattr(self, 'activity_bar') or not self.activity_bar:
debug_logger.debug("MAIN", "No activity bar, skipping focus check")
return
if not hasattr(self, 'window_manager') or not self.window_manager:
debug_logger.debug("MAIN", "No window manager, skipping focus check")
return
if not self.window_manager.is_available():
debug_logger.debug("MAIN", "Window manager not available, skipping")
return
debug_logger.debug("MAIN", "Finding EU window...")
eu_window = self.window_manager.find_eu_window()
if eu_window:
debug_logger.debug("MAIN", "EU window found, checking focus...")
is_focused = eu_window.is_focused()
debug_logger.debug("MAIN", f"EU focused: {is_focused}, last: {getattr(self, '_last_eu_focused', None)}")
if is_focused != getattr(self, '_last_eu_focused', False):
self._last_eu_focused = is_focused
if is_focused:
# EU just got focused - show activity bar
if not self.activity_bar.isVisible():
try:
debug_logger.debug("MAIN", "Showing activity bar (EU focused)")
self.activity_bar.show()
if hasattr(self, 'tray_icon') and self.tray_icon:
self.tray_icon.set_activity_bar_checked(True)
debug_logger.info("MAIN", "EU focused - Activity Bar shown")
except Exception as e:
debug_logger.error("MAIN", f"Error showing activity bar: {e}")
else:
# EU lost focus - hide activity bar
if self.activity_bar.isVisible():
try:
debug_logger.debug("MAIN", "Hiding activity bar (EU unfocused)")
self.activity_bar.hide()
if hasattr(self, 'tray_icon') and self.tray_icon:
self.tray_icon.set_activity_bar_checked(False)
debug_logger.info("MAIN", "EU unfocused - Activity Bar hidden")
except Exception as e:
debug_logger.error("MAIN", f"Error hiding activity bar: {e}")
# Reset fail count since we found EU
self._eu_not_found_count = 0
else:
debug_logger.debug("MAIN", "EU window not found")
# Track consecutive failures
self._eu_not_found_count = getattr(self, '_eu_not_found_count', 0) + 1
# If EU not found for 3 consecutive checks (15 seconds), slow down polling
if self._eu_not_found_count >= 3:
debug_logger.warn("MAIN", f"EU not found {self._eu_not_found_count} times - slowing down focus detection")
# Slow down to once per minute when EU isn't running
if hasattr(self, 'eu_focus_timer'):
self.eu_focus_timer.stop()
self.eu_focus_timer.start(60000) # 60 seconds
debug_logger.info("MAIN", "EU focus detection slowed to 60s interval")
except Exception as e:
debug_logger.error("MAIN", f"Error in EU focus check: {e}")
elapsed = debug_logger.end_timer("MAIN_check_eu_focus")
if elapsed > 100: # Log if taking more than 100ms
debug_logger.warn("MAIN", f"EU focus check took {elapsed:.2f}ms - SLOW!")
def _load_overlay_widgets(self):
"""Load saved overlay widgets."""
widget_settings = self.settings.get('overlay_widgets', {})
for widget_name, config in widget_settings.items():
if config.get('enabled', False):
try:
self.overlay_manager.create_widget(
widget_name.replace('_overlay', ''),
widget_name
)
except Exception as e:
print(f"Failed to load overlay widget {widget_name}: {e}")
def create_overlay_widget(self, widget_type, name):
"""Create an overlay widget."""
if self.overlay_manager:
return self.overlay_manager.create_widget(widget_type, name)
return None
def quit(self):
"""Quit the application."""
print("[Core] Shutting down...")
# Stop EU focus timer
if hasattr(self, 'eu_focus_timer'):
self.eu_focus_timer.stop()
# Stop log reader
if hasattr(self, 'log_reader'):
self.log_reader.stop()
# Close all notifications
if self.notification_manager:
self.notification_manager.close_all()
# Shutdown Event Bus
if hasattr(self, 'event_bus') and self.event_bus:
print("[Core] Shutting down Event Bus...")
self.event_bus.shutdown()
# Shutdown Audio
if hasattr(self, 'audio_manager') and self.audio_manager:
print("[Core] Shutting down Audio...")
self.audio_manager.shutdown()
# Shutdown Task Manager
if hasattr(self, 'task_manager') and self.task_manager:
print("[Core] Shutting down Task Manager...")
self.task_manager.shutdown(wait=True, timeout=30.0)
# Window manager has no persistent resources to clean up
if self.overlay_manager:
self.overlay_manager.hide_all()
if self.plugin_manager:
self.plugin_manager.shutdown_all()
self.app.quit()
def main():
"""Entry point."""
app = EUUtilityApp()
sys.exit(app.run())
if __name__ == "__main__":
main()