import torch import numpy as np import cv2 from PIL import Image import logging import gc import time from typing import Optional, Dict, Any, Tuple, List from pathlib import Path import warnings warnings.filterwarnings("ignore") from diffusers import StableDiffusionXLPipeline, DPMSolverMultistepScheduler import open_clip from mask_generator import MaskGenerator from image_blender import ImageBlender from quality_checker import QualityChecker logger = logging.getLogger(__name__) logger.setLevel(logging.INFO) class SceneWeaverCore: """ SceneWeaver with perfect background generation + fixed blending + memory optimization """ # Style presets for diversity generation mode STYLE_PRESETS = { "professional": { "name": "Professional Business", "modifier": "professional office environment, clean background, corporate setting, bright even lighting", "negative_extra": "casual, messy, cluttered", "guidance_scale": 8.0 }, "casual": { "name": "Casual Lifestyle", "modifier": "casual outdoor setting, natural environment, relaxed atmosphere, warm natural lighting", "negative_extra": "formal, studio", "guidance_scale": 7.5 }, "artistic": { "name": "Artistic Creative", "modifier": "artistic background, creative composition, vibrant colors, interesting lighting", "negative_extra": "boring, plain", "guidance_scale": 6.5 }, "nature": { "name": "Natural Scenery", "modifier": "beautiful natural scenery, outdoor landscape, scenic view, natural lighting", "negative_extra": "urban, indoor", "guidance_scale": 7.5 } } def __init__(self, device: str = "auto"): self.device = self._setup_device(device) # Model configurations - KEEP SAME FOR PERFECT GENERATION self.base_model_id = "stabilityai/stable-diffusion-xl-base-1.0" self.clip_model_name = "ViT-B-32" self.clip_pretrained = "openai" # Pipeline objects self.pipeline = None self.clip_model = None self.clip_preprocess = None self.clip_tokenizer = None self.is_initialized = False # Generation settings - KEEP SAME self.max_image_size = 1024 self.default_steps = 25 self.use_fp16 = True # Enhanced memory management self.generation_count = 0 self.cleanup_frequency = 1 # More frequent cleanup self.max_history = 3 # Limit generation history # Initialize helper classes self.mask_generator = MaskGenerator(self.max_image_size) self.image_blender = ImageBlender() self.quality_checker = QualityChecker() logger.info(f"OptimizedSceneWeaver initialized on {self.device}") def _setup_device(self, device: str) -> str: """Setup computation device""" if device == "auto": if torch.cuda.is_available(): return "cuda" elif hasattr(torch.backends, 'mps') and torch.backends.mps.is_available(): return "mps" else: return "cpu" return device def _ultra_memory_cleanup(self): """Ultra aggressive memory cleanup for Colab stability""" logger.debug("๐Ÿงน Ultra memory cleanup...") # Multiple rounds of garbage collection for i in range(5): gc.collect() if torch.cuda.is_available(): # Clear all cached memory torch.cuda.empty_cache() torch.cuda.ipc_collect() # Force synchronization torch.cuda.synchronize() # Clear any remaining memory fragments try: torch.cuda.memory.empty_cache() except: pass logger.debug("โœ… Ultra cleanup completed") def load_models(self, progress_callback: Optional[callable] = None): """Load AI models - KEEP SAME FOR PERFECT GENERATION""" if self.is_initialized: logger.info("Models already loaded") return logger.info("๐Ÿ“ฅ Loading AI models...") try: self._ultra_memory_cleanup() if progress_callback: progress_callback("Loading OpenCLIP for image understanding...", 20) # Load OpenCLIP - KEEP SAME self.clip_model, _, self.clip_preprocess = open_clip.create_model_and_transforms( self.clip_model_name, pretrained=self.clip_pretrained, device=self.device ) self.clip_tokenizer = open_clip.get_tokenizer(self.clip_model_name) self.clip_model.eval() logger.info("โœ… OpenCLIP loaded") if progress_callback: progress_callback("Loading SDXL text-to-image pipeline...", 60) # Load standard SDXL text-to-image pipeline - KEEP SAME self.pipeline = StableDiffusionXLPipeline.from_pretrained( self.base_model_id, torch_dtype=torch.float16 if self.use_fp16 else torch.float32, use_safetensors=True, variant="fp16" if self.use_fp16 else None ) # Use DPM solver for faster generation - KEEP SAME self.pipeline.scheduler = DPMSolverMultistepScheduler.from_config( self.pipeline.scheduler.config ) # Move to device self.pipeline = self.pipeline.to(self.device) if progress_callback: progress_callback("Applying optimizations...", 90) # Memory optimizations - ENHANCED try: self.pipeline.enable_xformers_memory_efficient_attention() logger.info("โœ… xformers enabled") except Exception: try: self.pipeline.enable_attention_slicing() logger.info("โœ… Attention slicing enabled") except Exception: logger.warning("โš ๏ธ No memory optimizations available") # Additional memory optimizations if hasattr(self.pipeline, 'enable_vae_tiling'): self.pipeline.enable_vae_tiling() if hasattr(self.pipeline, 'enable_vae_slicing'): self.pipeline.enable_vae_slicing() # Set to eval mode self.pipeline.unet.eval() if hasattr(self.pipeline, 'vae'): self.pipeline.vae.eval() # Enable sequential CPU offload if very low on memory try: if torch.cuda.is_available(): free_memory = torch.cuda.get_device_properties(0).total_memory - torch.cuda.memory_allocated() if free_memory < 4 * 1024**3: # Less than 4GB free self.pipeline.enable_sequential_cpu_offload() logger.info("โœ… Sequential CPU offload enabled for low memory") except: pass self.is_initialized = True if progress_callback: progress_callback("Models loaded successfully!", 100) # Memory status if torch.cuda.is_available(): memory_used = torch.cuda.memory_allocated() / 1024**3 memory_total = torch.cuda.get_device_properties(0).total_memory / 1024**3 logger.info(f"๐Ÿ“Š GPU Memory: {memory_used:.1f}GB / {memory_total:.1f}GB") except Exception as e: logger.error(f"โŒ Model loading failed: {e}") raise RuntimeError(f"Failed to load models: {str(e)}") def analyze_image_with_clip(self, image: Image.Image) -> str: """Analyze uploaded image using OpenCLIP - KEEP SAME""" if not self.clip_model: return "Image analysis not available" try: image_input = self.clip_preprocess(image).unsqueeze(0).to(self.device) categories = [ "a photo of a person", "a photo of an animal", "a photo of an object", "a photo of a character", "a photo of a cartoon", "a photo of nature", "a photo of a building", "a photo of a landscape" ] text_inputs = self.clip_tokenizer(categories).to(self.device) with torch.no_grad(): image_features = self.clip_model.encode_image(image_input) text_features = self.clip_model.encode_text(text_inputs) image_features /= image_features.norm(dim=-1, keepdim=True) text_features /= text_features.norm(dim=-1, keepdim=True) similarity = (100.0 * image_features @ text_features.T).softmax(dim=-1) best_match_idx = similarity.argmax().item() confidence = similarity[0, best_match_idx].item() category = categories[best_match_idx].replace("a photo of ", "") return f"Detected: {category} (confidence: {confidence:.1%})" except Exception as e: logger.error(f"CLIP analysis failed: {e}") return "Image analysis failed" def enhance_prompt( self, user_prompt: str, foreground_image: Image.Image ) -> str: """ Smart prompt enhancement based on image analysis. Adds appropriate lighting, atmosphere, and quality descriptors. Args: user_prompt: Original user-provided prompt foreground_image: Foreground image for analysis Returns: Enhanced prompt string """ logger.info("โœจ Enhancing prompt based on image analysis...") try: # Analyze image characteristics img_array = np.array(foreground_image.convert('RGB')) # === Analyze color temperature === # Convert to LAB to analyze color temperature lab = cv2.cvtColor(img_array, cv2.COLOR_RGB2LAB) avg_a = np.mean(lab[:, :, 1]) # a channel: green(-) to red(+) avg_b = np.mean(lab[:, :, 2]) # b channel: blue(-) to yellow(+) # Determine warm/cool tone is_warm = avg_b > 128 # b > 128 means more yellow/warm # === Analyze brightness === gray = cv2.cvtColor(img_array, cv2.COLOR_RGB2GRAY) avg_brightness = np.mean(gray) is_bright = avg_brightness > 127 # === Get subject type from CLIP === clip_analysis = self.analyze_image_with_clip(foreground_image) subject_type = "unknown" if "person" in clip_analysis.lower(): subject_type = "person" elif "animal" in clip_analysis.lower(): subject_type = "animal" elif "object" in clip_analysis.lower(): subject_type = "object" elif "character" in clip_analysis.lower() or "cartoon" in clip_analysis.lower(): subject_type = "character" elif "nature" in clip_analysis.lower() or "landscape" in clip_analysis.lower(): subject_type = "nature" # === Build prompt fragments library === lighting_options = { "warm_bright": "warm golden hour lighting, soft natural light", "warm_dark": "warm ambient lighting, cozy atmosphere", "cool_bright": "bright daylight, clear sky lighting", "cool_dark": "soft diffused light, gentle shadows" } atmosphere_options = { "person": "professional, elegant composition", "animal": "natural, harmonious setting", "object": "clean product photography style", "character": "artistic, vibrant, imaginative", "nature": "scenic, peaceful atmosphere", "unknown": "balanced composition" } quality_modifiers = "high quality, detailed, sharp focus, photorealistic" # === Select appropriate fragments === # Lighting based on color temperature and brightness if is_warm and is_bright: lighting = lighting_options["warm_bright"] elif is_warm and not is_bright: lighting = lighting_options["warm_dark"] elif not is_warm and is_bright: lighting = lighting_options["cool_bright"] else: lighting = lighting_options["cool_dark"] # Atmosphere based on subject type atmosphere = atmosphere_options.get(subject_type, atmosphere_options["unknown"]) # === Check for conflicts in user prompt === user_prompt_lower = user_prompt.lower() # Avoid adding conflicting descriptions if "sunset" in user_prompt_lower or "golden" in user_prompt_lower: lighting = "" # User already specified lighting if "dark" in user_prompt_lower or "night" in user_prompt_lower: lighting = lighting.replace("bright", "").replace("daylight", "") # === Combine enhanced prompt === fragments = [user_prompt] if lighting: fragments.append(lighting) if atmosphere: fragments.append(atmosphere) fragments.append(quality_modifiers) enhanced_prompt = ", ".join(filter(None, fragments)) logger.info(f"๐Ÿ“ Original prompt: {user_prompt[:50]}...") logger.info(f"๐Ÿ“ Enhanced prompt: {enhanced_prompt[:80]}...") return enhanced_prompt except Exception as e: logger.warning(f"โš ๏ธ Prompt enhancement failed: {e}, using original prompt") return user_prompt def _prepare_image(self, image: Image.Image) -> Image.Image: """Prepare image for processing - KEEP SAME""" # Convert to RGB if image.mode != 'RGB': image = image.convert('RGB') # Resize if too large width, height = image.size max_size = self.max_image_size if width > max_size or height > max_size: ratio = min(max_size/width, max_size/height) new_width = int(width * ratio) new_height = int(height * ratio) image = image.resize((new_width, new_height), Image.LANCZOS) # Ensure dimensions are multiple of 8 width, height = image.size new_width = (width // 8) * 8 new_height = (height // 8) * 8 if new_width != width or new_height != height: image = image.resize((new_width, new_height), Image.LANCZOS) return image def generate_background( self, prompt: str, width: int, height: int, negative_prompt: str = "blurry, low quality, distorted", num_inference_steps: int = 25, guidance_scale: float = 7.5, progress_callback: Optional[callable] = None ) -> Image.Image: """Generate complete background using standard text-to-image - KEEP SAME""" if not self.is_initialized: raise RuntimeError("Models not loaded. Call load_models() first.") logger.info(f"๐ŸŽจ Generating background: {prompt[:50]}...") try: with torch.inference_mode(): if progress_callback: progress_callback("Generating background with SDXL...", 50) # Standard text-to-image generation - KEEP SAME result = self.pipeline( prompt=prompt, negative_prompt=negative_prompt, width=width, height=height, num_inference_steps=num_inference_steps, guidance_scale=guidance_scale, generator=torch.Generator(device=self.device).manual_seed(42) ) generated_image = result.images[0] if progress_callback: progress_callback("Background generated successfully!", 100) logger.info("โœ… Background generation completed!") return generated_image except torch.cuda.OutOfMemoryError: logger.error("โŒ GPU memory exhausted") self._ultra_memory_cleanup() raise RuntimeError("GPU memory insufficient") except Exception as e: logger.error(f"โŒ Background generation failed: {e}") raise RuntimeError(f"Generation failed: {str(e)}") def generate_and_combine( self, original_image: Image.Image, prompt: str, combination_mode: str = "center", focus_mode: str = "person", negative_prompt: str = "blurry, low quality, distorted", num_inference_steps: int = 25, guidance_scale: float = 7.5, progress_callback: Optional[callable] = None, enable_prompt_enhancement: bool = True ) -> Dict[str, Any]: """ Generate background and combine with foreground using advanced blending. Args: original_image: Foreground image prompt: User's background description combination_mode: How to position foreground ("center", "left_half", "right_half", "full") focus_mode: Focus type ("person" for tight crop, "scene" for wider context) negative_prompt: What to avoid in generation num_inference_steps: SDXL inference steps guidance_scale: Classifier-free guidance scale progress_callback: Progress reporting callback enable_prompt_enhancement: Whether to use smart prompt enhancement Returns: Dictionary containing results and metadata """ if not self.is_initialized: raise RuntimeError("Models not loaded. Call load_models() first.") logger.info(f"๐ŸŽจ Starting generation and combination with advanced features...") try: # Enhanced memory management if self.generation_count % self.cleanup_frequency == 0: self._ultra_memory_cleanup() if progress_callback: progress_callback("Analyzing uploaded image...", 5) # Analyze original image image_analysis = self.analyze_image_with_clip(original_image) if progress_callback: progress_callback("Preparing images...", 10) # Prepare original image processed_original = self._prepare_image(original_image) target_width, target_height = processed_original.size if progress_callback: progress_callback("Optimizing prompt...", 15) # Smart prompt enhancement if enable_prompt_enhancement: enhanced_prompt = self.enhance_prompt(prompt, processed_original) else: enhanced_prompt = f"{prompt}, high quality, detailed, photorealistic, beautiful scenery" enhanced_negative = f"{negative_prompt}, people, characters, cartoons, logos" if progress_callback: progress_callback("Generating complete background scene...", 25) def bg_progress(msg, pct): if progress_callback: progress_callback(f"Background: {msg}", 25 + (pct/100) * 50) generated_background = self.generate_background( prompt=enhanced_prompt, width=target_width, height=target_height, negative_prompt=enhanced_negative, num_inference_steps=num_inference_steps, guidance_scale=guidance_scale, progress_callback=bg_progress ) if progress_callback: progress_callback("Creating intelligent mask for person detection...", 80) # Use intelligent mask generation with enhanced logging logger.info("๐ŸŽญ Starting intelligent mask generation...") combination_mask = self.mask_generator.create_gradient_based_mask( processed_original, combination_mode, focus_mode ) # Log mask quality for debugging try: mask_array = np.array(combination_mask) logger.info(f"๐Ÿ“Š Generated mask stats - Mean: {mask_array.mean():.1f}, Non-zero pixels: {np.count_nonzero(mask_array)}") except Exception as mask_debug_error: logger.warning(f"โš ๏ธ Mask debug logging failed: {mask_debug_error}") if progress_callback: progress_callback("Advanced image blending...", 90) # Use advanced image blending with logging logger.info("๐Ÿ–Œ๏ธ Starting advanced image blending...") combined_image = self.image_blender.simple_blend_images( processed_original, generated_background, combination_mask ) logger.info("โœ… Image blending completed successfully") if progress_callback: progress_callback("Creating debug images...", 95) # Generate debug images debug_images = self.image_blender.create_debug_images( processed_original, generated_background, combination_mask, combined_image ) # Memory cleanup after generation self._ultra_memory_cleanup() # Update generation count self.generation_count += 1 if progress_callback: progress_callback("Generation complete!", 100) logger.info("โœ… Complete generation and combination with fixed blending successful!") return { "combined_image": combined_image, "generated_scene": generated_background, "original_image": processed_original, "combination_mask": combination_mask, "debug_mask_gray": debug_images["mask_gray"], "debug_alpha_heatmap": debug_images["alpha_heatmap"], "image_analysis": image_analysis, "enhanced_prompt": enhanced_prompt, "original_prompt": prompt, "success": True, "generation_count": self.generation_count } except Exception as e: import traceback error_traceback = traceback.format_exc() logger.error(f"โŒ Generation and combination failed: {str(e)}") logger.error(f"๐Ÿ“ Full traceback:\n{error_traceback}") print(f"โŒ DETAILED ERROR in scene_weaver_core.generate_and_combine:") print(f"Error: {str(e)}") print(f"Traceback:\n{error_traceback}") self._ultra_memory_cleanup() # Cleanup on error too return { "success": False, "error": f"Failed: {str(e)}" } def generate_diversity_variants( self, original_image: Image.Image, prompt: str, selected_styles: Optional[List[str]] = None, combination_mode: str = "center", focus_mode: str = "person", negative_prompt: str = "blurry, low quality, distorted", progress_callback: Optional[callable] = None ) -> Dict[str, Any]: """ Generate multiple style variants of the background. Uses reduced quality for faster preview generation. Args: original_image: Foreground image prompt: Base background description selected_styles: List of style keys to use (None = all styles) combination_mode: Foreground positioning mode focus_mode: Focus type for mask generation negative_prompt: Base negative prompt progress_callback: Progress callback function Returns: Dictionary containing variants and metadata """ if not self.is_initialized: raise RuntimeError("Models not loaded. Call load_models() first.") logger.info("๐ŸŽจ Starting diversity generation mode...") # Determine which styles to generate styles_to_generate = selected_styles or list(self.STYLE_PRESETS.keys()) num_styles = len(styles_to_generate) results = { "variants": [], "success": True, "num_variants": 0 } try: # Pre-process image once processed_original = self._prepare_image(original_image) target_width, target_height = processed_original.size # Reduce resolution for faster generation preview_size = min(768, max(target_width, target_height)) scale = preview_size / max(target_width, target_height) preview_width = int(target_width * scale) // 8 * 8 preview_height = int(target_height * scale) // 8 * 8 # Generate mask once (reusable for all variants) if progress_callback: progress_callback("Creating foreground mask...", 5) combination_mask = self.mask_generator.create_gradient_based_mask( processed_original, combination_mode, focus_mode ) # Resize mask for preview resolution preview_mask = combination_mask.resize((preview_width, preview_height), Image.LANCZOS) preview_original = processed_original.resize((preview_width, preview_height), Image.LANCZOS) # Generate each style variant for idx, style_key in enumerate(styles_to_generate): if style_key not in self.STYLE_PRESETS: logger.warning(f"โš ๏ธ Unknown style: {style_key}, skipping") continue style = self.STYLE_PRESETS[style_key] style_name = style["name"] if progress_callback: base_pct = 10 + (idx / num_styles) * 80 progress_callback(f"Generating {style_name} variant...", int(base_pct)) logger.info(f"๐ŸŽจ Generating variant: {style_name}") try: # Build style-specific prompt styled_prompt = f"{prompt}, {style['modifier']}, high quality, detailed" styled_negative = f"{negative_prompt}, {style['negative_extra']}, people, characters" # Generate background with reduced steps for speed background = self.generate_background( prompt=styled_prompt, width=preview_width, height=preview_height, negative_prompt=styled_negative, num_inference_steps=15, # Reduced for speed guidance_scale=style["guidance_scale"] ) # Blend images combined = self.image_blender.simple_blend_images( preview_original, background, preview_mask, use_multi_scale=False # Skip for speed ) results["variants"].append({ "style_key": style_key, "style_name": style_name, "combined_image": combined, "background": background, "prompt_used": styled_prompt }) # Memory cleanup between variants self._ultra_memory_cleanup() except Exception as variant_error: logger.error(f"โŒ Failed to generate {style_name} variant: {variant_error}") continue results["num_variants"] = len(results["variants"]) if progress_callback: progress_callback("Diversity generation complete!", 100) logger.info(f"โœ… Generated {results['num_variants']} style variants") return results except Exception as e: logger.error(f"โŒ Diversity generation failed: {e}") self._ultra_memory_cleanup() return { "variants": [], "success": False, "error": str(e), "num_variants": 0 } def regenerate_high_quality( self, original_image: Image.Image, prompt: str, style_key: str, combination_mode: str = "center", focus_mode: str = "person", negative_prompt: str = "blurry, low quality, distorted", progress_callback: Optional[callable] = None ) -> Dict[str, Any]: """ Regenerate a specific style at full quality. Args: original_image: Original foreground image prompt: Base prompt style_key: Style preset key to use combination_mode: Foreground positioning focus_mode: Mask focus mode negative_prompt: Base negative prompt progress_callback: Progress callback Returns: Full quality result dictionary """ if style_key not in self.STYLE_PRESETS: return {"success": False, "error": f"Unknown style: {style_key}"} style = self.STYLE_PRESETS[style_key] # Build styled prompt styled_prompt = f"{prompt}, {style['modifier']}" styled_negative = f"{negative_prompt}, {style['negative_extra']}" # Use full generate_and_combine with style parameters return self.generate_and_combine( original_image=original_image, prompt=styled_prompt, combination_mode=combination_mode, focus_mode=focus_mode, negative_prompt=styled_negative, num_inference_steps=25, # Full quality guidance_scale=style["guidance_scale"], progress_callback=progress_callback, enable_prompt_enhancement=True ) def get_memory_status(self) -> Dict[str, Any]: """Enhanced memory status reporting""" status = {"device": self.device} if torch.cuda.is_available(): allocated = torch.cuda.memory_allocated() / 1024**3 total = torch.cuda.get_device_properties(0).total_memory / 1024**3 cached = torch.cuda.memory_reserved() / 1024**3 status.update({ "gpu_allocated_gb": round(allocated, 2), "gpu_total_gb": round(total, 2), "gpu_cached_gb": round(cached, 2), "gpu_free_gb": round(total - allocated, 2), "gpu_usage_percent": round((allocated / total) * 100, 1), "generation_count": self.generation_count }) return status