dune-weaver/dune-weaver-touch/png_cache_manager.py

"""PNG Cache Manager for dune-weaver-touch

Converts WebP previews to PNG format for optimal Qt/QML compatibility.
"""

import asyncio
import os
import logging
from pathlib import Path
from typing import List
try:
    from PIL import Image
except ImportError:
    Image = None

logger = logging.getLogger(__name__)

class PngCacheManager:
    """Manages PNG cache generation from WebP sources for touch interface"""
    
    def __init__(self, cache_dir: Path = None):
        # Default to the main cache directory relative to touch app
        self.cache_dir = cache_dir or Path("../patterns/cached_images")
        self.conversion_stats = {
            "total_webp_found": 0,
            "png_already_exist": 0,
            "converted_successfully": 0,
            "conversion_errors": 0
        }
    
    async def ensure_png_cache_available(self) -> bool:
        """
        Ensure PNG previews are available for all WebP files.
        Returns True if all conversions completed successfully.
        """
        if not Image:
            logger.error("PIL (Pillow) not available - cannot convert WebP to PNG")
            return False
        
        if not self.cache_dir.exists():
            logger.info(f"Cache directory {self.cache_dir} does not exist - no conversion needed")
            return True
        
        logger.info(f"Starting PNG cache check for directory: {self.cache_dir}")
        
        # Find all WebP files that need PNG conversion
        webp_files = await self._find_webp_files_needing_conversion()
        
        if not webp_files:
            logger.info("All WebP files already have PNG equivalents")
            return True
        
        logger.info(f"Found {len(webp_files)} WebP files needing PNG conversion")
        
        # Convert WebP files to PNG in batches
        success = await self._convert_webp_to_png_batch(webp_files)
        
        # Log conversion statistics
        self._log_conversion_stats()
        
        return success
    
    async def _find_webp_files_needing_conversion(self) -> List[Path]:
        """Find WebP files that don't have corresponding PNG files"""
        def _scan_webp():
            webp_files = []
            for webp_file in self.cache_dir.rglob("*.webp"):
                # Check if corresponding PNG exists
                png_file = webp_file.with_suffix(".png")
                if not png_file.exists():
                    webp_files.append(webp_file)
                else:
                    self.conversion_stats["png_already_exist"] += 1
                self.conversion_stats["total_webp_found"] += 1
            return webp_files
        
        return await asyncio.to_thread(_scan_webp)
    
    async def _convert_webp_to_png_batch(self, webp_files: List[Path]) -> bool:
        """Convert WebP files to PNG in parallel batches"""
        batch_size = 5  # Process 5 files at a time to avoid overwhelming the system
        all_success = True
        
        for i in range(0, len(webp_files), batch_size):
            batch = webp_files[i:i + batch_size]
            batch_tasks = [self._convert_single_webp_to_png(webp_file) for webp_file in batch]
            batch_results = await asyncio.gather(*batch_tasks, return_exceptions=True)
            
            # Check results
            for webp_file, result in zip(batch, batch_results):
                if isinstance(result, Exception):
                    logger.error(f"Failed to convert {webp_file}: {result}")
                    self.conversion_stats["conversion_errors"] += 1
                    all_success = False
                elif result:
                    self.conversion_stats["converted_successfully"] += 1
                    logger.debug(f"Converted {webp_file} to PNG")
                else:
                    self.conversion_stats["conversion_errors"] += 1
                    all_success = False
            
            # Log progress
            processed = min(i + batch_size, len(webp_files))
            logger.info(f"PNG conversion progress: {processed}/{len(webp_files)} files processed")
        
        return all_success
    
    async def _convert_single_webp_to_png(self, webp_file: Path) -> bool:
        """Convert a single WebP file to PNG format"""
        try:
            png_file = webp_file.with_suffix(".png")
            
            def _convert():
                # Open WebP image and convert to PNG
                with Image.open(webp_file) as img:
                    # Convert to RGB if necessary (PNG doesn't support some WebP modes)
                    if img.mode in ('RGBA', 'LA', 'P'):
                        # Keep transparency for these modes
                        img.save(png_file, "PNG", optimize=True)
                    else:
                        # Convert to RGB for other modes
                        rgb_img = img.convert('RGB')
                        rgb_img.save(png_file, "PNG", optimize=True)
                
                # Set file permissions to match the WebP file
                try:
                    webp_stat = webp_file.stat()
                    os.chmod(png_file, webp_stat.st_mode)
                except (OSError, PermissionError):
                    # Not critical if we can't set permissions
                    pass
            
            await asyncio.to_thread(_convert)
            return True
            
        except Exception as e:
            logger.error(f"Failed to convert {webp_file} to PNG: {e}")
            return False
    
    def _log_conversion_stats(self):
        """Log conversion statistics"""
        stats = self.conversion_stats
        logger.info("PNG Cache Conversion Statistics:")
        logger.info(f"  Total WebP files found: {stats['total_webp_found']}")
        logger.info(f"  PNG files already existed: {stats['png_already_exist']}")
        logger.info(f"  Files converted successfully: {stats['converted_successfully']}")
        logger.info(f"  Conversion errors: {stats['conversion_errors']}")
        
        if stats['conversion_errors'] > 0:
            logger.warning(f"⚠️ {stats['conversion_errors']} files failed to convert")
        else:
            logger.info("✅ All WebP to PNG conversions completed successfully")
    
    async def convert_specific_pattern(self, pattern_name: str) -> bool:
        """Convert a specific pattern's WebP to PNG if needed"""
        if not Image:
            return False
        
        # Handle both hierarchical and flat naming conventions
        webp_files = []
        
        # Try hierarchical structure first
        webp_hierarchical = self.cache_dir / f"{pattern_name}.webp"
        if webp_hierarchical.exists():
            png_hierarchical = webp_hierarchical.with_suffix(".png")
            if not png_hierarchical.exists():
                webp_files.append(webp_hierarchical)
        
        # Try flattened structure
        pattern_name_flat = pattern_name.replace("/", "_").replace("\\", "_")
        webp_flat = self.cache_dir / f"{pattern_name_flat}.webp"
        if webp_flat.exists():
            png_flat = webp_flat.with_suffix(".png")
            if not png_flat.exists():
                webp_files.append(webp_flat)
        
        if not webp_files:
            return True  # No conversion needed
        
        # Convert found WebP files
        tasks = [self._convert_single_webp_to_png(webp_file) for webp_file in webp_files]
        results = await asyncio.gather(*tasks)
        
        return all(results)


async def ensure_png_cache_startup():
    """
    Startup function to ensure PNG cache is available.
    Call this during application startup.
    """
    try:
        cache_manager = PngCacheManager()
        success = await cache_manager.ensure_png_cache_available()
        
        if success:
            logger.info("PNG cache startup check completed successfully")
        else:
            logger.warning("PNG cache startup check completed with some errors")
        
        return success
    except Exception as e:
        logger.error(f"PNG cache startup check failed: {e}")
        return False