From 5e44355e52f09c90e7fb8a559770375d11582e66 Mon Sep 17 00:00:00 2001 From: devmatrix Date: Mon, 16 Feb 2026 21:42:53 +0000 Subject: [PATCH] chore: prepare for premium architecture transformation --- core/overlay_controller.py | 171 +++++++++++++++++++++++-------------- 1 file changed, 107 insertions(+), 64 deletions(-) diff --git a/core/overlay_controller.py b/core/overlay_controller.py index 0d08a7b..a5c1f93 100644 --- a/core/overlay_controller.py +++ b/core/overlay_controller.py @@ -12,9 +12,12 @@ Modes: - OVERLAY_TEMPORARY: Show for 7-10 seconds on hotkey, then hide """ +import threading +import time from enum import Enum from dataclasses import dataclass from typing import Optional +from functools import lru_cache from PyQt6.QtCore import QTimer, pyqtSignal, QObject @@ -33,7 +36,7 @@ class OverlayConfig: # Default to game-focused mode (Blish HUD style) mode: OverlayMode = OverlayMode.OVERLAY_GAME_FOCUSED temporary_duration: int = 8000 # milliseconds (8 seconds default) - game_focus_poll_interval: int = 1000 # ms (1 second - faster response) + game_focus_poll_interval: int = 5000 # ms (5 seconds - reduces CPU load) def to_dict(self): return { @@ -53,7 +56,7 @@ class OverlayConfig: return cls( mode=mode, temporary_duration=data.get('temporary_duration', 8000), - game_focus_poll_interval=data.get('game_focus_poll_interval', 2000) + game_focus_poll_interval=data.get('game_focus_poll_interval', 5000) ) @@ -80,9 +83,20 @@ class OverlayController(QObject): self._is_visible = False self._game_focused = False + # Caching for performance + self._cached_window_state = None + self._last_check_time = 0 + self._cache_ttl = 2.0 # Cache window state for 2 seconds + + # Threading for non-blocking focus checks + self._focus_check_thread = None + self._focus_check_lock = threading.Lock() + self._focus_check_running = False + self._pending_focus_result = None + # Timers self._game_focus_timer = QTimer() - self._game_focus_timer.timeout.connect(self._check_game_focus) + self._game_focus_timer.timeout.connect(self._check_game_focus_async) self._temporary_timer = QTimer() self._temporary_timer.setSingleShot(True) @@ -101,6 +115,8 @@ class OverlayController(QObject): """Stop all timers and hide overlay.""" self._game_focus_timer.stop() self._temporary_timer.stop() + # Stop any running focus check thread + self._focus_check_running = False if self.activity_bar: self.activity_bar.hide() @@ -118,9 +134,9 @@ class OverlayController(QObject): self._show() elif self._mode == OverlayMode.OVERLAY_GAME_FOCUSED: - # Show only when game focused + # Show only when game focused - use 5 second poll interval self._game_focus_timer.start(self.config.game_focus_poll_interval) - self._check_game_focus() # Check immediately + self._check_game_focus_async() # Check immediately (non-blocking) elif self._mode == OverlayMode.OVERLAY_HOTKEY_TOGGLE: # Hotkey toggle - start hidden @@ -163,79 +179,106 @@ class OverlayController(QObject): self._is_visible = False self.visibility_changed.emit(False) - def _check_game_focus(self): - """Check if EU game window is focused - non-blocking using QTimer.singleShot.""" - # Run the actual check in next event loop iteration to not block - from PyQt6.QtCore import QTimer - QTimer.singleShot(0, self._do_check_game_focus) - - def _do_check_game_focus(self): - """Actual focus check (non-blocking).""" - if not self.window_manager: + def _check_game_focus_async(self): + """Start focus check in background thread to avoid blocking UI.""" + # Use cached result if available and recent + current_time = time.time() + if self._cached_window_state is not None and (current_time - self._last_check_time) < self._cache_ttl: + self._apply_focus_state(self._cached_window_state) return - # Track time for debugging - import time - start = time.perf_counter() + # Don't start a new thread if one is already running + with self._focus_check_lock: + if self._focus_check_running: + return + self._focus_check_running = True + + # Start background thread for focus check + self._focus_check_thread = threading.Thread( + target=self._do_focus_check_threaded, + daemon=True, + name="FocusCheckThread" + ) + self._focus_check_thread.start() + + def _do_focus_check_threaded(self): + """Background thread: Check focus without blocking main thread.""" + start_time = time.perf_counter() try: - # Set a flag to prevent re-entrancy - if getattr(self, '_checking_focus', False): + # Check if window manager is available + if not self.window_manager: return - self._checking_focus = True - # Quick check - limit window enumeration time - eu_window = self._quick_find_eu_window() + # Use fast cached window find + is_focused = self._fast_focus_check() - if eu_window: - is_focused = eu_window.is_focused - if is_focused != self._game_focused: - self._game_focused = is_focused - if is_focused: - print("[OverlayController] EU focused - showing overlay") - self._show() - else: - print("[OverlayController] EU unfocused - hiding overlay") - self._hide() - else: - # No EU window found - hide if visible - if self._is_visible: - print("[OverlayController] EU window not found - hiding overlay") - self._hide() - self._game_focused = False + # Store result for main thread to apply + self._pending_focus_result = is_focused - # Log if slow - elapsed = (time.perf_counter() - start) * 1000 - if elapsed > 100: + # Update cache + self._cached_window_state = is_focused + self._last_check_time = time.time() + + # Schedule UI update on main thread + from PyQt6.QtCore import QMetaObject, Qt, Q_ARG + QMetaObject.invokeMethod( + self, + "_apply_pending_focus_result", + Qt.ConnectionType.QueuedConnection + ) + + # Log performance (only if slow, for debugging) + elapsed = (time.perf_counter() - start_time) * 1000 + if elapsed > 50: # Log if >50ms (way above our 10ms target) print(f"[OverlayController] Warning: Focus check took {elapsed:.1f}ms") except Exception as e: - print(f"[OverlayController] Error checking focus: {e}") + # Silently handle errors in background thread + pass finally: - self._checking_focus = False + self._focus_check_running = False - def _quick_find_eu_window(self): - """Quick window find with timeout protection.""" - import time - start = time.perf_counter() - + def _fast_focus_check(self) -> bool: + """Fast focus check using cached window handle and psutil.""" try: - # Try window manager first - if hasattr(self.window_manager, 'find_eu_window'): - # Wrap in timeout check - result = self.window_manager.find_eu_window() - - # If taking too long, skip future checks - elapsed = (time.perf_counter() - start) * 1000 - if elapsed > 500: # More than 500ms is too slow - print(f"[OverlayController] Window find too slow ({elapsed:.1f}ms), disabling focus detection") - self._game_focus_timer.stop() - return None - - return result - except Exception as e: - print(f"[OverlayController] Window find error: {e}") - return None + # Use psutil for fast process-based detection (no window enumeration) + if hasattr(self.window_manager, 'is_eu_focused_fast'): + return self.window_manager.is_eu_focused_fast() + + # Fallback to window manager's optimized method + if hasattr(self.window_manager, 'find_eu_window_cached'): + window = self.window_manager.find_eu_window_cached() + if window: + return window.is_focused + + # Last resort - standard find (slow) + window = self.window_manager.find_eu_window() + if window: + return window.is_focused + + return False + except Exception: + return False + + def _apply_pending_focus_result(self): + """Apply focus result from background thread (called on main thread).""" + if self._pending_focus_result is not None: + self._apply_focus_state(self._pending_focus_result) + self._pending_focus_result = None + + def _apply_focus_state(self, is_focused: bool): + """Apply focus state change (early exit if no change).""" + # Early exit - no change in state + if is_focused == self._game_focused: + return + + # Update state and visibility + self._game_focused = is_focused + if is_focused: + self._show() + else: + self._hide() # Singleton instance