Coverage for pyPhotoAlbum/async

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()

107

108 logger.info(f"ImageCache initialized with {max_memory_mb}MB limit")

109

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

117

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)

123

124 def get(self, path: Path, target_size: Optional[Tuple[int, int]] = None) -> Optional[Image.Image]:

125 """

126 Get image from cache.

127

128 Args:

129 path: Path to image file

130 target_size: Optional target size (width, height)

131

132 Returns:

133 Cached PIL Image or None if not found

134 """

135 key = self._make_key(path, target_size)

136

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

144

145 logger.debug(f"Cache MISS: {key}")

146 return None

147

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.

151

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)

159

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

165

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)")

171

172 # Add new image

173 self._cache[key] = (img.copy(), img_size)

174 self.current_memory_bytes += img_size

175

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 )

182

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")

189

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 }

199

200

201class AsyncImageLoader(QObject):

202 """

203 Asynchronous image loader with priority queue and caching.

204

205 Loads images in background threads and emits signals when complete.

206 Supports concurrent loading, priority-based scheduling, and cancellation.

207

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 """

214

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)

218

219 def __init__(self, cache: Optional[ImageCache] = None, max_workers: int = 4):

220 """

221 Initialize async image loader.

222

223 Args:

224 cache: ImageCache instance (creates new if None)

225 max_workers: Maximum concurrent worker threads (default 4)

226 """

227 super().__init__()

228

229 self.cache = cache or ImageCache()

230 self.max_workers = max_workers

231 self.executor = ThreadPoolExecutor(max_workers=max_workers, thread_name_prefix="ImageLoader")

232

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

240

241 # Event loop for async operations

242 self._loop: Optional[asyncio.AbstractEventLoop] = None

243 self._loop_thread: Optional[threading.Thread] = None

244

245 logger.info(f"AsyncImageLoader initialized with {max_workers} workers")

246

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

252

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")

259

260 def stop(self):

261 """Stop the async backend and cleanup resources."""

262 if self._loop is None:

263 return

264

265 logger.info("Stopping AsyncImageLoader...")

266 self._shutdown = True

267

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}")

276

277 # Stop the event loop

278 self._loop.call_soon_threadsafe(self._loop.stop)

279

280 # Wait for thread to finish

281 if self._loop_thread:

282 self._loop_thread.join(timeout=5.0)

283

284 # Shutdown executor

285 self.executor.shutdown(wait=True)

286

287 logger.info("AsyncImageLoader stopped")

288

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)

293

294 # Create priority queue

295 self._queue = asyncio.PriorityQueue()

296

297 # Start queue processor as background task

298 self._loop.create_task(self._process_queue())

299

300 # Run event loop forever (until stopped)

301 self._loop.run_forever()

302

303 # Cleanup after loop stops

304 self._loop.close()

305

306 async def _process_queue(self):

307 """Process load requests from priority queue."""

308 logger.info("Queue processor started")

309

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)

314

315 # Skip if already cancelled

316 if request.path not in self._pending_requests:

317 continue

318

319 # Process request

320 task = asyncio.create_task(self._load_image(request))

321 self._active_tasks[request.path] = task

322

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)

327

328 logger.info("Queue processor stopped")

329

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()

335

336 if tasks:

337 await asyncio.gather(*tasks, return_exceptions=True)

338

339 self._active_tasks.clear()

340 self._pending_requests.clear()

341

342 async def _load_image(self, request: LoadRequest):

343 """

344 Load and process image asynchronously.

345

346 Args:

347 request: LoadRequest containing path, size, and callback info

348 """

349 path = request.path

350 target_size = request.target_size

351

352 try:

353 # Check if shutting down

354 if self._shutdown:

355 return

356

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

363

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)

367

368 # Check again if shutting down before emitting

369 if self._shutdown:

370 return

371

372 # Cache result

373 self.cache.put(path, img, target_size)

374

375 # Emit success signal

376 self._emit_loaded(path, img, request.user_data)

377

378 logger.debug(f"Loaded: {path} (size: {img.size})")

379

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

384

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)

390

391 finally:

392 # Cleanup tracking

393 with self._lock:

394 self._pending_requests.pop(path, None)

395 self._active_tasks.pop(path, None)

396

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).

400

401 Args:

402 path: Path to image file

403 target_size: Optional target size for downsampling

404

405 Returns:

406 Processed PIL Image

407 """

408 img: Image.Image = Image.open(path)

409 img = convert_to_rgba(img)

410

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}")

418

419 return img

420

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}")

431

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}")

442

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.

452

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

458

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

465

466 path = Path(path)

467

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

473

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

483

484 # Track as pending

485 self._pending_requests[path] = request

486

487 # Submit to queue (thread-safe)

488 assert self._queue is not None

489 asyncio.run_coroutine_threadsafe(self._queue.put(request), self._loop)

490

491 logger.debug(f"Queued load: {path} (priority: {priority.name})")

492 return True

493

494 def cancel_load(self, path: Path) -> bool:

495 """

496 Cancel pending image load.

497

498 Args:

499 path: Path to image file

500

501 Returns:

502 True if load was cancelled, False if not found

503 """

504 path = Path(path)

505

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

512

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

519

520 return False

521

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 }

530

531

532class AsyncPDFGenerator(QObject):

533 """

534 Asynchronous PDF generator that doesn't block the UI.

535

536 Generates PDFs in background thread with progress updates.

537 Uses shared ImageCache to avoid redundant image loading.

538

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 """

546

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)

551

552 def __init__(self, image_cache: Optional[ImageCache] = None, max_workers: int = 2):

553 """

554 Initialize async PDF generator.

555

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__()

561

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")

565

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

571

572 # Event loop for async operations

573 self._loop: Optional[asyncio.AbstractEventLoop] = None

574 self._loop_thread: Optional[threading.Thread] = None

575

576 logger.info(f"AsyncPDFGenerator initialized with {max_workers} workers")

577

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

583

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")

590

591 def stop(self):

592 """Stop the async PDF generator and cleanup resources."""

593 if self._loop is None:

594 return

595

596 logger.info("Stopping AsyncPDFGenerator...")

597 self._shutdown = True

598

599 # Cancel active export

600 if self._current_export and not self._current_export.done():

601 self._current_export.cancel()

602

603 # Stop the event loop

604 if self._loop and not self._loop.is_closed():

605 self._loop.call_soon_threadsafe(self._loop.stop)

606

607 # Wait for thread to finish

608 if self._loop_thread:

609 self._loop_thread.join(timeout=5.0)

610

611 # Shutdown executor

612 self.executor.shutdown(wait=True)

613

614 logger.info("AsyncPDFGenerator stopped")

615

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)

620

621 # Run event loop forever (until stopped)

622 self._loop.run_forever()

623

624 # Cleanup after loop stops

625 self._loop.close()

626

627 def export_pdf(self, project, output_path: str, export_dpi: int = 300) -> bool:

628 """

629 Request PDF export (non-blocking).

630

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)

635

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

642

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

647

648 self._cancel_requested = False

649

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 )

654

655 logger.info(f"PDF export started: {output_path}")

656 return True

657

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")

665

666 async def _export_pdf_async(self, project, output_path: str, export_dpi: int):

667 """

668 Perform PDF export asynchronously.

669

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

678

679 # Create exporter

680 exporter = PDFExporter(project, export_dpi=export_dpi)

681

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

693

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 )

699

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")

709

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}")

717

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}")

725

726 finally:

727 with self._lock:

728 self._current_export = None

729

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.

733

734 This method patches the exporter to use our cached images.

735

736 Args:

737 exporter: PDFExporter instance

738 output_path: Output file path

739 progress_callback: Progress callback function

740

741 Returns:

742 Tuple of (success, warnings)

743 """

744 # Store original Image.open

745 original_open = Image.open

746

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))

751

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

760

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)

769

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

777

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()

782

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()}

Coverage for pyPhotoAlbum/async_backend.py: 73%

363 statements