-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathmain.py
More file actions
297 lines (243 loc) · 11.6 KB
/
Copy pathmain.py
File metadata and controls
297 lines (243 loc) · 11.6 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
import sys
import argparse
import ctypes
from PyQt6.QtWidgets import QApplication
from PyQt6.QtCore import Qt
from src.service_container import ServiceContainer
# Global Exception Handler - Captures uncaught exceptions before crash
def global_exception_handler(exc_type, exc_value, exc_traceback):
"""
Captures and logs uncaught exceptions before program termination.
This helps diagnose crashes that would otherwise leave no trace.
"""
try:
from src.logger import logger
logger.critical(
"UNCAUGHT EXCEPTION - Program will terminate",
exc_info=(exc_type, exc_value, exc_traceback)
)
# Force flush to ensure the exception is written to disk
for handler in logger.handlers:
handler.flush()
except Exception as e:
# Fallback if logging fails
print(f"CRITICAL: Failed to log exception: {e}")
print(f"Original exception: {exc_type.__name__}: {exc_value}")
# Call the default exception handler
sys.__excepthook__(exc_type, exc_value, exc_traceback)
# Install the global exception handler
sys.excepthook = global_exception_handler
from src.services.base_service import IConfigService, IVisionService, IOverlayService, IStateService, IDatabaseService, ITrayService, IAudioService
from src.services.config_service import ConfigService
from src.services.vision_service import VisionService
from src.services.overlay_service import OverlayService
from src.services.state_service import StateService
from src.services.database_service import DatabaseService
from src.services.tray_service import TrayService
from src.services.audio_service import AudioService
from src.ui.region_selector import RegionSelector
from src.ui.state_inspector import StateInspectorWindow
from src.ui.debug_overlay import DebugOverlayManager
from src.ui.ocr_tuner import OCRTunerWindow
# Set DPI Awareness
try:
# Try Per-Monitor V2 (Context -4)
ctypes.windll.user32.SetProcessDpiAwarenessContext(-4)
except Exception:
try:
# Fallback to Per-Monitor (2)
ctypes.windll.shcore.SetProcessDpiAwareness(2)
except Exception:
try:
ctypes.windll.user32.SetProcessDPIAware()
except Exception:
print("Could not set DPI awareness")
import psutil
import os
class Launcher:
def __init__(self):
# Singleton Check
self.mutex_name = "Global\\EldenRingNightreignTimerMutex_V2"
self.kernel32 = ctypes.windll.kernel32
self.mutex = self.kernel32.CreateMutexW(None, False, self.mutex_name)
last_error = self.kernel32.GetLastError()
if last_error == 183: # ERROR_ALREADY_EXISTS
# Another instance is running
if not self.ask_kill_existing():
sys.exit(0)
# If we are here, user said YES and we tried to kill logic.
# Convert to "Restart" flow: Release old mutex handle and try again
try: self.kernel32.CloseHandle(self.mutex)
except: pass
import time
time.sleep(1.0) # Give more time for OS cleanup
self.mutex = self.kernel32.CreateMutexW(None, False, self.mutex_name)
if self.kernel32.GetLastError() == 183:
ctypes.windll.user32.MessageBoxW(0, "Unable to start: Previous instance is still holding the lock.", "Error", 0)
sys.exit(0)
# Qt Application
self.app = QApplication(sys.argv)
self.app.setQuitOnLastWindowClosed(False)
self.container = ServiceContainer()
self.setup_services()
def ask_kill_existing(self):
MB_YESNO = 0x04
MB_ICONQUESTION = 0x20
IDYES = 6
result = ctypes.windll.user32.MessageBoxW(0,
"An instance of Elden Ring Timer is already running.\nDo you want to terminate it and restart?",
"Instance Running",
MB_YESNO | MB_ICONQUESTION)
if result == IDYES:
self.kill_existing_instances()
return True
return False
def kill_existing_instances(self):
current_pid = os.getpid()
killed = False
for proc in psutil.process_iter(['pid', 'name', 'cmdline']):
try:
# Check for python processes running main.py
if proc.info['pid'] == current_pid:
continue
cmdline = proc.info.get('cmdline', [])
if not cmdline: continue
# Heuristic: matches 'main.py' or our exe name
is_target = False
if 'python' in proc.info['name'].lower():
if any('main.py' in arg for arg in cmdline):
is_target = True
elif 'eldenringtimer' in proc.info['name'].lower():
is_target = True
if is_target:
print(f"Killing process {proc.info['pid']}")
proc.kill()
killed = True
except (psutil.NoSuchProcess, psutil.AccessDenied):
pass
if not killed:
print("Could not find process to kill, but mutex was held.")
def setup_services(self):
# 1. Config
self.config_service = ConfigService()
self.container.register(IConfigService, self.config_service)
# 2. Vision
self.vision_service = VisionService(self.config_service)
self.container.register(IVisionService, self.vision_service)
# 3. Overlay
# Note: OverlayService now doesn't need root window like Tkinter did,
# but it owns the ModernOverlay.
self.overlay_service = OverlayService()
self.container.register(IOverlayService, self.overlay_service)
# 4. Database
self.db_service = DatabaseService()
self.container.register(IDatabaseService, self.db_service)
# 5. Audio
self.audio_service = AudioService(self.config_service)
self.container.register(IAudioService, self.audio_service)
# 6. Tray (System Icon)
self.tray_service = TrayService(self)
self.container.register(ITrayService, self.tray_service)
# 7. State
self.state_service = StateService(self.config_service, self.vision_service, self.overlay_service, self.db_service, self.audio_service, self.tray_service)
self.container.register(IStateService, self.state_service)
def run(self):
parser = argparse.ArgumentParser()
parser.add_argument("--config", action="store_true", help="Open configuration window")
args = parser.parse_args()
if args.config:
self.show_config_ui()
else:
self.start_application()
# Run Qt Event Loop
sys.exit(self.app.exec())
def show_config_ui(self):
# Initialize services if purely config mode (CLI)
if not self.container.services_initialized:
self.container.initialize_all()
# Instantiate if missing (CLI mode)
if not hasattr(self, 'settings'):
self.settings = OCRTunerWindow(self.vision_service, self.config_service, self.db_service, self.audio_service, state_service=self.state_service)
self._connect_tuner_signals()
self.settings.show()
self.settings.setWindowState(self.settings.windowState() & ~Qt.WindowState.WindowMinimized | Qt.WindowState.WindowActive)
self.settings.raise_()
self.settings.activateWindow()
def _connect_tuner_signals(self):
# Connect Capture Buttons from Tuner to Main Logic
self.settings.btn_select_region.clicked.connect(self.select_region)
self.settings.btn_select_level_region.clicked.connect(self.select_level_region)
self.settings.btn_select_runes_region.clicked.connect(self.select_runes_region)
self.settings.btn_select_runes_icon_region.clicked.connect(self.select_runes_icon_region)
self.settings.btn_select_char_region.clicked.connect(self.select_char_region)
def show_inspector_ui(self):
if not hasattr(self, 'inspector_window') or self.inspector_window is None:
self.inspector_window = StateInspectorWindow(self.state_service)
self.inspector_window.show()
self.inspector_window.raise_()
self.inspector_window.activateWindow()
def select_region(self):
# Modern properties of PyQt Region Selector
self.selector = RegionSelector()
self.selector.region_selected.connect(self.on_region_selected)
self.selector.show()
def select_level_region(self):
self.selector = RegionSelector()
self.selector.region_selected.connect(self.on_level_region_selected)
self.selector.show()
def select_runes_region(self):
self.selector = RegionSelector()
self.selector.region_selected.connect(self.on_runes_region_selected)
self.selector.show()
def select_runes_icon_region(self):
self.selector = RegionSelector()
self.selector.region_selected.connect(self.on_runes_icon_region_selected)
self.selector.show()
def select_char_region(self):
self.selector = RegionSelector()
self.selector.region_selected.connect(self.on_char_region_selected)
self.selector.show()
def on_region_selected(self, region):
self.config_service.set("monitor_region", region)
self.vision_service.set_region(region)
print(f"Region saved: {region}")
# Bring settings back to front if needed
if hasattr(self, 'settings'): self.settings.raise_()
def on_level_region_selected(self, region):
self.config_service.set("level_region", region)
self.vision_service.set_level_region(region)
print(f"Level Region saved: {region}")
# Bring settings back to front if needed
if hasattr(self, 'settings'): self.settings.raise_()
def on_runes_region_selected(self, region):
self.config_service.set("runes_region", region)
self.vision_service.set_runes_region(region)
print(f"Runes Region saved: {region}")
if hasattr(self, 'settings'): self.settings.raise_()
def on_runes_icon_region_selected(self, region):
self.config_service.set("runes_icon_region", region)
self.vision_service.set_runes_icon_region(region)
print(f"Runes Icon Region saved: {region}")
if hasattr(self, 'settings'): self.settings.raise_()
def on_char_region_selected(self, region):
self.config_service.set("char_region", region)
# self.vision_service.set_char_region(region) # Future implementation
print(f"Character Region saved: {region}")
if hasattr(self, 'settings'): self.settings.raise_()
def start_application(self):
print("Starting Services...")
self.container.initialize_all()
# Initialize Debug Overlay (After config is loaded)
self.debug_manager = DebugOverlayManager(self.config_service)
self.vision_service.set_debug_callback(self.debug_manager.update)
# Show Tuner (Unified Settings)
self.settings = OCRTunerWindow(self.vision_service, self.config_service, self.db_service, self.audio_service, state_service=self.state_service)
self._connect_tuner_signals()
# Connect Hotkey to Settings (F9)
def _f9_cb():
print("[DEBUG] F9 Callback Triggered in Main")
self.overlay_service.schedule(0, self.show_config_ui)
self.state_service.tuner_callback = _f9_cb
if __name__ == "__main__":
launcher = Launcher()
launcher.run()