Coverage for pyPhotoAlbum/async_backend.py: 73%
363 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-20 12:55 +0000
1"""
2Async backend for non-blocking image loading and PDF generation.
4This module provides:
5- AsyncImageLoader: Load and process images in background
6- AsyncPDFGenerator: Generate PDFs without blocking UI
7- ImageCache: Intelligent caching with LRU eviction
8- WorkerPool: Thread pool for CPU-bound operations
9"""
11import asyncio
12import logging
13from concurrent.futures import ThreadPoolExecutor
14from dataclasses import dataclass, field
15from enum import IntEnum
16from pathlib import Path
17from typing import Optional, Callable, Dict, Any, Tuple, Union
18from concurrent.futures import Future
19from collections import OrderedDict
20import threading
22from PIL import Image
23from PyQt6.QtCore import QObject, pyqtSignal
25from pyPhotoAlbum.image_utils import convert_to_rgba, resize_to_fit
27logger = logging.getLogger(__name__)
30class LoadPriority(IntEnum):
31 """Priority levels for load requests."""
33 LOW = 0 # Offscreen, not visible
34 NORMAL = 1 # Potentially visible soon
35 HIGH = 2 # Visible on screen
36 URGENT = 3 # User is actively interacting with
39def get_image_dimensions(image_path: str, max_size: Optional[int] = None) -> Optional[Tuple[int, int]]:
40 """
41 Extract image dimensions without loading the full image.
43 Uses PIL's lazy loading to read only the header, making this a fast
44 operation suitable for UI code that needs dimensions before async loading.
46 Args:
47 image_path: Path to the image file (absolute or relative)
48 max_size: Optional maximum dimension - if provided, dimensions are
49 scaled down proportionally to fit within this limit
51 Returns:
52 Tuple of (width, height) or None if the image cannot be read
54 Example:
55 # Get raw dimensions
56 dims = get_image_dimensions("/path/to/image.jpg")
58 # Get dimensions scaled to fit within 300px
59 dims = get_image_dimensions("/path/to/image.jpg", max_size=300)
60 """
61 try:
62 with Image.open(image_path) as img:
63 width, height = img.size
65 if max_size and (width > max_size or height > max_size):
66 scale = min(max_size / width, max_size / height)
67 width = int(width * scale)
68 height = int(height * scale)
70 return (width, height)
72 except Exception as e:
73 logger.warning(f"Could not extract dimensions for {image_path}: {e}")
74 return None
77@dataclass(order=True)
78class LoadRequest:
79 """Request to load and process an image."""
81 priority: LoadPriority = field(compare=True)
82 request_id: int = field(compare=True) # Tie-breaker for same priority
83 path: Path = field(compare=False)
84 target_size: Optional[Tuple[int, int]] = field(default=None, compare=False)
85 callback: Optional[Callable] = field(default=None, compare=False)
86 user_data: Any = field(default=None, compare=False)
89class ImageCache:
90 """
91 Thread-safe LRU cache for PIL images with memory management.
93 Caches both original images and scaled variants to avoid redundant processing.
94 """
96 def __init__(self, max_memory_mb: int = 512):
97 """
98 Initialize cache.
100 Args:
101 max_memory_mb: Maximum memory to use for cached images (default 512MB)
102 """
103 self.max_memory_bytes = max_memory_mb * 1024 * 1024
104 self.current_memory_bytes = 0
105 self._cache: OrderedDict[str, Tuple[Image.Image, int]] = OrderedDict()
106 self._lock = threading.Lock()
108 logger.info(f"ImageCache initialized with {max_memory_mb}MB limit")
110 def _estimate_image_size(self, img: Image.Image) -> int:
111 """Estimate memory size of PIL image in bytes."""
112 # PIL images are typically width * height * bytes_per_pixel
113 # RGBA = 4 bytes, RGB = 3 bytes, L = 1 byte
114 mode_sizes = {"RGBA": 4, "RGB": 3, "L": 1, "LA": 2}
115 bytes_per_pixel = mode_sizes.get(img.mode, 4)
116 return img.width * img.height * bytes_per_pixel
118 def _make_key(self, path: Path, target_size: Optional[Tuple[int, int]] = None) -> str:
119 """Create cache key from path and optional target size."""
120 if target_size:
121 return f"{path}:{target_size[0]}x{target_size[1]}"
122 return str(path)
124 def get(self, path: Path, target_size: Optional[Tuple[int, int]] = None) -> Optional[Image.Image]:
125 """
126 Get image from cache.
128 Args:
129 path: Path to image file
130 target_size: Optional target size (width, height)
132 Returns:
133 Cached PIL Image or None if not found
134 """
135 key = self._make_key(path, target_size)
137 with self._lock:
138 if key in self._cache:
139 # Move to end (most recently used)
140 img, size = self._cache.pop(key)
141 self._cache[key] = (img, size)
142 logger.debug(f"Cache HIT: {key}")
143 return img.copy() # Return copy to avoid external modifications
145 logger.debug(f"Cache MISS: {key}")
146 return None
148 def put(self, path: Path, img: Image.Image, target_size: Optional[Tuple[int, int]] = None):
149 """
150 Add image to cache with LRU eviction.
152 Args:
153 path: Path to image file
154 img: PIL Image to cache
155 target_size: Optional target size used for this variant
156 """
157 key = self._make_key(path, target_size)
158 img_size = self._estimate_image_size(img)
160 with self._lock:
161 # Remove if already exists (update size)
162 if key in self._cache:
163 _, old_size = self._cache.pop(key)
164 self.current_memory_bytes -= old_size
166 # Evict LRU items if needed
167 while self.current_memory_bytes + img_size > self.max_memory_bytes and len(self._cache) > 0:
168 evicted_key, (evicted_img, evicted_size) = self._cache.popitem(last=False)
169 self.current_memory_bytes -= evicted_size
170 logger.debug(f"Cache EVICT: {evicted_key} ({evicted_size / 1024 / 1024:.1f}MB)")
172 # Add new image
173 self._cache[key] = (img.copy(), img_size)
174 self.current_memory_bytes += img_size
176 logger.debug(
177 f"Cache PUT: {key} ({img_size / 1024 / 1024:.1f}MB) "
178 f"[Total: {self.current_memory_bytes / 1024 / 1024:.1f}MB / "
179 f"{self.max_memory_bytes / 1024 / 1024:.1f}MB, "
180 f"Items: {len(self._cache)}]"
181 )
183 def clear(self):
184 """Clear entire cache."""
185 with self._lock:
186 self._cache.clear()
187 self.current_memory_bytes = 0
188 logger.info("Cache cleared")
190 def get_stats(self) -> Dict[str, Any]:
191 """Get cache statistics."""
192 with self._lock:
193 return {
194 "items": len(self._cache),
195 "memory_mb": self.current_memory_bytes / 1024 / 1024,
196 "max_memory_mb": self.max_memory_bytes / 1024 / 1024,
197 "utilization": (self.current_memory_bytes / self.max_memory_bytes) * 100,
198 }
201class AsyncImageLoader(QObject):
202 """
203 Asynchronous image loader with priority queue and caching.
205 Loads images in background threads and emits signals when complete.
206 Supports concurrent loading, priority-based scheduling, and cancellation.
208 Example:
209 loader = AsyncImageLoader()
210 loader.image_loaded.connect(on_image_ready)
211 loader.start()
212 loader.request_load(Path("photo.jpg"), priority=LoadPriority.HIGH)
213 """
215 # Signals for Qt integration
216 image_loaded = pyqtSignal(object, object, object) # (path, image, user_data)
217 load_failed = pyqtSignal(object, str, object) # (path, error_msg, user_data)
219 def __init__(self, cache: Optional[ImageCache] = None, max_workers: int = 4):
220 """
221 Initialize async image loader.
223 Args:
224 cache: ImageCache instance (creates new if None)
225 max_workers: Maximum concurrent worker threads (default 4)
226 """
227 super().__init__()
229 self.cache = cache or ImageCache()
230 self.max_workers = max_workers
231 self.executor = ThreadPoolExecutor(max_workers=max_workers, thread_name_prefix="ImageLoader")
233 # Priority queue and tracking
234 self._queue: Optional[asyncio.PriorityQueue[Any]] = None # Created when event loop starts
235 self._pending_requests: Dict[Path, LoadRequest] = {}
236 self._active_tasks: Dict[Path, asyncio.Task] = {}
237 self._next_request_id = 0
238 self._lock = threading.Lock()
239 self._shutdown = False
241 # Event loop for async operations
242 self._loop: Optional[asyncio.AbstractEventLoop] = None
243 self._loop_thread: Optional[threading.Thread] = None
245 logger.info(f"AsyncImageLoader initialized with {max_workers} workers")
247 def start(self):
248 """Start the async backend event loop."""
249 if self._loop_thread is not None:
250 logger.warning("AsyncImageLoader already started")
251 return
253 self._shutdown = False
254 self._loop_thread = threading.Thread(
255 target=self._run_event_loop, daemon=True, name="AsyncImageLoader-EventLoop"
256 )
257 self._loop_thread.start()
258 logger.info("AsyncImageLoader event loop started")
260 def stop(self):
261 """Stop the async backend and cleanup resources."""
262 if self._loop is None:
263 return
265 logger.info("Stopping AsyncImageLoader...")
266 self._shutdown = True
268 # Cancel all active tasks and wait for them to finish
269 if self._loop and not self._loop.is_closed():
270 future = asyncio.run_coroutine_threadsafe(self._cancel_all_tasks(), self._loop)
271 try:
272 # Wait for cancellation to complete with timeout
273 future.result(timeout=2.0)
274 except Exception as e:
275 logger.warning(f"Error during task cancellation: {e}")
277 # Stop the event loop
278 self._loop.call_soon_threadsafe(self._loop.stop)
280 # Wait for thread to finish
281 if self._loop_thread:
282 self._loop_thread.join(timeout=5.0)
284 # Shutdown executor
285 self.executor.shutdown(wait=True)
287 logger.info("AsyncImageLoader stopped")
289 def _run_event_loop(self):
290 """Run asyncio event loop in background thread."""
291 self._loop = asyncio.new_event_loop()
292 asyncio.set_event_loop(self._loop)
294 # Create priority queue
295 self._queue = asyncio.PriorityQueue()
297 # Start queue processor as background task
298 self._loop.create_task(self._process_queue())
300 # Run event loop forever (until stopped)
301 self._loop.run_forever()
303 # Cleanup after loop stops
304 self._loop.close()
306 async def _process_queue(self):
307 """Process load requests from priority queue."""
308 logger.info("Queue processor started")
310 while not self._shutdown:
311 try:
312 # Wait for request with timeout to check shutdown flag
313 request = await asyncio.wait_for(self._queue.get(), timeout=0.5)
315 # Skip if already cancelled
316 if request.path not in self._pending_requests:
317 continue
319 # Process request
320 task = asyncio.create_task(self._load_image(request))
321 self._active_tasks[request.path] = task
323 except asyncio.TimeoutError:
324 continue # Check shutdown flag
325 except Exception as e:
326 logger.error(f"Queue processor error: {e}", exc_info=True)
328 logger.info("Queue processor stopped")
330 async def _cancel_all_tasks(self):
331 """Cancel all active loading tasks."""
332 tasks = list(self._active_tasks.values())
333 for task in tasks:
334 task.cancel()
336 if tasks:
337 await asyncio.gather(*tasks, return_exceptions=True)
339 self._active_tasks.clear()
340 self._pending_requests.clear()
342 async def _load_image(self, request: LoadRequest):
343 """
344 Load and process image asynchronously.
346 Args:
347 request: LoadRequest containing path, size, and callback info
348 """
349 path = request.path
350 target_size = request.target_size
352 try:
353 # Check if shutting down
354 if self._shutdown:
355 return
357 # Check cache first
358 cached_img = self.cache.get(path, target_size)
359 if cached_img is not None:
360 logger.debug(f"Loaded from cache: {path}")
361 self._emit_loaded(path, cached_img, request.user_data)
362 return
364 # Load in thread pool (I/O bound)
365 loop = asyncio.get_event_loop()
366 img = await loop.run_in_executor(self.executor, self._load_and_process_image, path, target_size)
368 # Check again if shutting down before emitting
369 if self._shutdown:
370 return
372 # Cache result
373 self.cache.put(path, img, target_size)
375 # Emit success signal
376 self._emit_loaded(path, img, request.user_data)
378 logger.debug(f"Loaded: {path} (size: {img.size})")
380 except asyncio.CancelledError:
381 # Task was cancelled during shutdown - this is expected
382 logger.debug(f"Load cancelled for {path}")
383 raise # Re-raise to properly cancel the task
385 except Exception as e:
386 # Only emit error if not shutting down
387 if not self._shutdown:
388 logger.error(f"Failed to load {path}: {e}", exc_info=True)
389 self._emit_failed(path, str(e), request.user_data)
391 finally:
392 # Cleanup tracking
393 with self._lock:
394 self._pending_requests.pop(path, None)
395 self._active_tasks.pop(path, None)
397 def _load_and_process_image(self, path: Path, target_size: Optional[Tuple[int, int]]) -> "Image.Image":
398 """
399 Load image from disk and process (runs in thread pool).
401 Args:
402 path: Path to image file
403 target_size: Optional target size for downsampling
405 Returns:
406 Processed PIL Image
407 """
408 img: Image.Image = Image.open(path)
409 img = convert_to_rgba(img)
411 # Downsample if target size specified (preserving aspect ratio)
412 if target_size:
413 max_size = target_size[0] # Assume square target (2048, 2048)
414 original_size = img.size
415 img = resize_to_fit(img, max_size)
416 if img.size != original_size:
417 logger.debug(f"Downsampled {path}: {original_size} -> {img.size}")
419 return img
421 def _emit_loaded(self, path: Path, img: Image.Image, user_data: Any):
422 """Emit image_loaded signal (thread-safe)."""
423 # Check if object is still valid before emitting
424 if self._shutdown:
425 return
426 try:
427 self.image_loaded.emit(path, img, user_data)
428 except RuntimeError as e:
429 # Object was deleted - log but don't crash
430 logger.debug(f"Could not emit image_loaded for {path}: {e}")
432 def _emit_failed(self, path: Path, error_msg: str, user_data: Any):
433 """Emit load_failed signal (thread-safe)."""
434 # Check if object is still valid before emitting
435 if self._shutdown:
436 return
437 try:
438 self.load_failed.emit(path, error_msg, user_data)
439 except RuntimeError as e:
440 # Object was deleted - log but don't crash
441 logger.debug(f"Could not emit load_failed for {path}: {e}")
443 def request_load(
444 self,
445 path: Path,
446 priority: LoadPriority = LoadPriority.NORMAL,
447 target_size: Optional[Tuple[int, int]] = None,
448 user_data: Any = None,
449 ) -> bool:
450 """
451 Request image load with specified priority.
453 Args:
454 path: Path to image file
455 priority: Load priority level
456 target_size: Optional target size (width, height) for downsampling
457 user_data: Optional user data passed to callback
459 Returns:
460 True if request was queued, False if already pending/active
461 """
462 if not self._loop or self._shutdown:
463 logger.warning("Cannot request load: backend not started")
464 return False
466 path = Path(path)
468 with self._lock:
469 # Skip if already pending or active
470 if path in self._pending_requests or path in self._active_tasks:
471 logger.debug(f"Load already pending: {path}")
472 return False
474 # Create request
475 request = LoadRequest(
476 priority=priority,
477 request_id=self._next_request_id,
478 path=path,
479 target_size=target_size,
480 user_data=user_data,
481 )
482 self._next_request_id += 1
484 # Track as pending
485 self._pending_requests[path] = request
487 # Submit to queue (thread-safe)
488 assert self._queue is not None
489 asyncio.run_coroutine_threadsafe(self._queue.put(request), self._loop)
491 logger.debug(f"Queued load: {path} (priority: {priority.name})")
492 return True
494 def cancel_load(self, path: Path) -> bool:
495 """
496 Cancel pending image load.
498 Args:
499 path: Path to image file
501 Returns:
502 True if load was cancelled, False if not found
503 """
504 path = Path(path)
506 with self._lock:
507 # Remove from pending
508 if path in self._pending_requests:
509 del self._pending_requests[path]
510 logger.debug(f"Cancelled pending load: {path}")
511 return True
513 # Cancel active task
514 if path in self._active_tasks:
515 task = self._active_tasks[path]
516 task.cancel()
517 logger.debug(f"Cancelled active load: {path}")
518 return True
520 return False
522 def get_stats(self) -> Dict[str, Any]:
523 """Get loader statistics."""
524 with self._lock:
525 return {
526 "pending": len(self._pending_requests),
527 "active": len(self._active_tasks),
528 "cache": self.cache.get_stats(),
529 }
532class AsyncPDFGenerator(QObject):
533 """
534 Asynchronous PDF generator that doesn't block the UI.
536 Generates PDFs in background thread with progress updates.
537 Uses shared ImageCache to avoid redundant image loading.
539 Example:
540 generator = AsyncPDFGenerator(image_cache)
541 generator.progress_updated.connect(on_progress)
542 generator.export_complete.connect(on_complete)
543 generator.start()
544 generator.export_pdf(project, "output.pdf")
545 """
547 # Signals for Qt integration
548 progress_updated = pyqtSignal(int, int, str) # (current, total, message)
549 export_complete = pyqtSignal(bool, list) # (success, warnings)
550 export_failed = pyqtSignal(str) # (error_message)
552 def __init__(self, image_cache: Optional[ImageCache] = None, max_workers: int = 2):
553 """
554 Initialize async PDF generator.
556 Args:
557 image_cache: Shared ImageCache instance (creates new if None)
558 max_workers: Maximum concurrent workers for PDF generation (default 2)
559 """
560 super().__init__()
562 self.image_cache = image_cache or ImageCache()
563 self.max_workers = max_workers
564 self.executor = ThreadPoolExecutor(max_workers=max_workers, thread_name_prefix="PDFGenerator")
566 # Export state
567 self._current_export: Optional[Future[Any]] = None
568 self._cancel_requested = False
569 self._lock = threading.RLock() # Use RLock to allow re-entrant locking
570 self._shutdown = False
572 # Event loop for async operations
573 self._loop: Optional[asyncio.AbstractEventLoop] = None
574 self._loop_thread: Optional[threading.Thread] = None
576 logger.info(f"AsyncPDFGenerator initialized with {max_workers} workers")
578 def start(self):
579 """Start the async PDF generator event loop."""
580 if self._loop_thread is not None:
581 logger.warning("AsyncPDFGenerator already started")
582 return
584 self._shutdown = False
585 self._loop_thread = threading.Thread(
586 target=self._run_event_loop, daemon=True, name="AsyncPDFGenerator-EventLoop"
587 )
588 self._loop_thread.start()
589 logger.info("AsyncPDFGenerator event loop started")
591 def stop(self):
592 """Stop the async PDF generator and cleanup resources."""
593 if self._loop is None:
594 return
596 logger.info("Stopping AsyncPDFGenerator...")
597 self._shutdown = True
599 # Cancel active export
600 if self._current_export and not self._current_export.done():
601 self._current_export.cancel()
603 # Stop the event loop
604 if self._loop and not self._loop.is_closed():
605 self._loop.call_soon_threadsafe(self._loop.stop)
607 # Wait for thread to finish
608 if self._loop_thread:
609 self._loop_thread.join(timeout=5.0)
611 # Shutdown executor
612 self.executor.shutdown(wait=True)
614 logger.info("AsyncPDFGenerator stopped")
616 def _run_event_loop(self):
617 """Run asyncio event loop in background thread."""
618 self._loop = asyncio.new_event_loop()
619 asyncio.set_event_loop(self._loop)
621 # Run event loop forever (until stopped)
622 self._loop.run_forever()
624 # Cleanup after loop stops
625 self._loop.close()
627 def export_pdf(self, project, output_path: str, export_dpi: int = 300) -> bool:
628 """
629 Request PDF export (non-blocking).
631 Args:
632 project: Project instance to export
633 output_path: Path where PDF should be saved
634 export_dpi: Target DPI for images (default 300)
636 Returns:
637 True if export started, False if already exporting or backend not started
638 """
639 if not self._loop or self._shutdown:
640 logger.warning("Cannot export: backend not started")
641 return False
643 with self._lock:
644 if self._current_export and not self._current_export.done():
645 logger.warning("Export already in progress")
646 return False
648 self._cancel_requested = False
650 # Submit export task
651 self._current_export = asyncio.run_coroutine_threadsafe(
652 self._export_pdf_async(project, output_path, export_dpi), self._loop
653 )
655 logger.info(f"PDF export started: {output_path}")
656 return True
658 def cancel_export(self):
659 """Request cancellation of current export."""
660 with self._lock:
661 self._cancel_requested = True
662 if self._current_export and not self._current_export.done():
663 self._current_export.cancel()
664 logger.info("PDF export cancellation requested")
666 async def _export_pdf_async(self, project, output_path: str, export_dpi: int):
667 """
668 Perform PDF export asynchronously.
670 Args:
671 project: Project to export
672 output_path: Output PDF file path
673 export_dpi: Export DPI setting
674 """
675 try:
676 # Import PDF exporter (lazy import to avoid circular dependencies)
677 from pyPhotoAlbum.pdf_exporter import PDFExporter
679 # Create exporter
680 exporter = PDFExporter(project, export_dpi=export_dpi)
682 # Progress callback wrapper
683 def progress_callback(current, total, message):
684 if self._cancel_requested or self._shutdown:
685 return False # Signal cancellation
686 try:
687 self.progress_updated.emit(current, total, message)
688 except RuntimeError as e:
689 # Object was deleted - log but don't crash
690 logger.debug(f"Could not emit progress_updated: {e}")
691 return False
692 return True
694 # Run export in thread pool
695 loop = asyncio.get_event_loop()
696 success, warnings = await loop.run_in_executor(
697 self.executor, self._export_with_cache, exporter, output_path, progress_callback
698 )
700 # Emit completion signal
701 if not self._cancel_requested and not self._shutdown:
702 try:
703 self.export_complete.emit(success, warnings)
704 logger.info(f"PDF export completed: {output_path} (warnings: {len(warnings)})")
705 except RuntimeError as e:
706 logger.debug(f"Could not emit export_complete: {e}")
707 else:
708 logger.info("PDF export cancelled")
710 except asyncio.CancelledError:
711 logger.info("PDF export cancelled by user")
712 if not self._shutdown:
713 try:
714 self.export_failed.emit("Export cancelled")
715 except RuntimeError as e:
716 logger.debug(f"Could not emit export_failed: {e}")
718 except Exception as e:
719 logger.error(f"PDF export failed: {e}", exc_info=True)
720 if not self._shutdown:
721 try:
722 self.export_failed.emit(str(e))
723 except RuntimeError as e:
724 logger.debug(f"Could not emit export_failed: {e}")
726 finally:
727 with self._lock:
728 self._current_export = None
730 def _export_with_cache(self, exporter: Any, output_path: str, progress_callback: Any) -> Tuple[bool, list[Any]]:
731 """
732 Run PDF export with image cache integration.
734 This method patches the exporter to use our cached images.
736 Args:
737 exporter: PDFExporter instance
738 output_path: Output file path
739 progress_callback: Progress callback function
741 Returns:
742 Tuple of (success, warnings)
743 """
744 # Store original Image.open
745 original_open = Image.open
747 # Patch Image.open to use cache
748 def cached_open(path, *args, **kwargs):
749 # Only use cache for file paths, not BytesIO or other file-like objects
750 is_file_path = isinstance(path, (str, Path))
752 if is_file_path:
753 # Try cache first
754 # Note: We cache the unrotated image so rotation can be applied per-element
755 path_obj = Path(path) if isinstance(path, str) else path
756 cached_img = self.image_cache.get(path_obj)
757 if cached_img:
758 logger.debug(f"PDF using cached image: {path}")
759 return cached_img
761 # Load and cache (unrotated - rotation is applied per-element)
762 img = original_open(path, *args, **kwargs)
763 img = convert_to_rgba(img)
764 self.image_cache.put(path_obj, img)
765 return img
766 else:
767 # For BytesIO and other file-like objects, just use original open
768 return original_open(path, *args, **kwargs)
770 # Temporarily patch Image.open
771 try:
772 Image.open = cached_open # type: ignore[assignment]
773 return exporter.export(output_path, progress_callback) # type: ignore[no-any-return]
774 finally:
775 # Restore original
776 Image.open = original_open
778 def is_exporting(self) -> bool:
779 """Check if export is currently in progress."""
780 with self._lock:
781 return self._current_export is not None and not self._current_export.done()
783 def get_stats(self) -> Dict[str, Any]:
784 """Get generator statistics."""
785 with self._lock:
786 return {"exporting": self.is_exporting(), "cache": self.image_cache.get_stats()}