""" Async loading mixin for non-blocking image loading and PDF generation. """ import os from pathlib import Path from typing import TYPE_CHECKING, Optional, cast import logging from PyQt6.QtCore import QObject from PyQt6.QtWidgets import QProgressDialog from pyPhotoAlbum.async_backend import AsyncImageLoader, AsyncPDFGenerator, ImageCache, LoadPriority if TYPE_CHECKING: from PyQt6.QtWidgets import QMainWindow logger = logging.getLogger(__name__) class AsyncLoadingMixin: # Type hints for expected attributes from mixing class _pdf_progress_dialog: Optional[QProgressDialog] def update(self) -> None: # type: ignore[empty-body] """Expected from QWidget""" ... def window(self) -> "QMainWindow": # type: ignore[empty-body] """Expected from QWidget""" ... """ Mixin to add async loading capabilities to GLWidget. Provides non-blocking image loading and PDF generation with progressive updates and shared caching. """ def _init_async_loading(self): """Initialize async loading components.""" logger.info("Initializing async loading system...") # Create shared image cache (512MB) self.image_cache = ImageCache(max_memory_mb=512) # Create async image loader self.async_image_loader = AsyncImageLoader(cache=self.image_cache, max_workers=4) self.async_image_loader.image_loaded.connect(self._on_image_loaded) self.async_image_loader.load_failed.connect(self._on_image_load_failed) self.async_image_loader.start() # Create async PDF generator self.async_pdf_generator = AsyncPDFGenerator(image_cache=self.image_cache, max_workers=2) self.async_pdf_generator.progress_updated.connect(self._on_pdf_progress) self.async_pdf_generator.export_complete.connect(self._on_pdf_complete) self.async_pdf_generator.export_failed.connect(self._on_pdf_failed) self.async_pdf_generator.start() logger.info("Async loading system initialized") def _cleanup_async_loading(self): """Cleanup async loading components.""" logger.info("Cleaning up async loading system...") if hasattr(self, "async_image_loader"): self.async_image_loader.stop() if hasattr(self, "async_pdf_generator"): self.async_pdf_generator.stop() if hasattr(self, "image_cache"): self.image_cache.clear() logger.info("Async loading system cleaned up") def _on_image_loaded(self, path: Path, image, user_data): """ Handle image loaded callback. Args: path: Path to loaded image image: Loaded PIL Image user_data: User data (ImageData element) """ logger.debug(f"Image loaded callback: {path}") if user_data and hasattr(user_data, "_on_async_image_loaded"): user_data._on_async_image_loaded(image) # Trigger re-render to show newly loaded image self.update() def _on_image_load_failed(self, path: Path, error_msg: str, user_data): """ Handle image load failure. Args: path: Path that failed to load error_msg: Error message user_data: User data (ImageData element) """ logger.warning(f"Image load failed: {path} - {error_msg}") if user_data and hasattr(user_data, "_on_async_image_load_failed"): user_data._on_async_image_load_failed(error_msg) def _on_pdf_progress(self, current: int, total: int, message: str): """ Handle PDF export progress updates. Args: current: Current progress (pages completed) total: Total pages message: Progress message """ logger.debug(f"PDF progress: {current}/{total} - {message}") # Update progress dialog if it exists # Use local reference to avoid race condition dialog = getattr(self, "_pdf_progress_dialog", None) if dialog is not None: dialog.setValue(current) dialog.setLabelText(message) def _on_pdf_complete(self, success: bool, warnings: list): """ Handle PDF export completion. Args: success: Whether export succeeded warnings: List of warning messages """ logger.info(f"PDF export complete: success={success}, warnings={len(warnings)}") # Close progress dialog if hasattr(self, "_pdf_progress_dialog") and self._pdf_progress_dialog: self._pdf_progress_dialog.close() self._pdf_progress_dialog = None # Show completion message main_window = self.window() if hasattr(main_window, "show_status"): if success: if warnings: main_window.show_status(f"PDF exported successfully with {len(warnings)} warnings", 5000) else: main_window.show_status("PDF exported successfully", 3000) else: main_window.show_status("PDF export failed", 5000) def _on_pdf_failed(self, error_msg: str): """ Handle PDF export failure. Args: error_msg: Error message """ logger.error(f"PDF export failed: {error_msg}") # Close progress dialog if hasattr(self, "_pdf_progress_dialog") and self._pdf_progress_dialog: self._pdf_progress_dialog.close() self._pdf_progress_dialog = None # Show error message main_window = self.window() if hasattr(main_window, "show_status"): main_window.show_status(f"PDF export failed: {error_msg}", 5000) def request_image_load(self, image_data, priority: LoadPriority = LoadPriority.NORMAL): """ Request async load for an ImageData element. Args: image_data: ImageData element to load priority: Load priority level """ if not hasattr(self, "async_image_loader"): logger.warning("Async image loader not initialized") return if not image_data.image_path: return # Security: only load images from the assets folder if not image_data.image_path.startswith("assets/"): logger.warning(f"Skipping path not in assets folder (needs healing): {image_data.image_path}") return # Use ImageData's path resolution (delegates to project layer) image_full_path = image_data.resolve_image_path() if not image_full_path: logger.warning(f"Image not found (needs healing): {image_data.image_path}") return # Calculate target size (max 2048px like original) target_size = (2048, 2048) # Will be downsampled if larger # Request load self.async_image_loader.request_load( Path(image_full_path), priority=priority, target_size=target_size, user_data=image_data, # Pass element for callback ) def export_pdf_async(self, project, output_path: str, export_dpi: int = 300): """ Export PDF asynchronously without blocking UI. Args: project: Project to export output_path: Output PDF file path export_dpi: Export DPI (default 300) """ if not hasattr(self, "async_pdf_generator"): logger.warning("Async PDF generator not initialized") return False # Create progress dialog from PyQt6.QtWidgets import QProgressDialog from PyQt6.QtCore import Qt total_pages = sum(1 if page.is_cover else (2 if page.is_double_spread else 1) for page in project.pages) from PyQt6.QtWidgets import QWidget self._pdf_progress_dialog = QProgressDialog("Exporting to PDF...", "Cancel", 0, total_pages, cast(QWidget, self)) self._pdf_progress_dialog.setWindowModality(Qt.WindowModality.WindowModal) self._pdf_progress_dialog.setWindowTitle("PDF Export") self._pdf_progress_dialog.canceled.connect(self._on_pdf_cancel) self._pdf_progress_dialog.show() # Start async export return self.async_pdf_generator.export_pdf(project, output_path, export_dpi) def _on_pdf_cancel(self): """Handle PDF export cancellation.""" logger.info("User requested PDF export cancellation") if hasattr(self, "async_pdf_generator"): self.async_pdf_generator.cancel_export() def get_async_stats(self) -> dict: """Get async loading system statistics.""" stats = {} if hasattr(self, "async_image_loader"): stats["image_loader"] = self.async_image_loader.get_stats() if hasattr(self, "async_pdf_generator"): stats["pdf_generator"] = self.async_pdf_generator.get_stats() return stats