chore: prepare for premium architecture transformation
This commit is contained in:
parent
0e5a7148fd
commit
5e44355e52
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Reference in New Issue