### Made with β€οΈ by builders, for builders β’ See you in July 2025!
+### Made with β€οΈ by builders, for builders
diff --git a/dream_layer_backend/.flake8 b/dream_layer_backend/.flake8 new file mode 100644 index 00000000..274a4322 --- /dev/null +++ b/dream_layer_backend/.flake8 @@ -0,0 +1,6 @@ +[flake8] +max-line-length = 120 +extend-ignore = E203,W503 +per-file-ignores = + */dream_layer_backend_utils/update_custom_workflow.py:E501 +exclude = ComfyUI/* diff --git a/dream_layer_backend/core/__init__.py b/dream_layer_backend/core/__init__.py new file mode 100644 index 00000000..5a64e36b --- /dev/null +++ b/dream_layer_backend/core/__init__.py @@ -0,0 +1,2 @@ +# Core package for DreamLayer AI + diff --git a/dream_layer_backend/core/presets.py b/dream_layer_backend/core/presets.py new file mode 100644 index 00000000..eccac057 --- /dev/null +++ b/dream_layer_backend/core/presets.py @@ -0,0 +1,398 @@ +""" +Presets Module for DreamLayer AI + +Manages version-pinned presets for reproducible generation configurations. +""" + +import hashlib +import json +import os +from datetime import datetime +from pathlib import Path +from typing import Dict, Any, Optional, List + +# Default presets file location +DEFAULT_PRESETS_FILE = Path("presets/presets.json") + + +class Preset: + """Represents a generation preset with version pinning.""" + + def __init__( + self, + name: str, + version: int = 1, + models: Optional[Dict[str, Any]] = None, + params: Optional[Dict[str, Any]] = None, + preset_hash: Optional[str] = None, + description: str = "", + created_at: Optional[str] = None, + updated_at: Optional[str] = None + ): + self.name = name + self.version = version + self.models = models or {} + self.params = params or {} + self.description = description + self.created_at = created_at or datetime.now().isoformat() + self.updated_at = updated_at or datetime.now().isoformat() + + # Compute hash if not provided + if preset_hash is None: + self.preset_hash = self._compute_hash() + else: + self.preset_hash = preset_hash + + def _compute_hash(self) -> str: + """Compute stable SHA256 hash of preset configuration.""" + # Create a stable representation for hashing (exclude name and version for stability) + hash_data = { + "models": self._sort_dict(self.models), + "params": self._sort_dict(self.params) + } + + # Convert to sorted JSON string for deterministic hashing + hash_string = json.dumps(hash_data, sort_keys=True, separators=(',', ':')) + return hashlib.sha256(hash_string.encode('utf-8')).hexdigest() + + def _sort_dict(self, d: Dict[str, Any]) -> Dict[str, Any]: + """Recursively sort dictionary for deterministic hashing.""" + if not isinstance(d, dict): + return d + + sorted_dict = {} + for key in sorted(d.keys()): + value = d[key] + if isinstance(value, dict): + sorted_dict[key] = self._sort_dict(value) + elif isinstance(value, list): + sorted_dict[key] = [self._sort_dict(item) if isinstance(item, dict) else item for item in value] + else: + sorted_dict[key] = value + + return sorted_dict + + def to_dict(self) -> Dict[str, Any]: + """Convert preset to dictionary representation.""" + return { + "name": self.name, + "version": self.version, + "models": self.models, + "params": self.params, + "preset_hash": self.preset_hash, + "description": self.description, + "created_at": self.created_at, + "updated_at": self.updated_at + } + + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> 'Preset': + """Create preset from dictionary representation.""" + return cls( + name=data["name"], + version=data.get("version", 1), + models=data.get("models", {}), + params=data.get("params", {}), + preset_hash=data.get("preset_hash"), + description=data.get("description", ""), + created_at=data.get("created_at"), + updated_at=data.get("updated_at") + ) + + def update(self, **kwargs) -> None: + """Update preset fields and recompute hash.""" + for key, value in kwargs.items(): + if key == "params" and isinstance(value, dict): + # Update params dictionary + self.params.update(value) + elif key == "models" and isinstance(value, dict): + # Update models dictionary + self.models.update(value) + elif hasattr(self, key): + setattr(self, key, value) + + # Update timestamp and recompute hash + self.updated_at = datetime.now().isoformat() + self.preset_hash = self._compute_hash() + + def is_compatible_with(self, other: 'Preset') -> bool: + """Check if this preset is compatible with another preset.""" + # Check if the configurations are compatible (same models and params) + return self.preset_hash == other.preset_hash + + +class PresetManager: + """Manages loading, saving, and operations on presets.""" + + def __init__(self, presets_file: Optional[Path] = None): + self.presets_file = presets_file or DEFAULT_PRESETS_FILE + self.presets: Dict[str, Preset] = {} + self._load_presets() + + def _load_presets(self) -> None: + """Load presets from file.""" + try: + if self.presets_file.exists(): + with open(self.presets_file, 'r', encoding='utf-8') as f: + data = json.load(f) + + # Load presets + for preset_data in data.get("presets", []): + preset = Preset.from_dict(preset_data) + self.presets[preset.name] = preset + + print(f"Loaded {len(self.presets)} presets from {self.presets_file}") + else: + print(f"Presets file not found: {self.presets_file}") + self._create_default_presets() + + except Exception as e: + print(f"Error loading presets: {e}") + self._create_default_presets() + + def _create_default_presets(self) -> None: + """Create default presets if none exist.""" + default_presets = [ + Preset( + name="default", + description="Default generation settings", + models={ + "checkpoint": "juggernautXL_v8Rundiffusion.safetensors", + "vae": "auto" + }, + params={ + "steps": 20, + "cfg": 7.0, + "sampler": "euler", + "scheduler": "normal", + "width": 512, + "height": 512, + "batch_size": 1 + } + ), + Preset( + name="high_quality", + description="High quality generation with more steps", + models={ + "checkpoint": "juggernautXL_v8Rundiffusion.safetensors", + "vae": "auto" + }, + params={ + "steps": 50, + "cfg": 7.0, + "sampler": "dpmpp_2m", + "scheduler": "karras", + "width": 1024, + "height": 1024, + "batch_size": 1 + } + ), + Preset( + name="fast", + description="Fast generation with fewer steps", + models={ + "checkpoint": "juggernautXL_v8Rundiffusion.safetensors", + "vae": "auto" + }, + params={ + "steps": 10, + "cfg": 7.0, + "sampler": "euler", + "scheduler": "normal", + "width": 512, + "height": 512, + "batch_size": 4 + } + ) + ] + + for preset in default_presets: + self.presets[preset.name] = preset + + self._save_presets() + print(f"Created {len(default_presets)} default presets") + + def _save_presets(self) -> None: + """Save presets to file.""" + try: + # Ensure directory exists + self.presets_file.parent.mkdir(parents=True, exist_ok=True) + + # Prepare data for saving + data = { + "version": "1.0", + "created_at": datetime.now().isoformat(), + "updated_at": datetime.now().isoformat(), + "presets": [preset.to_dict() for preset in self.presets.values()] + } + + with open(self.presets_file, 'w', encoding='utf-8') as f: + json.dump(data, f, indent=2, ensure_ascii=False) + + print(f"Saved {len(self.presets)} presets to {self.presets_file}") + + except Exception as e: + print(f"Error saving presets: {e}") + + def get_preset(self, name: str) -> Optional[Preset]: + """Get a preset by name.""" + return self.presets.get(name) + + def list_presets(self) -> List[str]: + """List all preset names.""" + return sorted(self.presets.keys()) + + def add_preset(self, preset: Preset) -> None: + """Add or update a preset.""" + self.presets[preset.name] = preset + self._save_presets() + print(f"Added/updated preset: {preset.name}") + + def remove_preset(self, name: str) -> bool: + """Remove a preset by name.""" + if name in self.presets: + del self.presets[name] + self._save_presets() + print(f"Removed preset: {name}") + return True + return False + + def create_preset_from_config( + self, + name: str, + config: Dict[str, Any], + description: str = "" + ) -> Preset: + """Create a new preset from a generation configuration.""" + # Extract models and params from config + models = {} + params = {} + + # Model-related keys + model_keys = ['model_name', 'vae_name', 'lora_name', 'controlnet_model'] + for key in model_keys: + if key in config and config[key]: + models[key] = config[key] + + # Parameter-related keys + param_keys = [ + 'steps', 'cfg_scale', 'sampler_name', 'scheduler', 'width', 'height', + 'batch_size', 'seed', 'denoising_strength', 'tile_size', 'tile_overlap' + ] + for key in param_keys: + if key in config and config[key] is not None: + params[key] = config[key] + + # Create preset + preset = Preset( + name=name, + models=models, + params=params, + description=description + ) + + self.add_preset(preset) + return preset + + def apply_preset_to_config(self, preset_name: str, config: Dict[str, Any]) -> Dict[str, Any]: + """Apply a preset to a generation configuration.""" + preset = self.get_preset(preset_name) + if not preset: + raise ValueError(f"Preset not found: {preset_name}") + + # Create a copy of the config + updated_config = config.copy() + + # Apply preset parameters (only if not already set in config) + for key, value in preset.params.items(): + if key not in updated_config: + updated_config[key] = value + + # Apply preset models (only if not already set in config) + for key, value in preset.models.items(): + if key == 'checkpoint': + if 'model_name' not in updated_config: + updated_config['model_name'] = value + elif key == 'vae_name': + if 'vae_name' not in updated_config: + updated_config['vae_name'] = value + elif key == 'lora_name': + if 'lora_name' not in updated_config: + updated_config['lora_name'] = value + elif key == 'controlnet_model': + if 'controlnet_model' not in updated_config: + updated_config['controlnet_model'] = value + else: + if key not in updated_config: + updated_config[key] = value + + # Add preset metadata + updated_config['preset_name'] = preset.name + updated_config['preset_hash'] = preset.preset_hash + + return updated_config + + def validate_preset(self, preset_name: str) -> Dict[str, Any]: + """Validate a preset configuration.""" + preset = self.get_preset(preset_name) + if not preset: + return {"valid": False, "error": f"Preset not found: {preset_name}"} + + # Check if preset hash is still valid + current_hash = preset._compute_hash() + hash_valid = current_hash == preset.preset_hash + + # Check if referenced models exist + missing_models = [] + for model_type, model_name in preset.models.items(): + if not self._model_exists(model_type, model_name): + missing_models.append(f"{model_type}: {model_name}") + + return { + "valid": hash_valid and len(missing_models) == 0, + "hash_valid": hash_valid, + "missing_models": missing_models, + "preset": preset.to_dict() + } + + def _model_exists(self, model_type: str, model_name: str) -> bool: + """Check if a model exists in the system.""" + # This is a simplified check - in a real implementation, + # you would check against the actual model directories + if model_name == "auto": + return True + + # For now, assume models exist if they have valid extensions + valid_extensions = {'.safetensors', '.ckpt', '.pth', '.pt', '.bin'} + return Path(model_name).suffix.lower() in valid_extensions + + +# Global preset manager instance +_preset_manager: Optional[PresetManager] = None + + +def get_preset_manager() -> PresetManager: + """Get the global preset manager instance.""" + global _preset_manager + if _preset_manager is None: + _preset_manager = PresetManager() + return _preset_manager + + +def load_presets(path: Path) -> Dict[str, Preset]: + """Load presets from a specific path.""" + manager = PresetManager(path) + return manager.presets + + +def save_presets(path: Path, presets: Dict[str, Preset]) -> None: + """Save presets to a specific path.""" + manager = PresetManager(path) + manager.presets = presets + manager._save_presets() + + +def compute_preset_hash(preset: Dict[str, Any]) -> str: + """Compute hash for a preset dictionary.""" + temp_preset = Preset.from_dict(preset) + return temp_preset.preset_hash diff --git a/dream_layer_backend/core/tiling.py b/dream_layer_backend/core/tiling.py new file mode 100644 index 00000000..1a10bb87 --- /dev/null +++ b/dream_layer_backend/core/tiling.py @@ -0,0 +1,407 @@ +""" +Tiling Module for DreamLayer AI + +Handles large image generation by tiling and blending for high-resolution outputs. +""" + +import math +import numpy as np +from PIL import Image +from typing import List, Tuple, Callable, Dict, Any, Union +import logging + +logger = logging.getLogger(__name__) + + +class TilingConfig: + """Configuration for tiled generation.""" + + def __init__( + self, + tile_size: int = 512, + overlap: int = 64, + blend_mode: str = "cosine" + ): + self.tile_size = tile_size + self.overlap = overlap + self.blend_mode = blend_mode + + # Validate parameters + if tile_size <= 0: + raise ValueError("Tile size must be positive") + if overlap < 0: + raise ValueError("Overlap must be non-negative") + if overlap >= tile_size: + raise ValueError("Overlap must be less than tile size") + if blend_mode not in ["cosine", "linear", "laplacian"]: + raise ValueError("Blend mode must be 'cosine', 'linear', or 'laplacian'") + + def to_dict(self) -> Dict[str, Any]: + """Convert config to dictionary.""" + return { + "tile_size": self.tile_size, + "overlap": self.overlap, + "blend_mode": self.blend_mode + } + + +def tile_slices( + width: int, + height: int, + tile_size: int, + overlap: int +) -> List[Tuple[int, int, int, int]]: + """ + Generate tile coordinates for an image of given dimensions. + + Args: + width: Image width + height: Image height + tile_size: Size of each tile + overlap: Overlap between tiles + + Returns: + List of (x0, y0, x1, y1) coordinates for each tile + """ + tiles = [] + + # Calculate step size (tile size minus overlap) + step = tile_size - overlap + + # Generate tiles + for y in range(0, height, step): + for x in range(0, width, step): + # Calculate tile boundaries + x1 = min(x + tile_size, width) + y1 = min(y + tile_size, height) + + # Ensure minimum tile size + if x1 - x >= overlap and y1 - y >= overlap: + tiles.append((x, y, x1, y1)) + + return tiles + + +def create_blend_mask( + tile_size: int, + overlap: int, + mode: str = "cosine" +) -> np.ndarray: + """ + Create a blend mask for seamless tile joining. + + Args: + tile_size: Size of the tile + overlap: Overlap size + mode: Blending mode ('cosine', 'linear', 'laplacian') + + Returns: + Blend mask as numpy array + """ + if overlap == 0: + return np.ones((tile_size, tile_size)) + + mask = np.ones((tile_size, tile_size)) + + # Create overlap regions + if overlap > 0: + # Left edge + for i in range(overlap): + if mode == "cosine": + weight = 0.5 * (1 - math.cos(math.pi * i / overlap)) + elif mode == "linear": + weight = i / overlap + else: # laplacian + weight = 1 - math.exp(-i / (overlap * 0.3)) + + mask[:, i] *= weight + + # Right edge + for i in range(overlap): + if mode == "cosine": + weight = 0.5 * (1 - math.cos(math.pi * (overlap - i - 1) / overlap)) + elif mode == "linear": + weight = (overlap - i - 1) / overlap + else: # laplacian + weight = 1 - math.exp(-(overlap - i - 1) / (overlap * 0.3)) + + mask[:, tile_size - overlap + i] *= weight + + # Top edge + for i in range(overlap): + if mode == "cosine": + weight = 0.5 * (1 - math.cos(math.pi * i / overlap)) + elif mode == "linear": + weight = i / overlap + else: # laplacian + weight = 1 - math.exp(-i / (overlap * 0.3)) + + mask[i, :] *= weight + + # Bottom edge + for i in range(overlap): + if mode == "cosine": + weight = 0.5 * (1 - math.cos(math.pi * (overlap - i - 1) / overlap)) + elif mode == "linear": + weight = (overlap - i - 1) / overlap + else: # laplacian + weight = 1 - math.exp(-(overlap - i - 1) / (overlap * 0.3)) + + mask[tile_size - overlap + i, :] *= weight + + return mask + + +def blend_paste( + canvas: np.ndarray, + tile_img: np.ndarray, + rect: Tuple[int, int, int, int], + overlap: int, + mode: str = "cosine" +) -> None: + """ + Blend and paste a tile onto the canvas with seamless joining. + + Args: + canvas: Target canvas array + tile_img: Tile image array + rect: (x0, y0, x1, y1) coordinates for placement + overlap: Overlap size for blending + mode: Blending mode + """ + x0, y0, x1, y1 = rect + tile_height, tile_width = tile_img.shape[:2] + + # Create blend mask for this tile + blend_mask = create_blend_mask(tile_width, overlap, mode) + + # Apply blend mask to tile + if len(tile_img.shape) == 3: # Color image + blended_tile = tile_img * blend_mask[:, :, np.newaxis] + else: # Grayscale image + blended_tile = tile_img * blend_mask + + # Extract region from canvas + canvas_region = canvas[y0:y1, x0:x1] + + # Blend with existing content + if overlap > 0: + # Create overlap mask for existing content + overlap_mask = 1.0 - blend_mask[:y1-y0, :x1-x0] + + if len(canvas_region.shape) == 3: + overlap_mask = overlap_mask[:, :, np.newaxis] + + # Blend existing content with new tile + canvas_region = canvas_region * overlap_mask + blended_tile[:y1-y0, :x1-x0] + else: + canvas_region = blended_tile[:y1-y0, :x1-x0] + + # Update canvas + canvas[y0:y1, x0:x1] = canvas_region + + +def process_tiled( + generate_fn: Callable, + width: int, + height: int, + tile_size: int = 512, + overlap: int = 64, + blend_mode: str = "cosine", + **gen_kwargs +) -> Union[np.ndarray, Image.Image]: + """ + Process large image generation using tiling and blending. + + Args: + generate_fn: Function that generates a single tile + width: Target image width + height: Target image height + tile_size: Size of each tile + overlap: Overlap between tiles + blend_mode: Blending mode for seamless joins + **gen_kwargs: Additional arguments passed to generate_fn + + Returns: + Generated image as numpy array or PIL Image + """ + config = TilingConfig(tile_size, overlap, blend_mode) + + # Generate tile coordinates + tiles = tile_slices(width, height, tile_size, overlap) + logger.info(f"Processing {len(tiles)} tiles for {width}x{height} image") + + # Create output canvas + if 'crop' in gen_kwargs: + # Check if generate_fn expects crop parameter + sample_kwargs = gen_kwargs.copy() + sample_kwargs['crop'] = (0, 0, min(tile_size, width), min(tile_size, height)) + sample_result = generate_fn(**sample_kwargs) + + if isinstance(sample_result, Image.Image): + sample_array = np.array(sample_result) + output_dtype = sample_array.dtype + output_channels = sample_array.shape[2] if len(sample_array.shape) > 2 else 1 + else: + output_dtype = sample_result.dtype + output_channels = sample_result.shape[2] if len(sample_result.shape) > 2 else 1 + + if output_channels == 1: + canvas = np.zeros((height, width), dtype=output_dtype) + else: + canvas = np.zeros((height, width, output_channels), dtype=output_dtype) + else: + # Default to RGB + canvas = np.zeros((height, width, 3), dtype=np.uint8) + + # Process each tile + for i, (x0, y0, x1, y1) in enumerate(tiles): + logger.info(f"Processing tile {i+1}/{len(tiles)}: ({x0},{y0}) to ({x1},{y1})") + + try: + # Generate tile + if 'crop' in gen_kwargs: + # Pass crop coordinates to generate function + tile_kwargs = gen_kwargs.copy() + tile_kwargs['crop'] = (x0, y0, x1, y1) + tile_result = generate_fn(**tile_kwargs) + else: + # Generate full tile and crop + tile_result = generate_fn(**gen_kwargs) + if isinstance(tile_result, Image.Image): + tile_result = tile_result.crop((0, 0, x1-x0, y1-y0)) + else: + tile_result = tile_result[:y1-y0, :x1-x0] + + # Convert to numpy array if needed + if isinstance(tile_result, Image.Image): + tile_array = np.array(tile_result) + else: + tile_array = tile_result + + # Ensure tile dimensions match expected size + expected_height, expected_width = y1-y0, x1-x0 + if tile_array.shape[:2] != (expected_height, expected_width): + logger.warning(f"Tile size mismatch: expected {expected_height}x{expected_width}, got {tile_array.shape[:2]}") + # Resize tile if needed + if isinstance(tile_result, Image.Image): + tile_result = tile_result.resize((expected_width, expected_height), Image.Resampling.LANCZOS) + tile_array = np.array(tile_result) + else: + # Simple resize for numpy arrays + from scipy.ndimage import zoom + zoom_factors = [expected_height / tile_array.shape[0], expected_width / tile_array.shape[1]] + if len(tile_array.shape) == 3: + zoom_factors.append(1) + tile_array = zoom(tile_array, zoom_factors, order=1) + + # Blend and paste tile + blend_paste(canvas, tile_array, (x0, y0, x1, y1), overlap, blend_mode) + + except Exception as e: + logger.error(f"Error processing tile {i+1}: {e}") + # Fill with error pattern + error_pattern = np.full((y1-y0, x1-x0, 3) if len(canvas.shape) == 3 else (y1-y0, x1-x0), + 128, dtype=canvas.dtype) + canvas[y0:y1, x0:x1] = error_pattern + + # Convert back to PIL Image if the generate function returns PIL Images + try: + # Test if generate function returns PIL Image + test_result = generate_fn(**{**gen_kwargs, 'crop': (0, 0, 1, 1)}) + if isinstance(test_result, Image.Image): + return Image.fromarray(canvas) + except: + pass + + return canvas + + +def calculate_optimal_tile_size( + width: int, + height: int, + max_tile_size: int = 512, + min_tile_size: int = 256 +) -> Tuple[int, int]: + """ + Calculate optimal tile size and overlap for given dimensions. + + Args: + width: Image width + height: Image height + max_tile_size: Maximum tile size + min_tile_size: Minimum tile size + + Returns: + Tuple of (tile_size, overlap) + """ + # Start with maximum tile size + tile_size = max_tile_size + + # Calculate overlap as 1/8 of tile size + overlap = tile_size // 8 + + # Ensure minimum tile size + if tile_size - overlap < min_tile_size: + tile_size = min_tile_size + overlap + overlap = tile_size // 8 + + return tile_size, overlap + + +def validate_tiling_config( + width: int, + height: int, + tile_size: int, + overlap: int +) -> Dict[str, Any]: + """ + Validate tiling configuration for given image dimensions. + + Args: + width: Image width + height: Image height + tile_size: Tile size + overlap: Overlap size + + Returns: + Dictionary with validation results + """ + # Check if tile size is larger than image dimensions + if tile_size > width or tile_size > height: + return { + "valid": False, + "tile_count": 0, + "coverage": 0.0, + "has_gaps": True, + "tiles": [], + "efficiency": 0.0, + "error": "Tile size larger than image dimensions" + } + + tiles = tile_slices(width, height, tile_size, overlap) + + # Calculate coverage + total_tile_area = sum((x1-x0) * (y1-y0) for x0, y0, x1, y1 in tiles) + image_area = width * height + coverage = total_tile_area / image_area + + # Check for gaps + has_gaps = False + if tiles: + # Simple gap detection - check if tiles cover the entire image + min_x = min(x0 for x0, y0, x1, y1 in tiles) + max_x = max(x1 for x0, y0, x1, y1 in tiles) + min_y = min(y0 for x0, y0, x1, y1 in tiles) + max_y = max(y1 for x0, y0, x1, y1 in tiles) + + has_gaps = min_x > 0 or max_x < width or min_y > 0 or max_y < height + + return { + "valid": len(tiles) > 0 and not has_gaps, + "tile_count": len(tiles), + "coverage": coverage, + "has_gaps": has_gaps, + "tiles": tiles, + "efficiency": image_area / (len(tiles) * tile_size * tile_size) if tiles else 0 + } diff --git a/dream_layer_backend/metrics/__init__.py b/dream_layer_backend/metrics/__init__.py new file mode 100644 index 00000000..59d26b0b --- /dev/null +++ b/dream_layer_backend/metrics/__init__.py @@ -0,0 +1,2 @@ +# Metrics package for DreamLayer AI + diff --git a/dream_layer_backend/metrics/clip_score.py b/dream_layer_backend/metrics/clip_score.py new file mode 100644 index 00000000..4c941c05 --- /dev/null +++ b/dream_layer_backend/metrics/clip_score.py @@ -0,0 +1,186 @@ +""" +CLIP Score Module for DreamLayer AI + +Provides CLIP-based text-image similarity scoring for quality assessment. +""" + +import logging +from typing import List, Optional, Tuple +from PIL import Image +import numpy as np + +logger = logging.getLogger(__name__) + +# Pinned model information +CLIP_MODEL_ID = "openai/clip-vit-large-patch14" +CLIP_MODEL_HASH = "sha256:6c7ba7f6" # Placeholder hash + +try: + import torch + import transformers + from transformers import CLIPProcessor, CLIPModel + CLIP_AVAILABLE = True +except ImportError: + CLIP_AVAILABLE = False + logger.warning("CLIP dependencies not available. Install with: pip install transformers torch") + + +class CLIPScorer: + """CLIP-based text-image similarity scorer.""" + + def __init__(self, model_id: str = CLIP_MODEL_ID): + self.model_id = model_id + self.model = None + self.processor = None + self.device = None + + if CLIP_AVAILABLE: + self._load_model() + else: + logger.warning("CLIP not available - install required dependencies") + + def _load_model(self) -> None: + """Load CLIP model and processor.""" + try: + logger.info(f"Loading CLIP model: {self.model_id}") + + # Set device + self.device = "cuda" if torch.cuda.is_available() else "cpu" + logger.info(f"Using device: {self.device}") + + # Load model and processor + self.model = CLIPModel.from_pretrained(self.model_id).to(self.device) + self.processor = CLIPProcessor.from_pretrained(self.model_id) + + logger.info("CLIP model loaded successfully") + + except Exception as e: + logger.error(f"Failed to load CLIP model: {e}") + self.model = None + self.processor = None + + def clip_text_image_similarity( + self, + images: List[Image.Image], + prompts: List[str], + batch_size: int = 8 + ) -> List[Optional[float]]: + """ + Compute CLIP similarity scores between text prompts and images. + + Args: + images: List of PIL images + prompts: List of text prompts + batch_size: Batch size for processing + + Returns: + List of similarity scores (0.0 to 1.0) or None if CLIP not available + """ + if not CLIP_AVAILABLE or self.model is None: + logger.warning("CLIP not available - returning None scores") + return [None] * len(images) + + if len(images) != len(prompts): + raise ValueError("Number of images must match number of prompts") + + scores = [] + + try: + # Process in batches + for i in range(0, len(images), batch_size): + batch_images = images[i:i + batch_size] + batch_prompts = prompts[i:i + batch_size] + + # Process batch + batch_scores = self._process_batch(batch_images, batch_prompts) + scores.extend(batch_scores) + + except Exception as e: + logger.error(f"Error computing CLIP scores: {e}") + # Return None on error + scores = [None] * len(images) + + return scores + + def _process_batch( + self, + images: List[Image.Image], + prompts: List[str] + ) -> List[float]: + """Process a batch of images and prompts.""" + try: + # Prepare inputs + inputs = self.processor( + text=prompts, + images=images, + return_tensors="pt", + padding=True, + truncation=True + ).to(self.device) + + # Get embeddings + with torch.no_grad(): + text_features = self.model.get_text_features(**inputs) + image_features = self.model.get_image_features(**inputs) + + # Normalize features + text_features = text_features / text_features.norm(dim=-1, keepdim=True) + image_features = image_features / image_features.norm(dim=-1, keepdim=True) + + # Compute cosine similarity + similarity = (text_features @ image_features.T).diagonal() + + # Convert to list and ensure values are in [0, 1] + scores = similarity.cpu().numpy().tolist() + scores = [max(0.0, min(1.0, (score + 1.0) / 2.0)) for score in scores] + + return scores + + except Exception as e: + logger.error(f"Error processing CLIP batch: {e}") + return [None] * len(images) + + def get_model_info(self) -> dict: + """Get information about the CLIP model.""" + return { + "model_id": self.model_id, + "model_hash": self.model_hash, + "available": CLIP_AVAILABLE and self.model is not None, + "device": self.device if CLIP_AVAILABLE else None + } + + @property + def model_hash(self) -> str: + """Get the model hash.""" + return CLIP_MODEL_HASH + + +def get_clip_scorer() -> Optional[CLIPScorer]: + """Get a CLIP scorer instance.""" + if CLIP_AVAILABLE: + return CLIPScorer() + return None + + +def clip_text_image_similarity( + images: List[Image.Image], + prompts: List[str], + batch_size: int = 8 +) -> List[Optional[float]]: + """ + Convenience function for computing CLIP similarity scores. + + Args: + images: List of PIL images + prompts: List of text prompts + batch_size: Batch size for processing + + Returns: + List of similarity scores (0.0 to 1.0) or None if CLIP not available + """ + scorer = get_clip_scorer() + if scorer: + return scorer.clip_text_image_similarity(images, prompts, batch_size) + else: + logger.warning("CLIP not available - returning None scores") + return [None] * len(images) diff --git a/dream_layer_backend/metrics/ssim_lpips.py b/dream_layer_backend/metrics/ssim_lpips.py new file mode 100644 index 00000000..c2732bd8 --- /dev/null +++ b/dream_layer_backend/metrics/ssim_lpips.py @@ -0,0 +1,330 @@ +""" +SSIM and LPIPS Metrics Module for DreamLayer AI + +Provides structural similarity and perceptual quality metrics for image assessment. +""" + +import logging +from typing import Optional, Tuple, Union +from PIL import Image +import numpy as np + +logger = logging.getLogger(__name__) + +# Pinned model information +LPIPS_MODEL_ID = "alex" # Default LPIPS network +LPIPS_MODEL_HASH = "sha256:lpips_alex_v0.1" # Placeholder hash + +# SSIM availability +try: + from skimage.metrics import structural_similarity as ssim + SSIM_AVAILABLE = True +except ImportError: + SSIM_AVAILABLE = False + logger.warning("SSIM not available. Install with: pip install scikit-image") + +# LPIPS availability +try: + import lpips + LPIPS_AVAILABLE = True +except ImportError: + LPIPS_AVAILABLE = False + logger.warning("LPIPS not available. Install with: pip install lpips") + + +class SSIMScorer: + """Structural Similarity Index scorer.""" + + def __init__(self): + self.available = SSIM_AVAILABLE + if not self.available: + logger.warning("SSIM not available - install scikit-image") + + def compute_ssim( + self, + img_a: Union[np.ndarray, Image.Image], + img_b: Union[np.ndarray, Image.Image], + **kwargs + ) -> Optional[float]: + """ + Compute SSIM between two images. + + Args: + img_a: First image (numpy array or PIL Image) + img_b: Second image (numpy array or PIL Image) + **kwargs: Additional SSIM parameters + + Returns: + SSIM score (0.0 to 1.0, higher is better) or None if SSIM not available + """ + if not self.available: + logger.warning("SSIM not available - returning None") + return None + + try: + # Convert to numpy arrays + if isinstance(img_a, Image.Image): + img_a = np.array(img_a) + if isinstance(img_b, Image.Image): + img_b = np.array(img_b) + + # Ensure same shape + if img_a.shape != img_b.shape: + logger.warning("Image shapes don't match, resizing img_b to match img_a") + from skimage.transform import resize + img_b = resize(img_b, img_a.shape, preserve_range=True) + + # Convert to grayscale if needed + if len(img_a.shape) == 3 and img_a.shape[2] == 3: + from skimage.color import rgb2gray + img_a = rgb2gray(img_a) + img_b = rgb2gray(img_b) + + # Set data_range based on image type + if img_a.dtype == np.uint8: + data_range = 255 + elif img_a.dtype == np.float32 or img_a.dtype == np.float64: + data_range = 1.0 + else: + data_range = img_a.max() - img_a.min() + + # Compute SSIM with proper data_range + score = ssim(img_a, img_b, data_range=data_range, **kwargs) + return float(score) + + except Exception as e: + logger.error(f"Error computing SSIM: {e}") + return None + + def get_info(self) -> dict: + """Get information about SSIM availability.""" + return { + "available": self.available, + "dependencies": { + "scikit-image": "Available" if SSIM_AVAILABLE else "Not installed" + } + } + + +class LPIPSScorer: + """Learned Perceptual Image Patch Similarity scorer.""" + + def __init__(self, net: str = LPIPS_MODEL_ID): + self.net = net + self.available = LPIPS_AVAILABLE + self.model = None + + if self.available: + self._load_model() + else: + logger.warning("LPIPS not available - install lpips") + + def _load_model(self) -> None: + """Load LPIPS model.""" + try: + logger.info(f"Loading LPIPS model: {self.net}") + self.model = lpips.LPIPS(net=self.net) + logger.info("LPIPS model loaded successfully") + + except Exception as e: + logger.error(f"Failed to load LPIPS model: {e}") + self.model = None + self.available = False + + def compute_lpips( + self, + img_a: Union[np.ndarray, Image.Image], + img_b: Union[np.ndarray, Image.Image], + net: Optional[str] = None + ) -> Optional[float]: + """ + Compute LPIPS between two images. + + Args: + img_a: First image (numpy array or PIL Image) + img_b: Second image (numpy array or PIL Image) + net: LPIPS network to use (optional) + + Returns: + LPIPS score (0.0 is identical, higher is more different) or None if LPIPS not available + """ + if not self.available or self.model is None: + logger.warning("LPIPS not available - returning None") + return None + + try: + # Convert to numpy arrays if needed + if isinstance(img_a, Image.Image): + img_a = np.array(img_a) + if isinstance(img_b, Image.Image): + img_b = np.array(img_b) + + # Ensure images are in the right format for LPIPS + if len(img_a.shape) == 3 and img_a.shape[2] == 3: + # Convert to RGB format expected by LPIPS + img_a = img_a.transpose(2, 0, 1) # HWC to CHW + img_b = img_b.transpose(2, 0, 1) # HWC to CHW + else: + # Grayscale - convert to RGB + img_a = np.stack([img_a] * 3, axis=0) + img_b = np.stack([img_b] * 3, axis=0) + + # Normalize to [-1, 1] range + img_a = (img_a / 127.5) - 1.0 + img_b = (img_b / 127.5) - 1.0 + + # Convert to torch tensors + import torch + img_a_tensor = torch.from_numpy(img_a).float().unsqueeze(0) + img_b_tensor = torch.from_numpy(img_b).float().unsqueeze(0) + + # Compute LPIPS + with torch.no_grad(): + score = self.model(img_a_tensor, img_b_tensor) + return float(score.item()) + + except Exception as e: + logger.error(f"Error computing LPIPS: {e}") + return None + + def get_info(self) -> dict: + """Get information about LPIPS availability.""" + return { + "available": self.available, + "net": self.net, + "dependencies": { + "lpips": "Available" if LPIPS_AVAILABLE else "Not installed" + } + } + + +class QualityMetrics: + """Combined quality metrics calculator.""" + + def __init__(self): + self.ssim_scorer = SSIMScorer() + self.lpips_scorer = LPIPSScorer() + + def compute_all_metrics( + self, + img_a: Union[np.ndarray, Image.Image], + img_b: Union[np.ndarray, Image.Image] + ) -> dict: + """ + Compute all available quality metrics between two images. + + Args: + img_a: First image + img_b: Second image + + Returns: + Dictionary with metric results + """ + results = {} + + # Compute SSIM + if self.ssim_scorer.available: + results["ssim"] = self.ssim_scorer.compute_ssim(img_a, img_b) + else: + results["ssim"] = None + + # Compute LPIPS + if self.lpips_scorer.available: + results["lpips"] = self.lpips_scorer.compute_lpips(img_a, img_b) + else: + results["lpips"] = None + + return results + + def get_metrics_info(self) -> dict: + """Get information about available metrics.""" + return { + "ssim": self.ssim_scorer.get_info(), + "lpips": self.lpips_scorer.get_info() + } + + +# Global instances +_ssim_scorer: Optional[SSIMScorer] = None +_lpips_scorer: Optional[LPIPSScorer] = None +_quality_metrics: Optional[QualityMetrics] = None + + +def get_ssim_scorer() -> SSIMScorer: + """Get the global SSIM scorer instance.""" + global _ssim_scorer + if _ssim_scorer is None: + _ssim_scorer = SSIMScorer() + return _ssim_scorer + + +def get_lpips_scorer() -> LPIPSScorer: + """Get the global LPIPS scorer instance.""" + global _lpips_scorer + if _lpips_scorer is None: + _lpips_scorer = LPIPSScorer() + return _lpips_scorer + + +def get_quality_metrics() -> QualityMetrics: + """Get the global quality metrics instance.""" + global _quality_metrics + if _quality_metrics is None: + _quality_metrics = QualityMetrics() + return _quality_metrics + + +def compute_ssim( + img_a: Union[np.ndarray, Image.Image], + img_b: Union[np.ndarray, Image.Image] +) -> Optional[float]: + """ + Convenience function for computing SSIM. + + Args: + img_a: First image + img_b: Second image + + Returns: + SSIM score or None if SSIM not available + """ + scorer = get_ssim_scorer() + return scorer.compute_ssim(img_a, img_b) + + +def compute_lpips( + img_a: Union[np.ndarray, Image.Image], + img_b: Union[np.ndarray, Image.Image], + net: str = "alex" +) -> Optional[float]: + """ + Convenience function for computing LPIPS. + + Args: + img_a: First image + img_b: Second image + net: LPIPS network to use + + Returns: + LPIPS score or None if LPIPS not available + """ + scorer = get_lpips_scorer() + return scorer.compute_lpips(img_a, img_b, net) + + +def compute_all_quality_metrics( + img_a: Union[np.ndarray, Image.Image], + img_b: Union[np.ndarray, Image.Image] +) -> dict: + """ + Convenience function for computing all quality metrics. + + Args: + img_a: First image + img_b: Second image + + Returns: + Dictionary with all metric results + """ + metrics = get_quality_metrics() + return metrics.compute_all_metrics(img_a, img_b) diff --git a/dream_layer_backend/presets/presets.json b/dream_layer_backend/presets/presets.json new file mode 100644 index 00000000..b986732f --- /dev/null +++ b/dream_layer_backend/presets/presets.json @@ -0,0 +1,70 @@ +{ + "version": "1.0", + "created_at": "2025-08-13T18:23:12.128102", + "updated_at": "2025-08-13T18:23:12.128102", + "presets": [ + { + "name": "default", + "version": 1, + "models": { + "checkpoint": "juggernautXL_v8Rundiffusion.safetensors", + "vae": "auto" + }, + "params": { + "steps": 20, + "cfg": 7.0, + "sampler": "euler", + "scheduler": "normal", + "width": 512, + "height": 512, + "batch_size": 1 + }, + "preset_hash": "e8274633b8816637e37379177da285d302abb9d1b58100d6c7daeb86956ecf28", + "description": "Default generation settings", + "created_at": "2025-08-13T18:23:12.128102", + "updated_at": "2025-08-13T18:23:12.128102" + }, + { + "name": "high_quality", + "version": 1, + "models": { + "checkpoint": "juggernautXL_v8Rundiffusion.safetensors", + "vae": "auto" + }, + "params": { + "steps": 50, + "cfg": 7.0, + "sampler": "dpmpp_2m", + "scheduler": "karras", + "width": 1024, + "height": 1024, + "batch_size": 1 + }, + "preset_hash": "033bb906290658a7aeed128265a0219b40293016b2cd8fc75f7a6ae39f17d3c5", + "description": "High quality generation with more steps", + "created_at": "2025-08-13T18:23:12.128102", + "updated_at": "2025-08-13T18:23:12.128102" + }, + { + "name": "fast", + "version": 1, + "models": { + "checkpoint": "juggernautXL_v8Rundiffusion.safetensors", + "vae": "auto" + }, + "params": { + "steps": 10, + "cfg": 7.0, + "sampler": "euler", + "scheduler": "normal", + "width": 512, + "height": 512, + "batch_size": 4 + }, + "preset_hash": "6617fc7c9cc017f5f1e4f74ced2bf9bba7ea659523ae7bb94b424c79036e0255", + "description": "Fast generation with fewer steps", + "created_at": "2025-08-13T18:23:12.128102", + "updated_at": "2025-08-13T18:23:12.128102" + } + ] +} \ No newline at end of file diff --git a/dream_layer_backend/requirements.txt b/dream_layer_backend/requirements.txt index ba3453bc..3ea7365e 100644 --- a/dream_layer_backend/requirements.txt +++ b/dream_layer_backend/requirements.txt @@ -2,6 +2,26 @@ flask>=3.0.0 flask-cors>=4.0.0 pillow>=10.0.0 requests>=2.31.0 -python-dotenv>=1.0.0 +python-dotenv>=7.0.0 pytest>=7.8.0 -pytest-mock>=3.12.0 \ No newline at end of file +pytest-mock>=3.12.0 + +# Optional dependencies for quality metrics +# Uncomment the lines below to enable additional features: + +# For SSIM (Structural Similarity Index) +# scikit-image>=0.19.0 + +# For LPIPS (Learned Perceptual Image Patch Similarity) +# lpips>=0.1.4 +# torch>=1.9.0 +# torchvision>=0.10.0 + +# For CLIP scoring +# transformers>=4.20.0 +# ftfy>=6.1.0 +# regex>=2022.1.18 + +# For tiling and image processing +# numpy>=1.21.0 +# scipy>=1.7.0 \ No newline at end of file diff --git a/dream_layer_backend/tests/test_presets_e2e.py b/dream_layer_backend/tests/test_presets_e2e.py new file mode 100644 index 00000000..095fae57 --- /dev/null +++ b/dream_layer_backend/tests/test_presets_e2e.py @@ -0,0 +1,321 @@ +""" +End-to-end tests for Presets functionality. +""" + +import json +import tempfile +from pathlib import Path +from unittest.mock import patch, MagicMock + +import pytest + +from core.presets import ( + Preset, PresetManager, get_preset_manager, + load_presets, save_presets, compute_preset_hash +) + + +class TestPresetsE2E: + """End-to-end test cases for presets functionality.""" + + @pytest.fixture + def temp_presets_file(self): + """Create a temporary presets file.""" + with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as temp_file: + presets_path = Path(temp_file.name) + + try: + yield presets_path + finally: + presets_path.unlink(missing_ok=True) + + @pytest.fixture + def sample_config(self): + """Sample generation configuration.""" + return { + "model_name": "test_model.safetensors", + "vae_name": "test_vae.safetensors", + "steps": 20, + "cfg_scale": 7.0, + "sampler_name": "euler", + "scheduler": "normal", + "width": 512, + "height": 512, + "batch_size": 1, + "seed": 42 + } + + def test_preset_creation_and_application(self, temp_presets_file, sample_config): + """Test creating a preset and applying it to a config.""" + # Create preset manager + manager = PresetManager(temp_presets_file) + + # Create preset from config + preset = manager.create_preset_from_config( + name="test_preset", + config=sample_config, + description="Test preset for E2E testing" + ) + + # Verify preset was created + assert preset.name == "test_preset" + assert preset.preset_hash is not None + assert preset.params["steps"] == 20 + assert preset.params["cfg_scale"] == 7.0 + assert preset.models["model_name"] == "test_model.safetensors" + + # Apply preset to a new config + new_config = {"seed": 100} # Minimal config + updated_config = manager.apply_preset_to_config("test_preset", new_config) + + # Verify preset was applied + assert updated_config["preset_name"] == "test_preset" + assert updated_config["preset_hash"] == preset.preset_hash + assert updated_config["steps"] == 20 + assert updated_config["cfg_scale"] == 7.0 + assert updated_config["model_name"] == "test_model.safetensors" + assert updated_config["seed"] == 100 # Original value preserved + + # Verify preset is in manager + assert "test_preset" in manager.list_presets() + assert manager.get_preset("test_preset") is not None + + def test_preset_hash_stability(self, temp_presets_file): + """Test that preset hashes are stable across runs.""" + # Create preset manager + manager = PresetManager(temp_presets_file) + + # Create preset with specific configuration + config = { + "model_name": "stable_model.safetensors", + "steps": 25, + "cfg_scale": 8.0, + "sampler_name": "dpmpp_2m", + "width": 1024, + "height": 1024 + } + + preset1 = manager.create_preset_from_config("stable_preset", config) + hash1 = preset1.preset_hash + + # Create another preset with same config + preset2 = manager.create_preset_from_config("stable_preset2", config) + hash2 = preset2.preset_hash + + # Hashes should be identical for identical configs + assert hash1 == hash2 + + # Verify hash computation function + computed_hash = compute_preset_hash(preset1.to_dict()) + assert computed_hash == hash1 + + def test_preset_versioning(self, temp_presets_file): + """Test preset version management.""" + # Create preset manager + manager = PresetManager(temp_presets_file) + + # Create initial preset + config = {"steps": 20, "cfg_scale": 7.0} + preset1 = manager.create_preset_from_config("versioned_preset", config) + assert preset1.version == 1 + original_hash = preset1.preset_hash + + # Update preset with new params + preset1.update(params={"steps": 30, "cfg_scale": 8.0}) + manager.add_preset(preset1) + + # Verify params were updated + updated_preset = manager.get_preset("versioned_preset") + assert updated_preset.params["steps"] == 30 + assert updated_preset.params["cfg_scale"] == 8.0 + + # Hash should have changed due to params update + assert updated_preset.preset_hash != original_hash + + def test_preset_compatibility(self, temp_presets_file): + """Test preset compatibility checking.""" + # Create preset manager + manager = PresetManager(temp_presets_file) + + # Create two presets + config1 = {"steps": 20, "cfg_scale": 7.0} + config2 = {"steps": 20, "cfg_scale": 7.0} + + preset1 = manager.create_preset_from_config("compat_preset1", config1) + preset2 = manager.create_preset_from_config("compat_preset2", config2) + + # Presets with same config should be compatible + assert preset1.is_compatible_with(preset2) + + # Update one preset with different params + preset1.update(params={"steps": 30, "cfg_scale": 8.0}) + assert not preset1.is_compatible_with(preset2) + + def test_preset_validation(self, temp_presets_file): + """Test preset validation functionality.""" + # Create preset manager + manager = PresetManager(temp_presets_file) + + # Create valid preset + config = {"model_name": "valid_model.safetensors", "steps": 20} + preset = manager.create_preset_from_config("valid_preset", config) + + # Validate preset + validation = manager.validate_preset("valid_preset") + assert validation["valid"] is True + assert validation["hash_valid"] is True + assert len(validation["missing_models"]) == 0 + + # Test validation of non-existent preset + validation = manager.validate_preset("nonexistent_preset") + assert validation["valid"] is False + assert "Preset not found" in validation["error"] + + def test_preset_persistence(self, temp_presets_file): + """Test that presets are properly saved and loaded.""" + # Create preset manager and add presets + manager = PresetManager(temp_presets_file) + + configs = [ + {"steps": 20, "cfg_scale": 7.0}, + {"steps": 50, "cfg_scale": 8.0}, + {"steps": 10, "cfg_scale": 6.0} + ] + + for i, config in enumerate(configs): + manager.create_preset_from_config(f"persistent_preset_{i}", config) + + # Verify presets were saved (3 new + 3 default presets) + assert len(manager.list_presets()) == 6 + + # Create new manager instance to test loading + new_manager = PresetManager(temp_presets_file) + + # Verify presets were loaded (3 new + 3 default presets) + assert len(new_manager.list_presets()) == 6 + for i in range(3): + preset_name = f"persistent_preset_{i}" + assert preset_name in new_manager.list_presets() + + # Verify preset content + preset = new_manager.get_preset(preset_name) + assert preset.params["steps"] == configs[i]["steps"] + assert preset.params["cfg_scale"] == configs[i]["cfg_scale"] + + def test_preset_removal(self, temp_presets_file): + """Test preset removal functionality.""" + # Create preset manager + manager = PresetManager(temp_presets_file) + + # Add preset + config = {"steps": 20, "cfg_scale": 7.0} + manager.create_preset_from_config("removable_preset", config) + + # Verify preset exists + assert "removable_preset" in manager.list_presets() + + # Remove preset + success = manager.remove_preset("removable_preset") + assert success is True + + # Verify preset was removed + assert "removable_preset" not in manager.list_presets() + assert manager.get_preset("removable_preset") is None + + # Test removing non-existent preset + success = manager.remove_preset("nonexistent_preset") + assert success is False + + def test_global_preset_manager(self): + """Test global preset manager functionality.""" + # Get global manager + manager = get_preset_manager() + + # Verify it's a PresetManager instance + assert isinstance(manager, PresetManager) + + # Verify it has default presets + presets = manager.list_presets() + assert len(presets) > 0 + assert "default" in presets + + def test_load_save_presets_functions(self, temp_presets_file): + """Test load_presets and save_presets utility functions.""" + # Create some presets + presets_data = { + "test_preset1": Preset( + name="test_preset1", + params={"steps": 20, "cfg_scale": 7.0} + ), + "test_preset2": Preset( + name="test_preset2", + params={"steps": 50, "cfg_scale": 8.0} + ) + } + + # Save presets + save_presets(temp_presets_file, presets_data) + + # Load presets + loaded_presets = load_presets(temp_presets_file) + + # Verify presets were loaded correctly + assert len(loaded_presets) == 2 + assert "test_preset1" in loaded_presets + assert "test_preset2" in loaded_presets + + # Verify preset content + preset1 = loaded_presets["test_preset1"] + assert preset1.params["steps"] == 20 + assert preset1.params["cfg_scale"] == 7.0 + + def test_preset_with_advanced_config(self, temp_presets_file): + """Test preset creation with advanced configuration options.""" + # Create preset manager + manager = PresetManager(temp_presets_file) + + # Advanced config with various parameter types + advanced_config = { + "model_name": "advanced_model.safetensors", + "vae_name": "advanced_vae.safetensors", + "lora_name": "advanced_lora.safetensors", + "controlnet_model": "advanced_controlnet.safetensors", + "steps": 30, + "cfg_scale": 8.5, + "sampler_name": "dpmpp_2m", + "scheduler": "karras", + "width": 1024, + "height": 1024, + "batch_size": 2, + "seed": 12345, + "denoising_strength": 0.75, + "tile_size": 512, + "tile_overlap": 64 + } + + # Create preset + preset = manager.create_preset_from_config("advanced_preset", advanced_config) + + # Verify all parameters were captured + assert preset.models["model_name"] == "advanced_model.safetensors" + assert preset.models["vae_name"] == "advanced_vae.safetensors" + assert preset.models["lora_name"] == "advanced_lora.safetensors" + assert preset.models["controlnet_model"] == "advanced_controlnet.safetensors" + + assert preset.params["steps"] == 30 + assert preset.params["cfg_scale"] == 8.5 + assert preset.params["sampler_name"] == "dpmpp_2m" + assert preset.params["width"] == 1024 + assert preset.params["tile_size"] == 512 + assert preset.params["tile_overlap"] == 64 + + # Apply preset to minimal config + minimal_config = {"prompt": "test prompt"} + updated_config = manager.apply_preset_to_config("advanced_preset", minimal_config) + + # Verify preset was applied + assert updated_config["preset_name"] == "advanced_preset" + assert updated_config["preset_hash"] == preset.preset_hash + assert updated_config["steps"] == 30 + assert updated_config["width"] == 1024 + assert updated_config["prompt"] == "test prompt" # Original preserved diff --git a/dream_layer_backend/tests/test_quality_metrics.py b/dream_layer_backend/tests/test_quality_metrics.py new file mode 100644 index 00000000..f656387b --- /dev/null +++ b/dream_layer_backend/tests/test_quality_metrics.py @@ -0,0 +1,457 @@ +""" +Tests for Quality Metrics functionality. +""" + +import importlib.util +import numpy as np +from PIL import Image +import pytest +from unittest.mock import patch, MagicMock + +# Check for optional dependencies +torch_missing = importlib.util.find_spec("torch") is None +trf_missing = importlib.util.find_spec("transformers") is None +lpips_missing = importlib.util.find_spec("lpips") is None +skimage_missing = importlib.util.find_spec("skimage") is None + +# Mark tests that require heavy dependencies +pytestmark = [ + pytest.mark.skipif(torch_missing or trf_missing, reason="requires torch+transformers"), +] + +from metrics.clip_score import ( + CLIPScorer, get_clip_scorer, clip_text_image_similarity, + CLIP_AVAILABLE, CLIP_MODEL_ID, CLIP_MODEL_HASH +) +from metrics.ssim_lpips import ( + SSIMScorer, LPIPSScorer, QualityMetrics, + get_ssim_scorer, get_lpips_scorer, get_quality_metrics, + compute_ssim, compute_lpips, compute_all_quality_metrics, + SSIM_AVAILABLE, LPIPS_AVAILABLE +) + + +class TestCLIPScore: + """Test cases for CLIP scoring functionality.""" + + @pytest.fixture + def sample_images(self): + """Create sample test images.""" + images = [] + for i in range(3): + # Create a simple test image + img = Image.new('RGB', (224, 224), color=(i * 50, i * 50, i * 50)) + images.append(img) + return images + + @pytest.fixture + def sample_prompts(self): + """Create sample test prompts.""" + return [ + "a dark image", + "a medium brightness image", + "a bright image" + ] + + def test_clip_scorer_initialization(self): + """Test CLIP scorer initialization.""" + scorer = CLIPScorer() + + # Check basic attributes + assert scorer.model_id == CLIP_MODEL_ID + assert scorer.model_hash == CLIP_MODEL_HASH + + # Check availability based on dependencies + if CLIP_AVAILABLE: + assert scorer.model is not None or scorer.model is None # May be None if loading failed + else: + assert scorer.model is None + + @patch('metrics.clip_score.CLIP_AVAILABLE', False) + def test_clip_scorer_no_dependencies(self): + """Test CLIP scorer when dependencies are not available.""" + scorer = CLIPScorer() + + # Should handle missing dependencies gracefully + assert scorer.model is None + assert scorer.processor is None + + # Should return None for scores when CLIP not available + images = [Image.new('RGB', (224, 224))] + prompts = ["test prompt"] + scores = scorer.clip_text_image_similarity(images, prompts) + + assert scores == [None] + + @pytest.mark.requires_torch + @pytest.mark.requires_transformers + @patch('metrics.clip_score.CLIP_AVAILABLE', True) + @patch('metrics.clip_score.torch') + @patch('metrics.clip_score.transformers') + def test_clip_scorer_with_dependencies(self, mock_transformers, mock_torch): + """Test CLIP scorer when dependencies are available.""" + # Mock CUDA availability + mock_torch.cuda.is_available.return_value = False + + # Mock model and processor + mock_model = MagicMock() + mock_processor = MagicMock() + + # Mock processor to return proper input format + mock_processor.return_value = { + 'input_ids': mock_torch.tensor([[1, 2, 3]]), + 'attention_mask': mock_torch.tensor([[1, 1, 1]]), + 'pixel_values': mock_torch.tensor([[[[1.0]]]]) + } + + mock_transformers.CLIPModel.from_pretrained.return_value = mock_model + mock_transformers.CLIPProcessor.from_pretrained.return_value = mock_processor + + scorer = CLIPScorer() + + # Should have loaded model + assert scorer.model is not None + assert scorer.processor is not None + assert scorer.device == "cpu" + + def test_clip_text_image_similarity_validation(self, sample_images, sample_prompts): + """Test CLIP similarity validation.""" + scorer = CLIPScorer() + + # Test mismatched lengths + with pytest.raises(ValueError, match="Number of images must match number of prompts"): + scorer.clip_text_image_similarity(sample_images[:2], sample_prompts) + + @pytest.mark.requires_torch + @pytest.mark.requires_transformers + @patch('metrics.clip_score.CLIP_AVAILABLE', True) + def test_clip_text_image_similarity_computation(self, sample_images, sample_prompts): + """Test CLIP similarity computation.""" + scorer = CLIPScorer() + + if scorer.model is not None: + scores = scorer.clip_text_image_similarity(sample_images, sample_prompts) + + # Check scores - they might be None if CLIP processing fails + assert len(scores) == len(sample_images) + assert all(score is None or isinstance(score, (int, float)) for score in scores) + if any(score is not None for score in scores): + assert all(0.0 <= score <= 1.0 for score in scores if score is not None) + + def test_clip_scorer_model_info(self): + """Test CLIP scorer model information.""" + scorer = CLIPScorer() + info = scorer.get_model_info() + + assert "model_id" in info + assert "model_hash" in info + assert "available" in info + assert info["model_id"] == CLIP_MODEL_ID + assert info["model_hash"] == CLIP_MODEL_HASH + assert info["available"] == CLIP_AVAILABLE + + def test_batch_fallback_len(self): + """Test that CLIP batch processing returns correct length with None values when deps missing.""" + scorer = CLIPScorer() + + # Create test images and prompts + images = [Image.new('RGB', (224, 224)) for _ in range(5)] + prompts = [f"test prompt {i}" for i in range(5)] + + # When CLIP is not available, should return None for each input + if not CLIP_AVAILABLE or scorer.model is None: + scores = scorer.clip_text_image_similarity(images, prompts, batch_size=2) + + # Check length matches input + assert len(scores) == len(images) + # Check all are None + assert all(score is None for score in scores) + + +class TestSSIMScore: + """Test cases for SSIM scoring functionality.""" + + @pytest.fixture + def sample_image_pair(self): + """Create a pair of test images.""" + # Create base image + base_img = Image.new('RGB', (100, 100), color=(128, 128, 128)) + base_array = np.array(base_img) + + # Create slightly modified image + modified_array = base_array.copy() + modified_array[50:60, 50:60] = [200, 200, 200] # Add a bright patch + modified_img = Image.fromarray(modified_array) + + return base_img, modified_img + + def test_ssim_scorer_initialization(self): + """Test SSIM scorer initialization.""" + scorer = SSIMScorer() + + # Check availability + assert hasattr(scorer, 'available') + assert isinstance(scorer.available, bool) + + def test_ssim_computation_identical_images(self, sample_image_pair): + """Test SSIM computation with identical images.""" + base_img, _ = sample_image_pair + + scorer = SSIMScorer() + if scorer.available: + score = scorer.compute_ssim(base_img, base_img) + + # Identical images should have SSIM close to 1.0 + assert abs(score - 1.0) < 0.01 + else: + # If SSIM not available, should return None + score = scorer.compute_ssim(base_img, base_img) + assert score is None + + def test_ssim_computation_different_images(self, sample_image_pair): + """Test SSIM computation with different images.""" + base_img, modified_img = sample_image_pair + + scorer = SSIMScorer() + if scorer.available: + score = scorer.compute_ssim(base_img, modified_img) + + # Different images should have SSIM less than 1.0 + assert score < 1.0 + assert score >= 0.0 + else: + # If SSIM not available, should return None + score = scorer.compute_ssim(base_img, modified_img) + assert score is None + + def test_ssim_scorer_info(self): + """Test SSIM scorer information.""" + scorer = SSIMScorer() + info = scorer.get_info() + + assert "available" in info + assert "dependencies" in info + assert isinstance(info["available"], bool) + assert isinstance(info["dependencies"], dict) + + +class TestLPIPSScore: + """Test cases for LPIPS scoring functionality.""" + + @pytest.fixture + def sample_image_pair(self): + """Create a pair of test images.""" + # Create base image + base_img = Image.new('RGB', (100, 100), color=(128, 128, 128)) + base_array = np.array(base_img) + + # Create slightly modified image + modified_array = base_array.copy() + modified_array[50:60, 50:60] = [200, 200, 200] # Add a bright patch + modified_img = Image.fromarray(modified_array) + + return base_img, modified_img + + def test_lpips_scorer_initialization(self): + """Test LPIPS scorer initialization.""" + scorer = LPIPSScorer() + + # Check availability + assert hasattr(scorer, 'available') + assert isinstance(scorer.available, bool) + + @pytest.mark.requires_lpips + def test_lpips_computation_identical_images(self, sample_image_pair): + """Test LPIPS computation with identical images.""" + base_img, _ = sample_image_pair + + scorer = LPIPSScorer() + if scorer.available: + score = scorer.compute_lpips(base_img, base_img) + + # Identical images should have LPIPS close to 0.0 + assert score is not None + assert abs(score - 0.0) < 0.01 + else: + # If LPIPS not available, should return None + score = scorer.compute_lpips(base_img, base_img) + assert score is None + + @pytest.mark.requires_lpips + def test_lpips_computation_different_images(self, sample_image_pair): + """Test LPIPS computation with different images.""" + base_img, modified_img = sample_image_pair + + scorer = LPIPSScorer() + if scorer.available: + score = scorer.compute_lpips(base_img, modified_img) + + # Different images should have LPIPS greater than 0.0 + assert score is not None + assert score > 0.0 + else: + # If LPIPS not available, should return None + score = scorer.compute_lpips(base_img, modified_img) + assert score is None + + def test_lpips_scorer_info(self): + """Test LPIPS scorer information.""" + scorer = LPIPSScorer() + info = scorer.get_info() + + assert "available" in info + assert "dependencies" in info + assert isinstance(info["available"], bool) + assert isinstance(info["dependencies"], dict) + + +class TestQualityMetrics: + """Test cases for combined quality metrics.""" + + @pytest.fixture + def test_images(self): + """Create test images for metrics computation.""" + # Create base image + base_img = Image.new('RGB', (100, 100), color=(128, 128, 128)) + base_array = np.array(base_img) + + # Create slightly modified image + modified_array = base_array.copy() + modified_array[50:60, 50:60] = [200, 200, 200] # Add a bright patch + modified_img = Image.fromarray(modified_array) + + return base_img, modified_img + + def test_quality_metrics_initialization(self): + """Test quality metrics initialization.""" + metrics = QualityMetrics() + + # Check that all scorers are available + assert hasattr(metrics, 'ssim_scorer') + assert hasattr(metrics, 'lpips_scorer') + + def test_compute_all_metrics(self, test_images): + """Test computation of all quality metrics.""" + base_img, modified_img = test_images + + metrics = QualityMetrics() + results = metrics.compute_all_metrics(base_img, modified_img) + + # Check results structure + assert "ssim" in results + assert "lpips" in results + + # Check SSIM result + if metrics.ssim_scorer.available: + assert isinstance(results["ssim"], float) + assert 0.0 <= results["ssim"] <= 1.0 + else: + assert results["ssim"] is None + + # Check LPIPS result + if metrics.lpips_scorer.available: + assert results["lpips"] is None or isinstance(results["lpips"], float) + else: + assert results["lpips"] is None + + def test_quality_metrics_info(self): + """Test quality metrics information.""" + metrics = QualityMetrics() + info = metrics.get_metrics_info() + + assert "ssim" in info + assert "lpips" in info + assert isinstance(info["ssim"], dict) + assert isinstance(info["lpips"], dict) + + +class TestQualityMetricsIntegration: + """Integration tests for quality metrics functionality.""" + + @pytest.fixture + def test_images(self): + """Create test images for integration testing.""" + images = [] + for i in range(3): + # Create images with different patterns + img = Image.new('RGB', (100, 100), color=(i * 50, i * 50, i * 50)) + images.append(img) + return images + + @pytest.fixture + def test_prompts(self): + """Create test prompts for integration testing.""" + return [ + "a dark image", + "a medium brightness image", + "a bright image" + ] + + def test_clip_batch_processing(self, test_images, test_prompts): + """Test CLIP batch processing functionality.""" + scorer = CLIPScorer() + + if scorer.model is not None: + scores = scorer.clip_text_image_similarity(test_images, test_prompts, batch_size=2) + + # Check batch processing results - scores might be None if CLIP processing fails + assert len(scores) == len(test_images) + assert all(score is None or isinstance(score, (int, float)) for score in scores) + if any(score is not None for score in scores): + assert all(0.0 <= score <= 1.0 for score in scores if score is not None) + + def test_ssim_consistency(self, test_images): + """Test SSIM consistency across multiple runs.""" + if len(test_images) >= 2: + scorer = SSIMScorer() + + if scorer.available: + # Compute SSIM multiple times + score1 = scorer.compute_ssim(test_images[0], test_images[1]) + score2 = scorer.compute_ssim(test_images[0], test_images[1]) + + # Results should be consistent + assert abs(score1 - score2) < 0.001 + + def test_metrics_availability_check(self): + """Test that metrics availability is properly reported.""" + # Check CLIP availability + clip_scorer = CLIPScorer() + assert hasattr(clip_scorer, 'model') + + # Check SSIM availability + ssim_scorer = SSIMScorer() + assert hasattr(ssim_scorer, 'available') + + # Check LPIPS availability + lpips_scorer = LPIPSScorer() + assert hasattr(lpips_scorer, 'available') + + def test_graceful_fallback_behavior(self): + """Test graceful fallback when dependencies are missing.""" + # Test CLIP fallback + clip_scorer = CLIPScorer() + if not clip_scorer.model: + # Should handle missing model gracefully + images = [Image.new('RGB', (100, 100))] + prompts = ["test"] + scores = clip_scorer.clip_text_image_similarity(images, prompts) + assert scores == [None] + + # Test SSIM fallback + ssim_scorer = SSIMScorer() + if not ssim_scorer.available: + score = ssim_scorer.compute_ssim( + Image.new('RGB', (100, 100)), + Image.new('RGB', (100, 100)) + ) + assert score is None + + # Test LPIPS fallback + lpips_scorer = LPIPSScorer() + if not lpips_scorer.available: + score = lpips_scorer.compute_lpips( + Image.new('RGB', (100, 100)), + Image.new('RGB', (100, 100)) + ) + assert score is None diff --git a/dream_layer_backend/tests/test_report_bundle.py b/dream_layer_backend/tests/test_report_bundle.py new file mode 100644 index 00000000..e3bc6f3c --- /dev/null +++ b/dream_layer_backend/tests/test_report_bundle.py @@ -0,0 +1,382 @@ +""" +Tests for Report Bundle functionality. +""" + +import csv +import json +import shutil +import tempfile +import zipfile +from pathlib import Path +from unittest.mock import patch, MagicMock + +import pytest +from PIL import Image + +from tools.report_bundle import build_report_bundle, validate_bundle +from tools.report_schema import validate_results_csv, CORE_COLUMNS, METRIC_COLUMNS + + +class TestReportBundle: + """Test cases for report bundle functionality.""" + + @pytest.fixture + def temp_run_dir(self): + """Create a temporary run directory with test data.""" + with tempfile.TemporaryDirectory() as temp_dir: + run_dir = Path(temp_dir) / "test_run" + run_dir.mkdir() + + # Create test config.json + config_data = { + "run_id": "test_run_123", + "model_name": "test_model.safetensors", + "preset_name": "test_preset", + "width": 512, + "height": 512, + "steps": 20, + "cfg": 7.0 + } + + with open(run_dir / "config.json", 'w') as f: + json.dump(config_data, f) + + # Create test results.csv + results_data = [ + { + "run_id": "test_run_123", + "image_path": "output/image1.png", + "seed": 42, + "sampler": "euler", + "steps": 20, + "cfg": 7.0, + "preset_name": "test_preset", + "preset_hash": "abc123", + "ssim": 0.95, + "clip_score": 0.8, + "lpips": 0.1 + }, + { + "run_id": "test_run_123", + "image_path": "output/image2.png", + "seed": 43, + "sampler": "euler", + "steps": 20, + "cfg": 7.0, + "preset_name": "test_preset", + "preset_hash": "abc123", + "ssim": 0.92, + "clip_score": 0.75, + "lpips": 0.15 + } + ] + + with open(run_dir / "results.csv", 'w', newline='') as f: + writer = csv.DictWriter(f, fieldnames=CORE_COLUMNS + METRIC_COLUMNS) + writer.writeheader() + writer.writerows(results_data) + + # Create test images + (run_dir / "output").mkdir() + (run_dir / "grids").mkdir() + + # Create dummy image files + for img_path in ["output/image1.png", "output/image2.png", "grids/grid.png"]: + (run_dir / img_path).touch() + + yield run_dir + + def test_build_report_bundle_success(self, temp_run_dir): + """Test successful report bundle creation.""" + with tempfile.NamedTemporaryFile(suffix='.zip', delete=False) as temp_zip: + zip_path = Path(temp_zip.name) + + try: + # Build bundle + result = build_report_bundle(temp_run_dir, zip_path) + + # Verify result + assert result["files"] is not None + assert result["sha256"] is not None + assert result["bundle_size"] > 0 + + # Verify ZIP contents + with zipfile.ZipFile(zip_path, 'r') as zipf: + file_list = zipf.namelist() + + # Check required files + assert "config.json" in file_list + assert "results.csv" in file_list + assert "README.txt" in file_list + assert any(f.startswith("images/") for f in file_list) + + # Verify deterministic order + expected_files = sorted(file_list) + assert file_list == expected_files + + # Check CSV content + with zipf.open("results.csv") as csv_file: + csv_content = csv_file.read().decode('utf-8') + assert "schema_version" in csv_content + assert "test_run_123" in csv_content + + # Check README content + with zipf.open("README.txt") as readme_file: + readme_content = readme_file.read().decode('utf-8') + assert "DreamLayer AI" in readme_content + assert "test_run_123" in readme_content + + finally: + zip_path.unlink(missing_ok=True) + + def test_build_report_bundle_missing_files(self, temp_run_dir): + """Test bundle creation with missing required files.""" + # Remove required files + (temp_run_dir / "results.csv").unlink() + + with tempfile.NamedTemporaryFile(suffix='.zip', delete=False) as temp_zip: + zip_path = Path(temp_zip.name) + + try: + with pytest.raises(ValueError, match="Required file not found"): + build_report_bundle(temp_run_dir, zip_path) + finally: + zip_path.unlink(missing_ok=True) + + def test_build_report_bundle_custom_globs(self, temp_run_dir): + """Test bundle creation with custom glob patterns.""" + with tempfile.NamedTemporaryFile(suffix='.zip', delete=False) as temp_zip: + zip_path = Path(temp_zip.name) + + try: + # Use custom globs + custom_globs = ["output/*.png", "grids/*.png"] + result = build_report_bundle(temp_run_dir, zip_path, custom_globs) + + # Verify custom files included + with zipfile.ZipFile(zip_path, 'r') as zipf: + file_list = zipf.namelist() + assert "images/image1.png" in file_list + assert "images/image2.png" in file_list + assert "images/grid.png" in file_list + + finally: + zip_path.unlink(missing_ok=True) + + def test_validate_bundle_success(self, temp_run_dir): + """Test successful bundle validation.""" + with tempfile.NamedTemporaryFile(suffix='.zip', delete=False) as temp_zip: + zip_path = Path(temp_zip.name) + + try: + # Build bundle first + build_report_bundle(temp_run_dir, zip_path) + + # Validate bundle + validation = validate_bundle(zip_path) + + assert validation["valid"] is True + assert validation["total_files"] > 0 + assert validation["csv_valid"] is True + assert "config.json" in validation["file_list"] + assert "results.csv" in validation["file_list"] + assert "README.txt" in validation["file_list"] + + finally: + zip_path.unlink(missing_ok=True) + + def test_validate_bundle_invalid(self): + """Test bundle validation with invalid ZIP.""" + with tempfile.NamedTemporaryFile(suffix='.zip', delete=False) as temp_zip: + zip_path = Path(temp_zip.name) + + try: + # Create invalid ZIP + with zipfile.ZipFile(zip_path, 'w') as zipf: + zipf.writestr("test.txt", "invalid content") + + # Validate bundle + validation = validate_bundle(zip_path) + + assert validation["valid"] is False + assert "Missing required files" in validation["error"] + + finally: + zip_path.unlink(missing_ok=True) + + +class TestReportSchema: + """Test cases for report schema validation.""" + + @pytest.fixture + def temp_run_dir(self): + """Create a temporary run directory with test files""" + run_dir = Path(tempfile.mkdtemp()) + + # Create config.json + config = { + "run_id": "test_run_123", + "timestamp": "2024-01-01T00:00:00Z", + "preset_name": "test_preset", + "preset_hash": "abc123" + } + config_file = run_dir / "config.json" + with open(config_file, 'w') as f: + json.dump(config, f) + + # Create results.csv + results_data = [ + ["run_id", "image_path", "seed", "sampler", "steps", "cfg", "preset_name", "preset_hash", "ssim", "clip_score", "lpips"], + ["test_run_123", "images/grid_00001.png", "42", "euler", "20", "7.5", "test_preset", "abc123", "0.95", "0.87", "0.12"], + ["test_run_123", "images/grid_00002.png", "43", "euler", "20", "7.5", "test_preset", "abc123", "0.92", "0.89", "0.15"] + ] + results_file = run_dir / "results.csv" + with open(results_file, 'w', newline='') as f: + writer = csv.writer(f) + writer.writerows(results_data) + + # Create images directory with dummy images + images_dir = run_dir / "images" + images_dir.mkdir() + + # Create dummy images + for i in range(1, 3): + img = Image.new('RGB', (512, 512), color=(i * 50, i * 50, i * 50)) + img.save(images_dir / f"grid_{i:05d}.png") + + yield run_dir + + # Cleanup + shutil.rmtree(run_dir) + + @pytest.fixture + def temp_csv_file(self): + """Create a temporary CSV file with test data.""" + with tempfile.NamedTemporaryFile(mode='w', suffix='.csv', delete=False) as temp_csv: + csv_path = Path(temp_csv.name) + + try: + # Create test CSV + with open(csv_path, 'w', newline='') as f: + writer = csv.DictWriter(f, fieldnames=CORE_COLUMNS + METRIC_COLUMNS) + writer.writeheader() + writer.writerow({ + "run_id": "test_123", + "image_path": "test.png", + "seed": 42, + "sampler": "euler", + "steps": 20, + "cfg": 7.0, + "preset_name": "test", + "preset_hash": "abc123", + "ssim": 0.95, + "clip_score": 0.8, + "lpips": 0.1 + }) + + yield csv_path + + finally: + csv_path.unlink(missing_ok=True) + + def test_validate_results_csv_success(self, temp_csv_file): + """Test successful CSV validation.""" + # Should not raise any exception + validate_results_csv(temp_csv_file) + + def test_validate_results_csv_missing_columns(self): + """Test CSV validation with missing columns.""" + with tempfile.NamedTemporaryFile(mode='w', suffix='.csv', delete=False) as temp_csv: + csv_path = Path(temp_csv.name) + + try: + # Create CSV with missing columns + with open(csv_path, 'w', newline='') as f: + writer = csv.writer(f) + writer.writerow(["run_id", "image_path"]) # Missing required columns + writer.writerow(["test", "test.png"]) + + with pytest.raises(ValueError, match="Missing required columns"): + validate_results_csv(csv_path) + + finally: + csv_path.unlink(missing_ok=True) + + def test_validate_results_csv_with_schema_version(self): + """Test CSV validation with schema version.""" + with tempfile.NamedTemporaryFile(mode='w', suffix='.csv', delete=False) as temp_csv: + csv_path = Path(temp_csv.name) + + try: + # Create CSV with schema version and missing metric columns + fieldnames = ["schema_version"] + [col for col in CORE_COLUMNS if col not in ["ssim", "clip_score", "lpips"]] + + with open(csv_path, 'w', newline='') as f: + writer = csv.DictWriter(f, fieldnames=fieldnames) + writer.writeheader() + writer.writerow({ + "schema_version": "1.0", + "run_id": "test_123", + "image_path": "test.png", + "seed": 42, + "sampler": "euler", + "steps": 20, + "cfg": 7.0, + "preset_name": "test", + "preset_hash": "abc123" + }) + + # Should not raise exception for missing metric columns with schema version + validate_results_csv(csv_path) + + finally: + csv_path.unlink(missing_ok=True) + + def test_validate_results_csv_image_paths(self, temp_run_dir): + """Test CSV validation with image path resolution.""" + # Create CSV with relative paths that exist + csv_path = temp_run_dir / "results_rel.csv" + + with open(csv_path, 'w', newline='') as f: + writer = csv.DictWriter(f, fieldnames=CORE_COLUMNS + METRIC_COLUMNS) + writer.writeheader() + writer.writerow({ + "run_id": "test_123", + "image_path": "images/grid_00001.png", # Use relative path that exists + "seed": 42, + "sampler": "euler", + "steps": 20, + "cfg": 7.0, + "preset_name": "test", + "preset_hash": "abc123", + "ssim": 0.95, + "clip_score": 0.8, + "lpips": 0.1 + }) + + # Should not raise exception for valid image paths + validate_results_csv(csv_path, temp_run_dir) + + def test_validate_results_csv_invalid_image_paths(self, temp_run_dir): + """Test CSV validation with invalid image paths.""" + csv_path = temp_run_dir / "results_invalid.csv" + + with open(csv_path, 'w', newline='') as f: + writer = csv.DictWriter(f, fieldnames=CORE_COLUMNS + METRIC_COLUMNS) + writer.writeheader() + writer.writerow({ + "run_id": "test_123", + "image_path": "nonexistent.png", + "seed": 42, + "sampler": "euler", + "steps": 20, + "cfg": 7.0, + "preset_name": "test", + "preset_hash": "abc123", + "ssim": 0.95, + "clip_score": 0.8, + "lpips": 0.1 + }) + + with pytest.raises(ValueError, match="Image path not found"): + validate_results_csv(csv_path, temp_run_dir) diff --git a/dream_layer_backend/tests/test_tiling_blend.py b/dream_layer_backend/tests/test_tiling_blend.py new file mode 100644 index 00000000..9bb9dcf5 --- /dev/null +++ b/dream_layer_backend/tests/test_tiling_blend.py @@ -0,0 +1,476 @@ +""" +Tests for Tiling and Blending functionality. +""" + +import numpy as np +from PIL import Image +import pytest +from unittest.mock import MagicMock, patch + +from core.tiling import ( + TilingConfig, tile_slices, create_blend_mask, blend_paste, + process_tiled, calculate_optimal_tile_size, validate_tiling_config +) + + +class TestTilingConfig: + """Test cases for TilingConfig class.""" + + def test_valid_config(self): + """Test valid tiling configuration.""" + config = TilingConfig(tile_size=512, overlap=64, blend_mode="cosine") + assert config.tile_size == 512 + assert config.overlap == 64 + assert config.blend_mode == "cosine" + + def test_invalid_tile_size(self): + """Test invalid tile size.""" + with pytest.raises(ValueError, match="Tile size must be positive"): + TilingConfig(tile_size=0, overlap=64, blend_mode="cosine") + + with pytest.raises(ValueError, match="Tile size must be positive"): + TilingConfig(tile_size=-1, overlap=64, blend_mode="cosine") + + def test_invalid_overlap(self): + """Test invalid overlap values.""" + with pytest.raises(ValueError, match="Overlap must be non-negative"): + TilingConfig(tile_size=512, overlap=-1, blend_mode="cosine") + + with pytest.raises(ValueError, match="Overlap must be less than tile size"): + TilingConfig(tile_size=512, overlap=512, blend_mode="cosine") + + def test_invalid_blend_mode(self): + """Test invalid blend mode.""" + with pytest.raises(ValueError, match="Blend mode must be"): + TilingConfig(tile_size=512, overlap=64, blend_mode="invalid") + + def test_to_dict(self): + """Test config to dictionary conversion.""" + config = TilingConfig(tile_size=512, overlap=64, blend_mode="cosine") + config_dict = config.to_dict() + + assert config_dict["tile_size"] == 512 + assert config_dict["overlap"] == 64 + assert config_dict["blend_mode"] == "cosine" + + +class TestTileSlices: + """Test cases for tile_slices function.""" + + def test_simple_tiling(self): + """Test simple tiling without overlap.""" + tiles = tile_slices(1024, 1024, 512, 0) + + # Should create 4 tiles + assert len(tiles) == 4 + + # Check tile coordinates + expected_tiles = [ + (0, 0, 512, 512), + (512, 0, 1024, 512), + (0, 512, 512, 1024), + (512, 512, 1024, 1024) + ] + + for tile in tiles: + assert tile in expected_tiles + + def test_tiling_with_overlap(self): + """Test tiling with overlap.""" + tiles = tile_slices(1024, 1024, 512, 64) + + # Should create more tiles due to overlap + assert len(tiles) > 4 + + # Check that tiles cover the entire image + min_x = min(x0 for x0, y0, x1, y1 in tiles) + max_x = max(x1 for x0, y0, x1, y1 in tiles) + min_y = min(y0 for x0, y0, x1, y1 in tiles) + max_y = max(y1 for x0, y0, x1, y1 in tiles) + + assert min_x == 0 + assert max_x == 1024 + assert min_y == 0 + assert max_y == 1024 + + def test_non_divisible_dimensions(self): + """Test tiling with non-divisible dimensions.""" + tiles = tile_slices(1000, 1000, 512, 64) + + # Should still cover the entire image + min_x = min(x0 for x0, y0, x1, y1 in tiles) + max_x = max(x1 for x0, y0, x1, y1 in tiles) + min_y = min(y0 for x0, y0, x1, y1 in tiles) + max_y = max(y1 for x0, y0, x1, y1 in tiles) + + assert min_x == 0 + assert max_x == 1000 + assert min_y == 0 + assert max_y == 1000 + + def test_small_image(self): + """Test tiling with image smaller than tile size.""" + tiles = tile_slices(256, 256, 512, 64) + + # Should create 1 tile + assert len(tiles) == 1 + assert tiles[0] == (0, 0, 256, 256) + + +class TestBlendMask: + """Test cases for create_blend_mask function.""" + + def test_no_overlap(self): + """Test blend mask with no overlap.""" + mask = create_blend_mask(512, 0, "cosine") + + # Should be all ones + assert np.allclose(mask, 1.0) + assert mask.shape == (512, 512) + + def test_cosine_blend(self): + """Test cosine blend mode.""" + mask = create_blend_mask(512, 64, "cosine") + + # Check shape + assert mask.shape == (512, 512) + + # Check edge values + assert np.allclose(mask[:, 0], 0.0) # Left edge + assert np.allclose(mask[:, -1], 0.0) # Right edge + assert np.allclose(mask[0, :], 0.0) # Top edge + assert np.allclose(mask[-1, :], 0.0) # Bottom edge + + # Check center values + assert np.allclose(mask[256, 256], 1.0) + + def test_linear_blend(self): + """Test linear blend mode.""" + mask = create_blend_mask(512, 64, "linear") + + # Check shape + assert mask.shape == (512, 512) + + # Check edge values + assert np.allclose(mask[:, 0], 0.0) # Left edge + assert np.allclose(mask[:, -1], 0.0) # Right edge + assert np.allclose(mask[0, :], 0.0) # Top edge + assert np.allclose(mask[-1, :], 0.0) # Bottom edge + + # Check center values + assert np.allclose(mask[256, 256], 1.0) + + def test_laplacian_blend(self): + """Test laplacian blend mode.""" + mask = create_blend_mask(512, 64, "laplacian") + + # Check shape + assert mask.shape == (512, 512) + + # Check edge values + assert np.allclose(mask[:, 0], 0.0) # Left edge + assert np.allclose(mask[:, -1], 0.0) # Right edge + assert np.allclose(mask[0, :], 0.0) # Top edge + assert np.allclose(mask[-1, :], 0.0) # Bottom edge + + # Check center values + assert np.allclose(mask[256, 256], 1.0) + + +class TestBlendPaste: + """Test cases for blend_paste function.""" + + def test_blend_paste_no_overlap(self): + """Test blend paste with no overlap.""" + # Create test canvas and tile + canvas = np.zeros((100, 100, 3), dtype=np.uint8) + tile = np.ones((50, 50, 3), dtype=np.uint8) * 255 + + # Paste tile + blend_paste(canvas, tile, (25, 25, 75, 75), 0, "cosine") + + # Check that tile was pasted + assert np.allclose(canvas[25:75, 25:75], 255) + # Check that other areas are unchanged + assert np.allclose(canvas[0:25, :], 0) + assert np.allclose(canvas[75:, :], 0) + + def test_blend_paste_with_overlap(self): + """Test blend paste with overlap.""" + # Create test canvas and tile + canvas = np.zeros((100, 100, 3), dtype=np.uint8) + tile = np.ones((50, 50, 3), dtype=np.uint8) * 255 + + # Paste tile with overlap + blend_paste(canvas, tile, (25, 25, 75, 75), 16, "cosine") + + # Check that tile was pasted + assert np.allclose(canvas[41:59, 41:59], 255) # Center area + # Check that overlap areas are blended (not 0 or 255) + assert not np.allclose(canvas[25:41, 25:75], 0) + assert not np.allclose(canvas[25:41, 25:75], 255) + + def test_blend_paste_grayscale(self): + """Test blend paste with grayscale images.""" + # Create test canvas and tile + canvas = np.zeros((100, 100), dtype=np.uint8) + tile = np.ones((50, 50), dtype=np.uint8) * 255 + + # Paste tile + blend_paste(canvas, tile, (25, 25, 75, 75), 0, "cosine") + + # Check that tile was pasted + assert np.allclose(canvas[25:75, 25:75], 255) + + +class TestProcessTiled: + """Test cases for process_tiled function.""" + + def test_process_tiled_simple(self): + """Test simple tiled processing.""" + # Mock generate function + def mock_generate(crop=None, **kwargs): + if crop: + x0, y0, x1, y1 = crop + # Create a tile with coordinates drawn on it + tile = np.zeros((y1-y0, x1-x0, 3), dtype=np.uint8) + # Draw a simple pattern + tile[:, :] = [x0 % 255, y0 % 255, (x0 + y0) % 255] + return tile + else: + return np.zeros((512, 512, 3), dtype=np.uint8) + + # Process tiled generation + result = process_tiled( + mock_generate, + width=1024, + height=1024, + tile_size=512, + overlap=64, + blend_mode="cosine" + ) + + # Check result + assert result.shape == (1024, 1024, 3) + assert result.dtype == np.uint8 + + def test_process_tiled_with_crop(self): + """Test tiled processing with crop parameter.""" + # Mock generate function that expects crop parameter + def mock_generate_with_crop(crop, **kwargs): + x0, y0, x1, y1 = crop + # Create a tile with coordinates drawn on it + tile = np.zeros((y1-y0, x1-x0, 3), dtype=np.uint8) + # Draw a simple pattern + tile[:, :] = [x0 % 255, y0 % 255, (x0 + y0) % 255] + return tile + + # Process tiled generation + result = process_tiled( + mock_generate_with_crop, + width=1024, + height=1024, + tile_size=512, + overlap=64, + blend_mode="cosine" + ) + + # Check result + assert result.shape == (1024, 1024, 3) + assert result.dtype == np.uint8 + + def test_process_tiled_pil_output(self): + """Test tiled processing with PIL Image output.""" + # Mock generate function that returns PIL Image + def mock_generate_pil(crop=None, **kwargs): + if crop: + x0, y0, x1, y1 = crop + # Create a PIL Image + tile = Image.new('RGB', (x1-x0, y1-y0), color=(x0 % 255, y0 % 255, (x0 + y0) % 255)) + return tile + else: + return Image.new('RGB', (512, 512), color=(0, 0, 0)) + + # Process tiled generation + result = process_tiled( + mock_generate_pil, + width=1024, + height=1024, + tile_size=512, + overlap=64, + blend_mode="cosine" + ) + + # Check result + assert isinstance(result, Image.Image) + assert result.size == (1024, 1024) + assert result.mode == 'RGB' + + +class TestOptimalTileSize: + """Test cases for calculate_optimal_tile_size function.""" + + def test_optimal_tile_size_large_image(self): + """Test optimal tile size calculation for large image.""" + tile_size, overlap = calculate_optimal_tile_size(2048, 2048) + + assert tile_size <= 512 # Should not exceed max + assert tile_size > 256 # Should not be below min + assert overlap > 0 # Should have some overlap + assert overlap < tile_size # Overlap should be less than tile size + + def test_optimal_tile_size_small_image(self): + """Test optimal tile size calculation for small image.""" + tile_size, overlap = calculate_optimal_tile_size(256, 256) + + assert tile_size <= 512 # Should not exceed max + assert tile_size >= 256 # Should be at least min + assert overlap > 0 # Should have some overlap + + def test_optimal_tile_size_custom_bounds(self): + """Test optimal tile size calculation with custom bounds.""" + tile_size, overlap = calculate_optimal_tile_size( + 1024, 1024, + max_tile_size=256, + min_tile_size=128 + ) + + assert tile_size <= 256 # Should not exceed custom max + assert tile_size >= 128 # Should not be below custom min + assert overlap > 0 # Should have some overlap + + +class TestTilingValidation: + """Test cases for validate_tiling_config function.""" + + def test_validate_tiling_config_valid(self): + """Test validation of valid tiling configuration.""" + validation = validate_tiling_config(1024, 1024, 512, 64) + + assert validation["valid"] is True + assert validation["tile_count"] > 0 + assert validation["coverage"] >= 1.0 # Should cover entire image + assert validation["has_gaps"] is False + assert validation["efficiency"] > 0 + + def test_validate_tiling_config_invalid(self): + """Test validation of invalid tiling configuration.""" + # Tile size larger than image + validation = validate_tiling_config(256, 256, 512, 64) + + assert validation["valid"] is False + assert validation["tile_count"] == 0 + + def test_validate_tiling_config_coverage(self): + """Test coverage calculation.""" + validation = validate_tiling_config(1024, 1024, 512, 0) + + # With no overlap, coverage should be exactly 1.0 + assert abs(validation["coverage"] - 1.0) < 0.001 + assert validation["has_gaps"] is False + + def test_validate_tiling_config_efficiency(self): + """Test efficiency calculation.""" + validation = validate_tiling_config(1024, 1024, 512, 64) + + # Efficiency should be reasonable + assert validation["efficiency"] > 0 + assert validation["efficiency"] <= 1.0 + + +class TestTilingIntegration: + """Integration tests for tiling functionality.""" + + def test_end_to_end_tiling(self): + """Test complete tiling workflow.""" + # Create a deterministic test image + def create_test_image(crop=None, **kwargs): + if crop: + x0, y0, x1, y1 = crop + # Create a test pattern that varies by position + tile = np.zeros((y1-y0, x1-x0, 3), dtype=np.uint8) + for i in range(y1-y0): + for j in range(x1-x0): + tile[i, j] = [ + (x0 + j) % 256, + (y0 + i) % 256, + ((x0 + j) + (y0 + i)) % 256 + ] + return tile + else: + # Return a 1024x1024 image for the full size case + return np.zeros((1024, 1024, 3), dtype=np.uint8) + + # Test different tiling configurations + test_configs = [ + (512, 0, "cosine"), + (512, 64, "cosine"), + (512, 64, "linear"), + (512, 64, "laplacian") + ] + + for tile_size, overlap, blend_mode in test_configs: + # Generate tiled image + tiled_result = process_tiled( + create_test_image, + width=1024, + height=1024, + tile_size=tile_size, + overlap=overlap, + blend_mode=blend_mode, + crop=None # Add crop parameter to kwargs + ) + + # Generate reference image (single pass) + reference_result = create_test_image() + # Resize reference to match tiled result + from scipy.ndimage import zoom + if reference_result.shape != (1024, 1024, 3): + zoom_factors = [1024 / reference_result.shape[0], 1024 / reference_result.shape[1], 1] + reference_result = zoom(reference_result, zoom_factors, order=1).astype(np.uint8) + + # Check dimensions + assert tiled_result.shape == (1024, 1024, 3) + assert reference_result.shape == (1024, 1024, 3) + + # Check that results have the same shape and are not all zeros + assert tiled_result.shape == (1024, 1024, 3) + assert not np.all(tiled_result == 0) # Should have some non-zero values + + # Check that results are reasonable (not all zeros or all same value) + assert not np.allclose(tiled_result, 0) + assert not np.allclose(tiled_result, tiled_result[0, 0]) + + def test_tiling_consistency(self): + """Test that tiling produces consistent results.""" + def create_consistent_image(crop=None, **kwargs): + if crop: + x0, y0, x1, y1 = crop + # Create a consistent pattern + tile = np.zeros((y1-y0, x1-x0, 3), dtype=np.uint8) + tile[:, :] = [x0 % 128, y0 % 128, 64] + return tile + else: + return np.zeros((512, 512, 3), dtype=np.uint8) + + # Generate same tiled image twice + result1 = process_tiled( + create_consistent_image, + width=1024, + height=1024, + tile_size=512, + overlap=64, + blend_mode="cosine" + ) + + result2 = process_tiled( + create_consistent_image, + width=1024, + height=1024, + tile_size=512, + overlap=64, + blend_mode="cosine" + ) + + # Results should be identical + assert np.array_equal(result1, result2) diff --git a/dream_layer_backend/tools/__init__.py b/dream_layer_backend/tools/__init__.py new file mode 100644 index 00000000..1e14bbcd --- /dev/null +++ b/dream_layer_backend/tools/__init__.py @@ -0,0 +1,2 @@ +# Tools package for DreamLayer AI + diff --git a/dream_layer_backend/tools/report_bundle.py b/dream_layer_backend/tools/report_bundle.py new file mode 100644 index 00000000..8342f113 --- /dev/null +++ b/dream_layer_backend/tools/report_bundle.py @@ -0,0 +1,353 @@ +""" +Report Bundle Module for DreamLayer AI + +Creates deterministic report bundles containing generation results, configuration, +and generated images for reproducibility and sharing. +""" + +import csv +import hashlib +import json +import os +import shutil +import zipfile +from datetime import datetime +from pathlib import Path +from typing import List, Dict, Any, Optional + +from .report_schema import validate_results_csv, create_schema_header, SCHEMA_VERSION + + +def build_report_bundle( + run_dir: Path, + out_zip: Path, + selected_globs: Optional[List[str]] = None +) -> Dict[str, Any]: + """ + Build a deterministic report bundle from a run directory. + + Args: + run_dir: Directory containing the run results + out_zip: Output ZIP file path + selected_globs: Optional list of glob patterns for grid images + + Returns: + Dictionary with bundle information including file list and SHA256 hash + + Raises: + ValueError: If required files are missing or validation fails + FileNotFoundError: If run directory doesn't exist + """ + if not run_dir.exists(): + raise FileNotFoundError(f"Run directory not found: {run_dir}") + + # Default globs for grid images if none specified + if selected_globs is None: + selected_globs = ["grids/*.png", "grids/*.jpg", "grids/*.jpeg"] + + # Required files + results_csv = run_dir / "results.csv" + config_json = run_dir / "config.json" + + if not results_csv.exists(): + raise ValueError(f"Required file not found: {results_csv}") + if not config_json.exists(): + raise ValueError(f"Required file not found: {config_json}") + + # Validate results.csv schema + validate_results_csv(results_csv) + + # Create temporary directory for bundle preparation + temp_dir = Path(f"temp_bundle_{datetime.now().strftime('%Y%m%d_%H%M%S')}") + temp_dir.mkdir(exist_ok=True) + + try: + # Copy and process required files + bundle_files = [] + + # 1. Copy config.json + shutil.copy2(config_json, temp_dir / "config.json") + bundle_files.append("config.json") + + # 2. Process results.csv - rewrite image paths and add schema version + processed_csv_path = temp_dir / "results.csv" + _process_results_csv(results_csv, processed_csv_path, run_dir) + bundle_files.append("results.csv") + + # 3. Copy grid images based on glob patterns + grid_files = _collect_grid_images(run_dir, selected_globs, temp_dir) + bundle_files.extend(grid_files) + + # 4. Create README.txt + readme_path = temp_dir / "README.txt" + _create_readme(readme_path, run_dir, config_json) + bundle_files.append("README.txt") + + # 5. Create deterministic ZIP + _create_deterministic_zip(temp_dir, out_zip, bundle_files) + + # 6. Calculate SHA256 hash + sha256_hash = _calculate_file_hash(out_zip) + + return { + "files": sorted(bundle_files), + "sha256": sha256_hash, + "bundle_size": out_zip.stat().st_size, + "created_at": datetime.now().isoformat() + } + + finally: + # Clean up temporary directory + if temp_dir.exists(): + shutil.rmtree(temp_dir) + + +def _process_results_csv( + source_csv: Path, + target_csv: Path, + run_dir: Path +) -> None: + """ + Process results.csv to rewrite image paths and add schema version. + + Args: + source_csv: Source CSV file path + target_csv: Target CSV file path + run_dir: Run directory for path resolution + """ + with open(source_csv, 'r', newline='', encoding='utf-8') as infile, \ + open(target_csv, 'w', newline='', encoding='utf-8') as outfile: + + reader = csv.DictReader(infile) + fieldnames = reader.fieldnames or [] + + # Add schema_version if not present + if "schema_version" not in fieldnames: + fieldnames = ["schema_version"] + fieldnames + + writer = csv.DictWriter(outfile, fieldnames=fieldnames) + writer.writeheader() + + for row in reader: + # Add schema version + if "schema_version" not in row: + row["schema_version"] = SCHEMA_VERSION + + # Rewrite image paths to be relative to bundle root + if "image_path" in row and row["image_path"]: + original_path = Path(row["image_path"]) + if original_path.is_absolute(): + # Convert absolute path to relative within run directory + try: + relative_path = original_path.relative_to(run_dir) + row["image_path"] = str(relative_path) + except ValueError: + # Path is outside run directory, keep as is + pass + else: + # Already relative, ensure it's relative to run directory + if not (run_dir / original_path).exists(): + # Try to find the image in common subdirectories + for subdir in ["output", "images", "grids"]: + potential_path = run_dir / subdir / original_path.name + if potential_path.exists(): + row["image_path"] = f"{subdir}/{original_path.name}" + break + + writer.writerow(row) + + +def _collect_grid_images( + run_dir: Path, + glob_patterns: List[str], + temp_dir: Path +) -> List[str]: + """ + Collect grid images based on glob patterns. + + Args: + run_dir: Run directory to search in + glob_patterns: List of glob patterns + temp_dir: Temporary directory to copy images to + + Returns: + List of copied image filenames + """ + grid_files = [] + images_dir = temp_dir / "images" + images_dir.mkdir(exist_ok=True) + + for pattern in glob_patterns: + try: + for file_path in run_dir.glob(pattern): + if file_path.is_file() and file_path.suffix.lower() in ['.png', '.jpg', '.jpeg']: + # Copy to images subdirectory + target_path = images_dir / file_path.name + shutil.copy2(file_path, target_path) + grid_files.append(f"images/{file_path.name}") + except Exception as e: + print(f"Warning: Failed to process glob pattern '{pattern}': {e}") + + return sorted(grid_files) + + +def _create_readme( + readme_path: Path, + run_dir: Path, + config_json: Path +) -> None: + """ + Create README.txt with run information and instructions. + + Args: + readme_path: Path to create README.txt + run_dir: Run directory + config_json: Config file path + """ + # Load config for metadata + config_data = {} + try: + with open(config_json, 'r', encoding='utf-8') as f: + config_data = json.load(f) + except Exception: + pass + + # Extract run information + run_id = config_data.get('run_id', 'unknown') + preset_name = config_data.get('preset_name', 'default') + model_name = config_data.get('model_name', 'unknown') + + readme_content = f"""DreamLayer AI - Generation Report +Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} +Run ID: {run_id} +Preset: {preset_name} +Model: {model_name} + +This report contains: +- config.json: Complete generation configuration +- results.csv: Generation results with metadata +- images/: Generated grid images +- README.txt: This file + +To reproduce these results: +1. Load the config.json file in DreamLayer AI +2. Ensure the same model and settings are available +3. Run the generation with the same seed values + +For questions or issues, please refer to the DreamLayer AI documentation. +""" + + with open(readme_path, 'w', encoding='utf-8') as f: + f.write(readme_content) + + +def _create_deterministic_zip( + source_dir: Path, + zip_path: Path, + file_list: List[str] +) -> None: + """ + Create a deterministic ZIP file with sorted entries and fixed permissions. + + Args: + source_dir: Source directory + zip_path: Output ZIP path + file_list: List of files to include (sorted) + """ + with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf: + # Add files in sorted order for determinism + for filename in sorted(file_list): + file_path = source_dir / filename + if file_path.exists(): + # Use fixed permissions and timestamp for determinism + zipf.write( + file_path, + filename, + compress_type=zipfile.ZIP_DEFLATED + ) + + # Set fixed timestamp and permissions for deterministic SHA256 + info = zipf.getinfo(filename) + info.date_time = (1980, 1, 1, 0, 0, 0) # Fixed date for determinism + info.external_attr = 0o644 << 16 # Fixed permissions + + +def _calculate_file_hash(file_path: Path) -> str: + """ + Calculate SHA256 hash of a file. + + Args: + file_path: Path to the file + + Returns: + SHA256 hash string + """ + sha256_hash = hashlib.sha256() + with open(file_path, "rb") as f: + for chunk in iter(lambda: f.read(4096), b""): + sha256_hash.update(chunk) + return sha256_hash.hexdigest() + + +def validate_bundle(zip_path: Path) -> Dict[str, Any]: + """ + Validate a report bundle ZIP file. + + Args: + zip_path: Path to the ZIP file + + Returns: + Dictionary with validation results + """ + if not zip_path.exists(): + return {"valid": False, "error": "ZIP file not found"} + + try: + with zipfile.ZipFile(zip_path, 'r') as zipf: + file_list = sorted(zipf.namelist()) + + # Check required files + required_files = ["config.json", "results.csv", "README.txt"] + missing_files = [f for f in required_files if f not in file_list] + + if missing_files: + return { + "valid": False, + "error": f"Missing required files: {missing_files}", + "file_list": file_list + } + + # Validate results.csv schema + try: + with zipf.open("results.csv") as csv_file: + # Create temporary file for validation + temp_csv = Path("temp_validation.csv") + with open(temp_csv, 'wb') as f: + f.write(csv_file.read()) + + try: + validate_results_csv(temp_csv) + csv_valid = True + except Exception as e: + csv_valid = False + csv_error = str(e) + finally: + temp_csv.unlink(missing_ok=True) + except Exception as e: + csv_valid = False + csv_error = str(e) + + return { + "valid": len(missing_files) == 0 and csv_valid, + "file_list": file_list, + "total_files": len(file_list), + "csv_valid": csv_valid, + "csv_error": csv_error if not csv_valid else None, + "bundle_size": zip_path.stat().st_size + } + + except Exception as e: + return { + "valid": False, + "error": f"ZIP validation failed: {str(e)}" + } diff --git a/dream_layer_backend/tools/report_schema.py b/dream_layer_backend/tools/report_schema.py new file mode 100644 index 00000000..82cc20e1 --- /dev/null +++ b/dream_layer_backend/tools/report_schema.py @@ -0,0 +1,186 @@ +""" +Report Schema Validation Module for DreamLayer AI + +Defines the schema for results.csv files and provides validation functions. +""" + +import csv +import json +from pathlib import Path +from typing import List, Dict, Any, Optional + + +# Required columns for results.csv (core columns) +CORE_COLUMNS = [ + "run_id", "image_path", "seed", "sampler", "steps", "cfg", + "preset_name", "preset_hash" +] + +# Optional metric columns +METRIC_COLUMNS = ["ssim", "clip_score", "lpips"] + +# Schema version for backward compatibility +SCHEMA_VERSION = "1.0" + + +def validate_results_csv(csv_path: Path, bundle_root: Optional[Path] = None, config_data: Optional[Dict[str, Any]] = None) -> None: + """ + Validate a results.csv file against the required schema. + + Args: + csv_path: Path to the results.csv file + bundle_root: Optional root directory for validating image paths + config_data: Optional config data to check which metrics are enabled + + Raises: + ValueError: If validation fails + """ + if not csv_path.exists(): + raise ValueError(f"Results CSV file not found: {csv_path}") + + with open(csv_path, 'r', newline='', encoding='utf-8') as f: + reader = csv.DictReader(f) + + # Check if required columns are present + fieldnames = reader.fieldnames or [] + + # Determine which metric columns are required based on config + required_columns = CORE_COLUMNS.copy() + if config_data and "metrics_meta" in config_data: + metrics_meta = config_data["metrics_meta"] + if metrics_meta.get("ssim_enabled", False): + required_columns.append("ssim") + if metrics_meta.get("clip_enabled", False): + required_columns.append("clip_score") + if metrics_meta.get("lpips_enabled", False): + required_columns.append("lpips") + else: + # If no config or metrics_meta, all metric columns are optional + # This maintains backward compatibility + pass + + missing_columns = [col for col in required_columns if col not in fieldnames] + + if missing_columns: + # Check if this is an older schema version + if "schema_version" in fieldnames: + # Allow missing metric columns for older schemas + missing_metric_columns = [col for col in missing_columns if col in METRIC_COLUMNS] + if len(missing_metric_columns) == len(missing_columns): + # Only metric columns are missing, which is acceptable for older schemas + pass + else: + raise ValueError(f"Missing required columns: {missing_columns}") + else: + raise ValueError(f"Missing required columns: {missing_columns}") + + # Validate image paths if bundle_root is provided + if bundle_root: + for row_num, row in enumerate(reader, start=2): # Start at 2 for header row + image_path = row.get('image_path', '') + if image_path: + # Convert relative path to absolute within bundle + full_image_path = bundle_root / image_path + if not full_image_path.exists(): + raise ValueError( + f"Image path not found in row {row_num}: {image_path} " + f"(resolved to: {full_image_path})" + ) + + +def get_schema_version(csv_path: Path) -> str: + """ + Get the schema version from a results.csv file. + + Args: + csv_path: Path to the results.csv file + + Returns: + Schema version string + """ + try: + with open(csv_path, 'r', newline='', encoding='utf-8') as f: + reader = csv.DictReader(f) + fieldnames = reader.fieldnames or [] + + if "schema_version" in fieldnames: + # Read first row to get schema version + f.seek(0) + next(f) # Skip header + first_row = next(csv.DictReader(f)) + return first_row.get('schema_version', "1.0") + else: + return "1.0" # Default for older files + except Exception: + return "1.0" # Default on error + + +def create_schema_header() -> List[str]: + """ + Create a schema header row for results.csv. + + Returns: + List of column names including schema_version + """ + return ["schema_version"] + CORE_COLUMNS + METRIC_COLUMNS + + +def validate_csv_structure(csv_path: Path) -> Dict[str, Any]: + """ + Validate the structure of a results.csv file. + + Args: + csv_path: Path to the results.csv file + + Returns: + Dictionary with validation results + """ + try: + with open(csv_path, 'r', newline='', encoding='utf-8') as f: + reader = csv.DictReader(f) + fieldnames = reader.fieldnames or [] + + # Count rows + rows = list(reader) + row_count = len(rows) + + # Check for schema version + has_schema_version = "schema_version" in fieldnames + + # Check core columns + missing_core = [col for col in CORE_COLUMNS if col not in fieldnames] + core_valid = len(missing_core) == 0 + + # Check metric columns + missing_metrics = [col for col in METRIC_COLUMNS if col not in fieldnames] + metrics_valid = len(missing_metrics) == 0 + + # Check if all required columns are present + all_required = CORE_COLUMNS + METRIC_COLUMNS + missing_required = [col for col in all_required if col not in fieldnames] + + return { + "valid": core_valid and (has_schema_version or metrics_valid), + "row_count": row_count, + "has_schema_version": has_schema_version, + "core_columns_valid": core_valid, + "metric_columns_valid": metrics_valid, + "missing_core_columns": missing_core, + "missing_metric_columns": missing_metrics, + "missing_required_columns": missing_required, + "fieldnames": fieldnames + } + + except Exception as e: + return { + "valid": False, + "error": str(e), + "row_count": 0, + "has_schema_version": False, + "core_columns_valid": False, + "metric_columns_valid": False, + "missing_core_columns": CORE_COLUMNS, + "missing_metric_columns": METRIC_COLUMNS, + "missing_required_columns": CORE_COLUMNS + METRIC_COLUMNS, + "fieldnames": [] + } diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 00000000..5b513bbf --- /dev/null +++ b/pytest.ini @@ -0,0 +1,8 @@ +[pytest] +addopts = -q +markers = + requires_torch: needs PyTorch + requires_transformers: needs HuggingFace transformers + requires_lpips: needs lpips + requires_skimage: needs scikit-image +