import importlib import time import traceback from PyQt6.QtCore import QObject, QTimer from gui.v2.infrastructure.page_registry import ( PAGE_REGISTRY, ASYNC_PREPARE, REGULAR_FLOW_ONLY, PRELOAD_ORDER, PRELOAD_SKIP, ) from gui.v2.workers.page_data_worker import PageDataWorker from gui.v2.actions.flags import is_fast_registration_enabled from core.errors.logger import logger class Navigator(QObject): def __init__(self, custom_window): super().__init__() self.custom_window = custom_window self.page_stack = custom_window.page_stack self._instances = {} self._preload_queue = [] self._preload_total = 0 self._preload_done = 0 self._preload_failed = 0 self._preload_skipped_runtime = 0 self._preload_dispatched = 0 self._preload_inflight = 0 self._preload_workers = {} self._preload_summary_emitted = False self._preload_t_start = 0.0 self._preload_timer = QTimer(self) self._preload_timer.setSingleShot(True) self._preload_timer.timeout.connect(self._preload_next) def get_class(self, name): module_path, class_name = PAGE_REGISTRY[name] module = importlib.import_module(module_path) return getattr(module, class_name) def _instantiate(self, name, prepared=None): cls = self.get_class(name) if name in ASYNC_PREPARE: return cls(self.page_stack, self.custom_window, prepared=prepared) return cls(self.page_stack, self.custom_window) def register_instance(self, name, page): self.page_stack.addWidget(page) self._instances[name] = page def navigate(self, name): page = self._instances.get(name) if page is None: if name not in PAGE_REGISTRY: self._warn_not_migrated(name) return None page = self._instantiate(name) self.register_instance(name, page) self.page_stack.setCurrentIndex(self.page_stack.indexOf(page)) return page def get_cached(self, name): return self._instances.get(name) def update_if_cached(self, name, fn): page = self._instances.get(name) if page is not None: fn(page) return page def _warn_not_migrated(self, name): try: self.custom_window.update_status(f"Page '{name}' not migrated yet.") except Exception: pass def _preload_one(self, name): if name in self._instances: logger.debug(f"[preload] {name:30s} SKIP (already cached)") return False if name not in PAGE_REGISTRY: logger.debug(f"[preload] {name:30s} SKIP (not in registry)") return False module_path, class_name = PAGE_REGISTRY[name] logger.debug(f"[preload] {name:30s} LOAD {module_path}.{class_name}") t0 = time.perf_counter() try: page = self._instantiate(name) self.register_instance(name, page) dt_ms = (time.perf_counter() - t0) * 1000.0 logger.debug(f"[preload] {name:30s} OK ({dt_ms:6.1f} ms, " f"cached={len(self._instances)}, stack_size={self.page_stack.count()})") return True except Exception as e: dt_ms = (time.perf_counter() - t0) * 1000.0 logger.debug(f"[preload] {name:30s} FAIL ({dt_ms:6.1f} ms) " f"{type(e).__name__}: {e}") traceback.print_exc() from core.errors.logger import logger as core_logger core_logger.warning(f"Background preload failed for '{name}': " f"{type(e).__name__}: {e}") return False def start_preload(self, fast_mode=None): if fast_mode is None: fast_mode = is_fast_registration_enabled(self.custom_window.gui_config_file) self._preload_queue = [n for n in PRELOAD_ORDER if n not in PRELOAD_SKIP] skipped_regular = [] if fast_mode: skipped_regular = [n for n in self._preload_queue if n in REGULAR_FLOW_ONLY] self._preload_queue = [n for n in self._preload_queue if n not in REGULAR_FLOW_ONLY] self._preload_total = len(self._preload_queue) self._preload_done = 0 self._preload_failed = 0 self._preload_skipped_runtime = 0 self._preload_dispatched = 0 self._preload_inflight = 0 self._preload_summary_emitted = False self._preload_t_start = time.perf_counter() logger.debug(f"[preload] ============================================================") logger.debug(f"[preload] starting background preload: {self._preload_total} pages queued") logger.debug(f"[preload] order : {', '.join(self._preload_queue)}") logger.debug(f"[preload] skipped : {', '.join(sorted(PRELOAD_SKIP))} (unsafe init)") if fast_mode: logger.debug(f"[preload] fastmode: ON -> skipping regular-flow pages: " f"{', '.join(skipped_regular) if skipped_regular else '(none)'}") logger.debug(f"[preload] gap : 250 ms between loads, 700 ms initial delay") logger.debug(f"[preload] ============================================================") self._preload_timer.start(700) def _preload_next(self): while self._preload_queue and self._preload_queue[0] in self._instances: skipped = self._preload_queue.pop(0) self._preload_skipped_runtime += 1 logger.debug(f"[preload] {skipped:30s} SKIP (user navigated here first)") if not self._preload_queue: self._maybe_emit_preload_summary() return self._preload_dispatched += 1 idx = self._preload_dispatched name = self._preload_queue.pop(0) logger.debug(f"[preload] ---- {idx}/{self._preload_total} ----") if name in ASYNC_PREPARE: self._dispatch_async_preload(name) else: ok = self._preload_one(name) if ok: self._preload_done += 1 else: self._preload_failed += 1 if self._preload_queue: self._preload_timer.start(250) else: self._maybe_emit_preload_summary() def _prepare_page_data(self, name): module_path, class_name = PAGE_REGISTRY[name] module = importlib.import_module(module_path) cls = getattr(module, class_name) return cls.prepare_data(self.custom_window) def _dispatch_async_preload(self, name): module_path, class_name = PAGE_REGISTRY[name] logger.debug(f"[preload] {name:30s} ASYNC {module_path}.{class_name}") worker = PageDataWorker(name, self._prepare_page_data, name) worker.data_ready.connect(self._on_preload_data_ready) worker.failed.connect(self._on_preload_failed) worker.finished.connect(lambda n=name: self._cleanup_preload_worker(n)) self._preload_workers[name] = worker self._preload_inflight += 1 worker.start() def _on_preload_data_ready(self, name, payload): self._preload_inflight -= 1 if name in self._instances: logger.debug(f"[preload] {name:30s} SKIP (already cached, async result dropped)") else: try: page = self._instantiate(name, prepared=payload) self.register_instance(name, page) self._preload_done += 1 logger.debug(f"[preload] {name:30s} OK (async, cached={len(self._instances)}, " f"stack_size={self.page_stack.count()})") except Exception as e: self._preload_failed += 1 logger.debug(f"[preload] {name:30s} FAIL (async build) {type(e).__name__}: {e}") self._maybe_emit_preload_summary() def _on_preload_failed(self, name, error): self._preload_inflight -= 1 logger.debug(f"[preload] {name:30s} FAIL (async prepare) {error}") if name not in self._instances: ok = self._preload_one(name) if ok: self._preload_done += 1 else: self._preload_failed += 1 self._maybe_emit_preload_summary() def _cleanup_preload_worker(self, name): worker = self._preload_workers.pop(name, None) if worker is not None: worker.deleteLater() def _maybe_emit_preload_summary(self): if self._preload_summary_emitted: return if not self._preload_queue and self._preload_inflight == 0: self._preload_summary_emitted = True self._emit_preload_summary() def _emit_preload_summary(self): total_s = time.perf_counter() - self._preload_t_start logger.debug(f"[preload] ============================================================") logger.debug(f"[preload] complete: {self._preload_done} loaded, " f"{self._preload_failed} failed, " f"{self._preload_skipped_runtime} skipped (user beat us to it) " f"in {total_s:.2f}s") logger.debug(f"[preload] cached pages now: {sorted(self._instances.keys())}") logger.debug(f"[preload] ============================================================")