444 lines
16 KiB
Python
444 lines
16 KiB
Python
"""
|
|
EU-Utility - Performance Optimized Main Entry Point
|
|
|
|
Integrates all performance optimizations:
|
|
- Startup profiling
|
|
- Optimized LogWatcher
|
|
- Optimized OCR Service
|
|
- Memory leak detection
|
|
- UI rendering at 60+ FPS
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
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))
|
|
|
|
# Performance imports
|
|
from core.startup_profiler import get_startup_profiler
|
|
from core.memory_leak_detector import start_leak_detection, MemoryOptimizer
|
|
|
|
profiler = get_startup_profiler()
|
|
|
|
try:
|
|
with profiler.profile("pyqt_import"):
|
|
from PyQt6.QtWidgets import QApplication
|
|
from PyQt6.QtCore import Qt, QObject, pyqtSignal
|
|
PYQT_AVAILABLE = True
|
|
except ImportError:
|
|
PYQT_AVAILABLE = False
|
|
print("Error: PyQt6 is required.")
|
|
sys.exit(1)
|
|
|
|
try:
|
|
with profiler.profile("keyboard_import"):
|
|
import keyboard
|
|
KEYBOARD_AVAILABLE = True
|
|
except ImportError:
|
|
KEYBOARD_AVAILABLE = False
|
|
print("Warning: 'keyboard' library not installed.")
|
|
|
|
# Core imports with profiling
|
|
with profiler.profile("core_imports"):
|
|
from core.plugin_manager import PluginManager
|
|
from core.overlay_window import OverlayWindow
|
|
from core.floating_icon import FloatingIcon
|
|
from core.settings import get_settings
|
|
from core.overlay_widgets import OverlayManager
|
|
from core.plugin_api import get_api, APIType
|
|
|
|
# Use optimized components
|
|
try:
|
|
from core.log_watcher_optimized import get_log_watcher
|
|
USE_OPTIMIZED_LOG = True
|
|
except ImportError:
|
|
from core.log_reader import get_log_reader
|
|
USE_OPTIMIZED_LOG = False
|
|
print("[Performance] Using standard LogReader (optimized version not available)")
|
|
|
|
try:
|
|
from core.ocr_service_optimized_v2 import get_ocr_service
|
|
USE_OPTIMIZED_OCR = True
|
|
except ImportError:
|
|
from core.ocr_service import get_ocr_service
|
|
USE_OPTIMIZED_OCR = False
|
|
print("[Performance] Using standard OCR (optimized version not available)")
|
|
|
|
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
|
|
from core.data_store import get_data_store
|
|
|
|
# UI optimizations
|
|
try:
|
|
from core.ui_render_optimized import set_target_fps
|
|
USE_OPTIMIZED_UI = True
|
|
except ImportError:
|
|
USE_OPTIMIZED_UI = False
|
|
|
|
|
|
class HotkeyHandler(QObject):
|
|
"""Signal bridge for thread-safe hotkey handling."""
|
|
toggle_signal = pyqtSignal()
|
|
hide_overlays_signal = pyqtSignal()
|
|
|
|
|
|
class PerformanceOptimizedApp:
|
|
"""
|
|
Main application controller with performance optimizations.
|
|
"""
|
|
|
|
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
|
|
self.leak_detector = None
|
|
|
|
def run(self):
|
|
"""Start the application with profiling."""
|
|
profiler = get_startup_profiler()
|
|
|
|
with profiler.profile("qt_app_creation"):
|
|
# Create Qt Application
|
|
self.app = QApplication(sys.argv)
|
|
self.app.setQuitOnLastWindowClosed(False)
|
|
|
|
# Enable high DPI scaling
|
|
if hasattr(Qt, 'AA_EnableHighDpiScaling'):
|
|
self.app.setAttribute(Qt.ApplicationAttribute.AA_EnableHighDpiScaling)
|
|
|
|
with profiler.profile("plugin_api_init"):
|
|
# Initialize Plugin API
|
|
print("[Performance] Initializing Plugin API...")
|
|
self.api = get_api()
|
|
self._setup_api_services()
|
|
|
|
with profiler.profile("event_bus_init"):
|
|
# Initialize Event Bus
|
|
print("[Performance] Initializing Event Bus...")
|
|
self.event_bus = get_event_bus()
|
|
self._print_event_bus_stats()
|
|
|
|
with profiler.profile("notification_init"):
|
|
# Initialize Notification Manager
|
|
print("[Performance] Initializing Notification Manager...")
|
|
self.notification_manager = get_notification_manager()
|
|
self.notification_manager.initialize(self.app)
|
|
self.api.register_notification_service(self.notification_manager)
|
|
|
|
with profiler.profile("settings_load"):
|
|
# Load settings
|
|
self.settings = get_settings()
|
|
|
|
with profiler.profile("hotkey_setup"):
|
|
# Create hotkey handler (must be in main thread)
|
|
self.hotkey_handler = HotkeyHandler()
|
|
|
|
with profiler.profile("plugin_loading"):
|
|
# Initialize plugin manager
|
|
print("[Performance] Loading plugins...")
|
|
self.plugin_manager = PluginManager(None)
|
|
self.plugin_manager.load_all_plugins()
|
|
|
|
with profiler.profile("ui_creation"):
|
|
# Create overlay manager
|
|
self.overlay_manager = OverlayManager(self.app)
|
|
|
|
# Create overlay window
|
|
self.overlay = OverlayWindow(self.plugin_manager)
|
|
self.plugin_manager.overlay = self.overlay
|
|
|
|
# Create floating icon
|
|
print("[Performance] Creating floating icon...")
|
|
self.floating_icon = FloatingIcon()
|
|
self.floating_icon.clicked.connect(self._toggle_overlay)
|
|
self.floating_icon.show()
|
|
|
|
# Set target FPS for UI
|
|
if USE_OPTIMIZED_UI:
|
|
set_target_fps(60.0)
|
|
|
|
with profiler.profile("connections_setup"):
|
|
# Connect hotkey signals
|
|
self.hotkey_handler.toggle_signal.connect(self._on_toggle_signal)
|
|
|
|
# Setup global hotkeys
|
|
self._setup_hotkeys()
|
|
|
|
# Load saved overlay widgets
|
|
self._load_overlay_widgets()
|
|
|
|
with profiler.profile("leak_detection_start"):
|
|
# Start memory leak detection
|
|
self.leak_detector = start_leak_detection(check_interval=300.0) # 5 minutes
|
|
|
|
# Print startup report
|
|
profiler.print_report()
|
|
|
|
print("\n" + "=" * 60)
|
|
print("EU-Utility Started!")
|
|
print("=" * 60)
|
|
print("Press Ctrl+Shift+U to toggle overlay")
|
|
print("Press Ctrl+Shift+H to hide all overlays")
|
|
print("Or double-click the floating icon")
|
|
print(f"Loaded {len(self.plugin_manager.get_all_plugins())} plugins")
|
|
print(f"Optimizations enabled:")
|
|
print(f" - Optimized LogWatcher: {USE_OPTIMIZED_LOG}")
|
|
print(f" - Optimized OCR: {USE_OPTIMIZED_OCR}")
|
|
print(f" - Optimized UI Rendering: {USE_OPTIMIZED_UI}")
|
|
print(f" - Memory Leak Detection: Active")
|
|
print("=" * 60 + "\n")
|
|
|
|
# Show Event Bus stats
|
|
self._print_event_bus_stats()
|
|
|
|
# Run
|
|
return self.app.exec()
|
|
|
|
def _setup_api_services(self):
|
|
"""Setup shared API services with performance optimizations."""
|
|
|
|
# Initialize and start Log Reader (Optimized)
|
|
print("[Performance] Initializing Log Reader...")
|
|
if USE_OPTIMIZED_LOG:
|
|
self.log_reader = get_log_watcher()
|
|
else:
|
|
self.log_reader = get_log_reader()
|
|
|
|
if self.log_reader.start():
|
|
print(f"[Performance] Log Reader started - watching: {self.log_reader.log_path}")
|
|
else:
|
|
print("[Performance] Log Reader not available - log file not found")
|
|
|
|
self.api.register_log_service(self.log_reader.read_lines)
|
|
|
|
# Initialize Window Manager
|
|
print("[Performance] Initializing Window Manager...")
|
|
self.window_manager = get_window_manager()
|
|
if self.window_manager.is_available():
|
|
eu_window = self.window_manager.find_eu_window()
|
|
if eu_window:
|
|
print(f"[Performance] Found EU window: {eu_window.title} ({eu_window.width}x{eu_window.height})")
|
|
else:
|
|
print("[Performance] EU window not found - will retry when needed")
|
|
self.api.register_window_service(self.window_manager)
|
|
else:
|
|
print("[Performance] Window Manager not available (Windows only)")
|
|
|
|
# Screenshot Service - Initialize on startup
|
|
print("[Performance] 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"[Performance] Screenshot Service ready (backends: {', '.join(backends)})")
|
|
else:
|
|
print("[Performance] Screenshot Service not available")
|
|
|
|
# OCR Service - LAZY INITIALIZATION
|
|
print("[Performance] OCR Service configured (lazy init)")
|
|
self.ocr_service = get_ocr_service(background_init=True)
|
|
self.api.register_ocr_service(self._lazy_ocr_handler)
|
|
|
|
# Initialize Nexus API Service
|
|
print("[Performance] Initializing Nexus API Service...")
|
|
self.nexus_api = get_nexus_api()
|
|
self.api.register_nexus_service(self.nexus_api)
|
|
|
|
# HTTP Client - Initialize with caching
|
|
print("[Performance] Initializing HTTP Client...")
|
|
try:
|
|
self.http_client = get_http_client(
|
|
cache_dir="cache/http",
|
|
default_cache_ttl=3600,
|
|
rate_limit_delay=0.1,
|
|
max_retries=3,
|
|
backoff_factor=0.5,
|
|
respect_cache_control=True
|
|
)
|
|
self.api.register_http_service(self.http_client)
|
|
print("[Performance] HTTP Client initialized with caching")
|
|
except Exception as e:
|
|
print(f"[Performance] HTTP Client initialization failed: {e}")
|
|
|
|
# Initialize Audio Service
|
|
print("[Performance] 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"[Performance] Audio Service ready (backend: {backend}, volume: {volume}%)")
|
|
else:
|
|
print("[Performance] Audio Service not available")
|
|
|
|
# Initialize Task Manager
|
|
print("[Performance] 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("[Performance] Task Manager initialized with 4 workers")
|
|
except Exception as e:
|
|
print(f"[Performance] Task Manager initialization failed: {e}")
|
|
|
|
# Initialize Clipboard Manager
|
|
print("[Performance] Initializing Clipboard Manager...")
|
|
self.clipboard_manager = get_clipboard_manager()
|
|
if self.clipboard_manager.is_available():
|
|
self.api.register_clipboard_service(self.clipboard_manager)
|
|
print("[Performance] Clipboard Service ready")
|
|
else:
|
|
print("[Performance] Clipboard Service not available")
|
|
|
|
# Initialize Data Store
|
|
print("[Performance] Initializing Data Store...")
|
|
self.data_store = get_data_store()
|
|
self.api.register_data_service(self.data_store)
|
|
print("[Performance] Data Store ready")
|
|
|
|
print("[Performance] All API services registered")
|
|
|
|
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:
|
|
keyboard.add_hotkey('ctrl+shift+u', self._on_hotkey_pressed)
|
|
keyboard.add_hotkey('ctrl+shift+h', self._on_hide_overlays_pressed)
|
|
except Exception as e:
|
|
print(f"[Performance] 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_toggle_signal(self):
|
|
"""Handle toggle signal in main thread."""
|
|
self._toggle_overlay()
|
|
|
|
def _toggle_overlay(self):
|
|
"""Toggle overlay visibility."""
|
|
if self.overlay:
|
|
self.overlay.toggle_overlay()
|
|
|
|
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"[Performance] 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 with cleanup."""
|
|
print("[Performance] Shutting down...")
|
|
|
|
# Print final memory report
|
|
MemoryOptimizer.print_memory_summary()
|
|
|
|
# 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("[Performance] Shutting down Event Bus...")
|
|
self.event_bus.shutdown()
|
|
|
|
# Shutdown Audio
|
|
if hasattr(self, 'audio_manager') and self.audio_manager:
|
|
print("[Performance] Shutting down Audio...")
|
|
self.audio_manager.shutdown()
|
|
|
|
# Shutdown Task Manager
|
|
if hasattr(self, 'task_manager') and self.task_manager:
|
|
print("[Performance] Shutting down Task Manager...")
|
|
self.task_manager.shutdown(wait=True, timeout=30.0)
|
|
|
|
# Shutdown leak detector
|
|
if self.leak_detector:
|
|
self.leak_detector.stop()
|
|
|
|
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 = PerformanceOptimizedApp()
|
|
sys.exit(app.run())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|