Deep Dive: How Diffusion Models Work vs. GANs 2.0 for 2026 Image Generation

import torch
from diffusers import StableDiffusionPipeline, DDPMScheduler
from torch.utils.data import DataLoader
from torchvision import datasets, transforms
import logging
from pathlib import Path
import json

# Configure logging for training traceability
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.StreamHandler(), logging.FileHandler("diffusion_train.log")]
)
logger = logging.getLogger(__name__)

# Training configuration with benchmark-aligned settings
CONFIG = {
    "model_id": "stabilityai/stable-diffusion-3.5-medium",
    "batch_size": 8,
    "learning_rate": 1e-5,
    "num_epochs": 10,
    "image_size": 512,
    "custom_data_dir": Path("./custom_brand_assets"),
    "output_dir": Path("./sd3_5_finetuned"),
    "save_steps": 500,
    "mixed_precision": "fp16",
    "hardware": "NVIDIA A100 80GB"  # Benchmark hardware reference
}

def load_custom_dataset(config):
    """Load and preprocess custom brand asset dataset with error handling."""
    try:
        transform = transforms.Compose([
            transforms.Resize((config["image_size"], config["image_size"])),
            transforms.CenterCrop(config["image_size"]),
            transforms.ToTensor(),
            transforms.Normalize([0.5], [0.5])
        ])
        dataset = datasets.ImageFolder(config["custom_data_dir"], transform=transform)
        logger.info(f"Loaded {len(dataset)} samples from {config['custom_data_dir']}")
        return dataset
    except Exception as e:
        logger.error(f"Failed to load dataset: {str(e)}")
        raise

def train_diffusion_model(config):
    """Fine-tune Stable Diffusion 3.5 on custom data with OOM error handling."""
    try:
        # Initialize model with pretrained weights
        pipe = StableDiffusionPipeline.from_pretrained(
            config["model_id"],
            torch_dtype=torch.float16 if config["mixed_precision"] == "fp16" else torch.float32
        )
        pipe.to("cuda")
        logger.info(f"Initialized {config['model_id']} on {config['hardware']}")

        # Load dataset and dataloader
        dataset = load_custom_dataset(config)
        dataloader = DataLoader(dataset, batch_size=config["batch_size"], shuffle=True)

        # Configure optimizer and scheduler
        optimizer = torch.optim.AdamW(pipe.unet.parameters(), lr=config["learning_rate"])
        scheduler = DDPMScheduler.from_pretrained(config["model_id"], subfolder="scheduler")

        # Training loop with step-level error handling
        global_step = 0
        for epoch in range(config["num_epochs"]):
            pipe.unet.train()
            for batch in dataloader:
                try:
                    images, _ = batch
                    images = images.to("cuda", dtype=torch.float16 if config["mixed_precision"] == "fp16" else torch.float32)

                    # Forward pass with noise scheduling
                    noise = torch.randn_like(images)
                    timesteps = torch.randint(0, scheduler.num_train_timesteps, (images.shape[0],), device="cuda")
                    noisy_images = scheduler.add_noise(images, noise, timesteps)

                    # Predict noise and calculate loss
                    noise_pred = pipe.unet(noisy_images, timesteps).sample
                    loss = torch.nn.functional.mse_loss(noise_pred, noise)

                    # Backward pass and optimization
                    optimizer.zero_grad()
                    loss.backward()
                    optimizer.step()

                    global_step += 1
                    if global_step % 100 == 0:
                        logger.info(f"Epoch {epoch+1}, Step {global_step}, Loss: {loss.item():.4f}")

                    # Save checkpoint
                    if global_step % config["save_steps"] == 0:
                        pipe.save_pretrained(config["output_dir"] / f"checkpoint-{global_step}")
                        logger.info(f"Saved checkpoint to {config['output_dir']}/checkpoint-{global_step}")
                except RuntimeError as e:
                    if "CUDA out of memory" in str(e):
                        logger.warning(f"OOM error at step {global_step}, reducing batch size to {config['batch_size']//2}")
                        config["batch_size"] = config["batch_size"]//2
                        dataloader = DataLoader(dataset, batch_size=config["batch_size"], shuffle=True)
                    else:
                        logger.error(f"Training error at step {global_step}: {str(e)}")
                        raise
        # Save final model
        pipe.save_pretrained(config["output_dir"] / "final")
        logger.info(f"Training complete. Final model saved to {config['output_dir']}/final")
        return pipe
    except Exception as e:
        logger.error(f"Training failed: {str(e)}")
        raise

if __name__ == "__main__":
    # Validate config paths
    if not CONFIG["custom_data_dir"].exists():
        raise FileNotFoundError(f"Custom data directory {CONFIG['custom_data_dir']} does not exist")
    CONFIG["output_dir"].mkdir(parents=True, exist_ok=True)
    # Save config for reproducibility
    with open(CONFIG["output_dir"] / "train_config.json", "w") as f:
        json.dump(CONFIG, f, indent=2)
    logger.info(f"Starting diffusion training with config: {CONFIG}")
    train_diffusion_model(CONFIG)

import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader
from torchvision import datasets, transforms
import logging
from pathlib import Path
import json
import sys
sys.path.append("./stylegan3")  # Assumes stylegan3 cloned from https://github.com/NVlabs/stylegan3
from training.networks import Generator, Discriminator
from training.loss import StyleGAN2Loss

# Configure logging for GAN training traceability
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.StreamHandler(), logging.FileHandler("gan_train.log")]
)
logger = logging.getLogger(__name__)

# GANs 2.0 (StyleGAN3-Turbo) training configuration
CONFIG = {
    "generator_arch": "stylegan3-turbo",
    "discriminator_arch": "stylegan3-turbo-d",
    "batch_size": 32,
    "learning_rate_g": 2e-3,
    "learning_rate_d": 2e-3,
    "num_epochs": 15,
    "image_size": 512,
    "custom_data_dir": Path("./custom_brand_assets"),
    "output_dir": Path("./stylegan3_turbo_finetuned"),
    "save_steps": 1000,
    "mixed_precision": "fp16",
    "mode_collapse_threshold": 0.05,  # 5% mode collapse threshold
    "hardware": "NVIDIA A100 80GB"  # Benchmark hardware reference
}

def load_gan_dataset(config):
    """Load custom dataset for GAN training with augmentation to prevent mode collapse."""
    try:
        transform = transforms.Compose([
            transforms.Resize((config["image_size"], config["image_size"])),
            transforms.RandomHorizontalFlip(0.5),
            transforms.RandomRotation(10),
            transforms.ToTensor(),
            transforms.Normalize([0.5, 0.5, 0.5], [0.5, 0.5, 0.5])
        ])
        dataset = datasets.ImageFolder(config["custom_data_dir"], transform=transform)
        logger.info(f"Loaded {len(dataset)} samples for GAN training from {config['custom_data_dir']}")
        return dataset
    except Exception as e:
        logger.error(f"Failed to load GAN dataset: {str(e)}")
        raise

def calculate_mode_collapse(activations):
    """Calculate mode collapse score using feature activations (lower is better)."""
    try:
        # Use pairwise cosine similarity to detect duplicate modes
        activations = activations.view(activations.shape[0], -1)
        cos_sim = torch.nn.functional.cosine_similarity(activations.unsqueeze(1), activations.unsqueeze(0), dim=2)
        # Exclude self-similarity
        cos_sim.fill_diagonal_(0)
        mean_sim = cos_sim.mean().item()
        return mean_sim
    except Exception as e:
        logger.error(f"Mode collapse calculation failed: {str(e)}")
        return 1.0  # Default to high collapse if calculation fails

def train_gan_model(config):
    """Train StyleGAN3-Turbo on custom data with mode collapse detection."""
    try:
        # Initialize generator and discriminator
        G = Generator(z_dim=512, w_dim=512, img_resolution=config["image_size"], img_channels=3).to("cuda")
        D = Discriminator(img_resolution=config["image_size"], img_channels=3).to("cuda")
        logger.info(f"Initialized {config['generator_arch']} on {config['hardware']}")

        # Load dataset and dataloader
        dataset = load_gan_dataset(config)
        dataloader = DataLoader(dataset, batch_size=config["batch_size"], shuffle=True, num_workers=4)

        # Initialize optimizers and loss
        optimizer_G = optim.Adam(G.parameters(), lr=config["learning_rate_g"], betas=(0.0, 0.99))
        optimizer_D = optim.Adam(D.parameters(), lr=config["learning_rate_d"], betas=(0.0, 0.99))
        loss_fn = StyleGAN2Loss(device="cuda")

        # Training loop with mode collapse monitoring
        global_step = 0
        for epoch in range(config["num_epochs"]):
            G.train()
            D.train()
            for batch in dataloader:
                try:
                    real_images, _ = batch
                    real_images = real_images.to("cuda", dtype=torch.float16 if config["mixed_precision"] == "fp16" else torch.float32)
                    batch_size = real_images.shape[0]

                    # Train Discriminator
                    optimizer_D.zero_grad()
                    # Generate fake images
                    z = torch.randn(batch_size, 512, device="cuda")
                    fake_images = G(z, None)
                    # Discriminator loss
                    d_loss = loss_fn.d_loss(D, real_images, fake_images)
                    d_loss.backward()
                    optimizer_D.step()

                    # Train Generator every 2 steps to stabilize training
                    if global_step % 2 == 0:
                        optimizer_G.zero_grad()
                        z = torch.randn(batch_size, 512, device="cuda")
                        fake_images = G(z, None)
                        g_loss = loss_fn.g_loss(D, fake_images)
                        g_loss.backward()
                        optimizer_G.step()

                    # Mode collapse check every 100 steps
                    if global_step % 100 == 0:
                        with torch.no_grad():
                            z = torch.randn(100, 512, device="cuda")
                            fake_samples = G(z, None)
                            mode_collapse = calculate_mode_collapse(fake_samples)
                            logger.info(f"Epoch {epoch+1}, Step {global_step}, D Loss: {d_loss.item():.4f}, G Loss: {g_loss.item():.4f}, Mode Collapse: {mode_collapse:.4f}")
                            if mode_collapse > config["mode_collapse_threshold"]:
                                logger.warning(f"Mode collapse threshold exceeded: {mode_collapse:.4f} > {config['mode_collapse_threshold']}")

                    global_step += 1
                    # Save checkpoint
                    if global_step % config["save_steps"] == 0:
                        torch.save(G.state_dict(), config["output_dir"] / f"g_checkpoint-{global_step}.pth")
                        torch.save(D.state_dict(), config["output_dir"] / f"d_checkpoint-{global_step}.pth")
                        logger.info(f"Saved GAN checkpoints to {config['output_dir']}")
                except RuntimeError as e:
                    if "CUDA out of memory" in str(e):
                        logger.warning(f"OOM error at step {global_step}, reducing batch size to {config['batch_size']//2}")
                        config["batch_size"] = config["batch_size"]//2
                        dataloader = DataLoader(dataset, batch_size=config["batch_size"], shuffle=True, num_workers=2)
                    else:
                        logger.error(f"GAN training error at step {global_step}: {str(e)}")
                        raise
        # Save final models
        torch.save(G.state_dict(), config["output_dir"] / "g_final.pth")
        torch.save(D.state_dict(), config["output_dir"] / "d_final.pth")
        logger.info(f"GAN training complete. Final models saved to {config['output_dir']}")
        return G, D
    except Exception as e:
        logger.error(f"GAN training failed: {str(e)}")
        raise

if __name__ == "__main__":
    # Validate config and clone stylegan3 if not present
    if not Path("./stylegan3").exists():
        logger.info("Cloning StyleGAN3 from https://github.com/NVlabs/stylegan3")
        import subprocess
        subprocess.run(["git", "clone", "https://github.com/NVlabs/stylegan3"], check=True)
    if not CONFIG["custom_data_dir"].exists():
        raise FileNotFoundError(f"Custom data directory {CONFIG['custom_data_dir']} does not exist")
    CONFIG["output_dir"].mkdir(parents=True, exist_ok=True)
    # Save config for reproducibility
    with open(CONFIG["output_dir"] / "gan_train_config.json", "w") as f:
        json.dump(CONFIG, f, indent=2)
    logger.info(f"Starting GAN training with config: {CONFIG}")
    train_gan_model(CONFIG)

import torch
import time
import logging
from diffusers import StableDiffusionPipeline
from pathlib import Path
import json
import sys
sys.path.append("./stylegan3")
from training.networks import Generator
from torchvision.utils import save_image
from piq import FID  # https://github.com/photosynthesis-team/piq for FID calculation

# Configure logging for inference benchmarking
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.StreamHandler(), logging.FileHandler("inference_benchmark.log")]
)
logger = logging.getLogger(__name__)

# Inference configuration aligned with benchmark methodology
CONFIG = {
    "diffusion_model_path": Path("./sd3_5_finetuned/final"),
    "gan_model_path": Path("./stylegan3_turbo_finetuned/g_final.pth"),
    "gan_config": {
        "z_dim": 512,
        "w_dim": 512,
        "img_resolution": 512,
        "img_channels": 3
    },
    "num_samples": 1000,
    "batch_size": 8,
    "image_size": 512,
    "output_dir": Path("./benchmark_results"),
    "prompt": "A modern brand-compliant logo for a tech startup, minimalist, blue and white",
    "hardware": "NVIDIA A100 80GB",
    "pytorch_version": "2.4.0",
    "cuda_version": "12.4"
}

def load_diffusion_pipeline(config):
    """Load fine-tuned diffusion model for inference."""
    try:
        pipe = StableDiffusionPipeline.from_pretrained(
            config["diffusion_model_path"],
            torch_dtype=torch.float16
        )
        pipe.to("cuda")
        pipe.enable_xformers_memory_efficient_attention()
        logger.info(f"Loaded diffusion pipeline from {config['diffusion_model_path']}")
        return pipe
    except Exception as e:
        logger.error(f"Failed to load diffusion pipeline: {str(e)}")
        raise

def load_gan_generator(config):
    """Load fine-tuned GAN generator for inference."""
    try:
        G = Generator(
            z_dim=config["gan_config"]["z_dim"],
            w_dim=config["gan_config"]["w_dim"],
            img_resolution=config["gan_config"]["img_resolution"],
            img_channels=config["gan_config"]["img_channels"]
        ).to("cuda")
        G.load_state_dict(torch.load(config["gan_model_path"], map_location="cuda"))
        G.eval()
        logger.info(f"Loaded GAN generator from {config['gan_model_path']}")
        return G
    except Exception as e:
        logger.error(f"Failed to load GAN generator: {str(e)}")
        raise

def benchmark_diffusion(pipe, config):
    """Benchmark diffusion model inference latency and FID."""
    try:
        latencies = []
        generated_images = []
        logger.info(f"Starting diffusion benchmark for {config['num_samples']} samples")
        for i in range(0, config["num_samples"], config["batch_size"]):
            batch_size = min(config["batch_size"], config["num_samples"] - i)
            start_time = time.time()
            # Generate images with error handling for individual batches
            try:
                outputs = pipe(
                    [config["prompt"]] * batch_size,
                    num_inference_steps=30,  # 2026 default for SD3.5
                    guidance_scale=7.5
                )
                latency = time.time() - start_time
                latencies.append(latency / batch_size)
                # Convert to tensor for FID calculation
                for img in outputs.images:
                    img_tensor = transforms.ToTensor()(img)
                    generated_images.append(img_tensor)
                # Save sample images
                if i < 10:
                    save_image(img_tensor, config["output_dir"] / f"diffusion_sample_{i}.png")
            except Exception as e:
                logger.error(f"Diffusion batch {i} failed: {str(e)}")
                continue
        avg_latency = sum(latencies) / len(latencies) * 1000  # ms per image
        logger.info(f"Diffusion avg latency: {avg_latency:.2f} ms per image")
        return avg_latency, generated_images
    except Exception as e:
        logger.error(f"Diffusion benchmark failed: {str(e)}")
        raise

def benchmark_gan(generator, config):
    """Benchmark GAN model inference latency and FID."""
    try:
        latencies = []
        generated_images = []
        logger.info(f"Starting GAN benchmark for {config['num_samples']} samples")
        for i in range(0, config["num_samples"], config["batch_size"]):
            batch_size = min(config["batch_size"], config["num_samples"] - i)
            start_time = time.time()
            try:
                z = torch.randn(batch_size, config["gan_config"]["z_dim"], device="cuda")
                with torch.no_grad():
                    fake_images = generator(z, None)
                latency = time.time() - start_time
                latencies.append(latency / batch_size)
                # Normalize and convert to PIL for saving
                fake_images = (fake_images + 1) / 2  # Denormalize from [-1,1] to [0,1]
                for j in range(batch_size):
                    img_tensor = fake_images[j].cpu()
                    generated_images.append(img_tensor)
                    if i < 10:
                        save_image(img_tensor, config["output_dir"] / f"gan_sample_{i+j}.png")
            except Exception as e:
                logger.error(f"GAN batch {i} failed: {str(e)}")
                continue
        avg_latency = sum(latencies) / len(latencies) * 1000  # ms per image
        logger.info(f"GAN avg latency: {avg_latency:.2f} ms per image")
        return avg_latency, generated_images
    except Exception as e:
        logger.error(f"GAN benchmark failed: {str(e)}")
        raise

def calculate_fid(generated_images, real_images):
    """Calculate FID between generated and real images."""
    try:
        fid = FID()
        # Convert lists to tensors
        generated = torch.stack(generated_images).to("cuda")
        real = torch.stack(real_images).to("cuda")
        fid_score = fid(generated, real)
        logger.info(f"FID Score: {fid_score:.2f}")
        return fid_score
    except Exception as e:
        logger.error(f"FID calculation failed: {str(e)}")
        return 0.0

if __name__ == "__main__":
    # Validate paths
    if not CONFIG["diffusion_model_path"].exists():
        raise FileNotFoundError(f"Diffusion model path {CONFIG['diffusion_model_path']} does not exist")
    if not CONFIG["gan_model_path"].exists():
        raise FileNotFoundError(f"GAN model path {CONFIG['gan_model_path']} does not exist")
    CONFIG["output_dir"].mkdir(parents=True, exist_ok=True)
    # Save benchmark config
    with open(CONFIG["output_dir"] / "benchmark_config.json", "w") as f:
        json.dump(CONFIG, f, indent=2)
    # Load models
    diffusion_pipe = load_diffusion_pipeline(CONFIG)
    gan_generator = load_gan_generator(CONFIG)
    # Run benchmarks
    logger.info("Running diffusion benchmark...")
    diffusion_latency, diffusion_images = benchmark_diffusion(diffusion_pipe, CONFIG)
    logger.info("Running GAN benchmark...")
    gan_latency, gan_images = benchmark_gan(gan_generator, CONFIG)
    # Save results
    results = {
        "diffusion_avg_latency_ms": diffusion_latency,
        "gan_avg_latency_ms": gan_latency,
        "diffusion_fid": None,  # FID requires real images, omitted for brevity
        "gan_fid": None,
        "benchmark_methodology": {
            "hardware": CONFIG["hardware"],
            "pytorch_version": CONFIG["pytorch_version"],
            "cuda_version": CONFIG["cuda_version"],
            "num_samples": CONFIG["num_samples"]
        }
    }
    with open(CONFIG["output_dir"] / "benchmark_results.json", "w") as f:
        json.dump(results, f, indent=2)
    logger.info(f"Benchmark complete. Results saved to {CONFIG['output_dir']}")
    # Print summary
    print(f"\n=== Benchmark Summary ===")
    print(f"Diffusion Avg Latency: {diffusion_latency:.2f} ms per image")
    print(f"GAN Avg Latency: {gan_latency:.2f} ms per image")
    print(f"GAN is {diffusion_latency/gan_latency:.1f}x faster than diffusion")

# Log FID to Weights & Biases during benchmarking
import wandb
wandb.init(project="image-gen-benchmarks", config=CONFIG)
fid_score = calculate_fid(generated_images, real_images)
wandb.log({
    "fid_score": fid_score,
    "model_type": "diffusion",
    "latency_ms": avg_latency,
    "dataset": "custom_brand_assets"
})

# Enable gradient checkpointing for diffusion model training
from diffusers import StableDiffusionPipeline
pipe = StableDiffusionPipeline.from_pretrained("stabilityai/stable-diffusion-3.5-medium")
pipe.unet.enable_gradient_checkpointing()
print("Gradient checkpointing enabled for UNet")

# Step-level mode collapse detection for GAN training
def detect_mode_collapse(generator, batch_size=100, threshold=0.05):
    z = torch.randn(batch_size, 512, device="cuda")
    with torch.no_grad():
        fake_images = generator(z, None).view(batch_size, -1)
    cos_sim = torch.nn.functional.cosine_similarity(fake_images.unsqueeze(1), fake_images.unsqueeze(0), dim=2)
    cos_sim.fill_diagonal_(0)
    mean_sim = cos_sim.mean().item()
    return mean_sim > threshold, mean_sim