test.py

# ---------------------------------------------------------------
# Copyright (c) 2022, NVIDIA CORPORATION. All rights reserved.
#
# This work is licensed under the NVIDIA Source Code License
# for Denoising Diffusion GAN. To view a copy of this license, see the LICENSE file.
# ---------------------------------------------------------------
import argparse
import json
import os

import numpy as np
import torch
import torchvision
from pytorch_fid.fid_score import calculate_fid_given_paths
from score_sde.models.ncsnpp_generator_adagn import NCSNpp


# %% Diffusion coefficients
def var_func_vp(t, beta_min, beta_max):
    log_mean_coeff = -0.25 * t**2 * (beta_max - beta_min) - 0.5 * t * beta_min
    var = 1.0 - torch.exp(2.0 * log_mean_coeff)
    return var


def var_func_geometric(t, beta_min, beta_max):
    return beta_min * ((beta_max / beta_min) ** t)


def extract(input, t, shape):
    out = torch.gather(input, 0, t)
    reshape = [shape[0]] + [1] * (len(shape) - 1)
    out = out.reshape(*reshape)

    return out


def get_time_schedule(args, device):
    n_timestep = args.num_timesteps
    eps_small = 1e-3
    t = np.arange(0, n_timestep + 1, dtype=np.float64)
    t = t / n_timestep
    t = torch.from_numpy(t) * (1.0 - eps_small) + eps_small
    return t.to(device)


def get_sigma_schedule(args, device):
    n_timestep = args.num_timesteps
    beta_min = args.beta_min
    beta_max = args.beta_max
    eps_small = 1e-3

    t = np.arange(0, n_timestep + 1, dtype=np.float64)
    t = t / n_timestep
    t = torch.from_numpy(t) * (1.0 - eps_small) + eps_small

    if args.use_geometric:
        var = var_func_geometric(t, beta_min, beta_max)
    else:
        var = var_func_vp(t, beta_min, beta_max)
    alpha_bars = 1.0 - var
    betas = 1 - alpha_bars[1:] / alpha_bars[:-1]

    first = torch.tensor(1e-8)
    betas = torch.cat((first[None], betas)).to(device)
    betas = betas.type(torch.float32)
    sigmas = betas**0.5
    a_s = torch.sqrt(1 - betas)
    return sigmas, a_s, betas


# %% posterior sampling
class Posterior_Coefficients:
    def __init__(self, args, device):

        _, _, self.betas = get_sigma_schedule(args, device=device)

        # we don't need the zeros
        self.betas = self.betas.type(torch.float32)[1:]

        self.alphas = 1 - self.betas
        self.alphas_cumprod = torch.cumprod(self.alphas, 0)
        self.alphas_cumprod_prev = torch.cat(
            (
                torch.tensor([1.0], dtype=torch.float32, device=device),
                self.alphas_cumprod[:-1],
            ),
            0,
        )
        self.posterior_variance = self.betas * (1 - self.alphas_cumprod_prev) / (1 - self.alphas_cumprod)

        self.sqrt_alphas_cumprod = torch.sqrt(self.alphas_cumprod)
        self.sqrt_recip_alphas_cumprod = torch.rsqrt(self.alphas_cumprod)
        self.sqrt_recipm1_alphas_cumprod = torch.sqrt(1 / self.alphas_cumprod - 1)

        self.posterior_mean_coef1 = self.betas * torch.sqrt(self.alphas_cumprod_prev) / (1 - self.alphas_cumprod)
        self.posterior_mean_coef2 = (
            (1 - self.alphas_cumprod_prev) * torch.sqrt(self.alphas) / (1 - self.alphas_cumprod)
        )

        self.posterior_log_variance_clipped = torch.log(self.posterior_variance.clamp(min=1e-20))


def sample_posterior(coefficients, x_0, x_t, t):

    def q_posterior(x_0, x_t, t):
        mean = (
            extract(coefficients.posterior_mean_coef1, t, x_t.shape) * x_0
            + extract(coefficients.posterior_mean_coef2, t, x_t.shape) * x_t
        )
        var = extract(coefficients.posterior_variance, t, x_t.shape)
        log_var_clipped = extract(coefficients.posterior_log_variance_clipped, t, x_t.shape)
        return mean, var, log_var_clipped

    def p_sample(x_0, x_t, t):
        mean, _, log_var = q_posterior(x_0, x_t, t)

        noise = torch.randn_like(x_t)

        nonzero_mask = 1 - (t == 0).type(torch.float32)

        return mean + nonzero_mask[:, None, None, None] * torch.exp(0.5 * log_var) * noise

    sample_x_pos = p_sample(x_0, x_t, t)

    return sample_x_pos


def sample_from_model(coefficients, generator, n_time, x_init, T, opt):
    x = x_init
    with torch.no_grad():
        for i in reversed(range(n_time)):
            t = torch.full((x.size(0),), i, dtype=torch.int64).to(x.device)

            t_time = t
            latent_z = torch.randn(x.size(0), opt.nz, device=x.device)
            x_0 = generator(x, t_time, latent_z)
            x_new = sample_posterior(coefficients, x_0, x, t)
            x = x_new.detach()
        x
    return x


# %%
def sample_and_test(args, epoch_id):
    torch.manual_seed(42)
    device = "cuda:0"
    print(f"Epoch {epoch_id}")

    algo = "rduot"
    if args.phi1 == "none":
        algo = "ddgan"
    samples_dir = f"./generated_samples/{algo}/{args.dataset}"
    result_dir = f"./saved_info/{algo}/{args.dataset}"
    save_dir = f"./generated_samples/{algo}/{args.dataset}"

    if args.perturb_percent > 0:
        samples_dir += f"_{int(args.perturb_percent)}p_{args.perturb_dataset}"
        result_dir += f"_{int(args.perturb_percent)}p_{args.perturb_dataset}"
        save_dir += f"_{int(args.perturb_percent)}p_{args.perturb_dataset}"

    samples_folder = samples_dir + f"/{args.version}"
    result_folder = result_dir + f"/{args.version}/fid_1.12"
    save_dir += f"/{args.version}"
    result_dir += f"/{args.version}/netG_{epoch_id}.pth"

    if args.dataset == "cifar10":
        real_img_dir = "pytorch_fid/cifar10_train_stat.npy"
    elif args.dataset == "lsun":
        real_img_dir = "pytorch_fid/lsun_church_stat.npy"
    elif args.dataset == "stl10":
        real_img_dir = "pytorch_fid/stl10_stat.npy"
    elif args.dataset == "celeba_64":
        real_img_dir = "pytorch_fid/celeba_train_stat.npy"
    elif args.dataset == "celebahq_256" or args.dataset == "celeba_hq":
        real_img_dir = "pytorch_fid/celebahq_train_stat.npy"
    else:
        real_img_dir = args.real_img_dir
    print(real_img_dir)

    def to_range_0_1(x):
        return (x + 1.0) / 2.0

    if not os.path.exists(result_folder):
        os.makedirs(result_folder)

    json_filename = f"{result_folder}/stat.json"
    print(json_filename)

    netG = NCSNpp(args).to(device)
    ckpt = torch.load(result_dir, map_location=device)

    # loading weights from ddp in single gpu
    for key in list(ckpt.keys()):
        ckpt[key[7:]] = ckpt.pop(key)
    netG.load_state_dict(ckpt)
    netG.eval()

    T = get_time_schedule(args, device)

    pos_coeff = Posterior_Coefficients(args, device)

    iters_needed = 50000 // args.batch_size

    if not os.path.exists(save_dir):
        os.makedirs(save_dir)
    if not os.path.exists(samples_folder):
        os.makedirs(samples_folder)

    if args.compute_fid:
        for i in range(iters_needed):
            with torch.no_grad():
                x_t_1 = torch.randn(args.batch_size, args.num_channels, args.image_size, args.image_size).to(device)
                fake_sample = sample_from_model(pos_coeff, netG, args.num_timesteps, x_t_1, T, args)

                fake_sample = to_range_0_1(fake_sample)
                for j, x in enumerate(fake_sample):
                    index = i * args.batch_size + j
                    torchvision.utils.save_image(x, f"{samples_folder}/{index}.jpg")
                print("generating batch ", i)

        paths = [save_dir, real_img_dir]

        kwargs = {"batch_size": 100, "device": device, "dims": 2048}
        # Specify the filename for the JSON file

        fid = calculate_fid_given_paths(paths=paths, **kwargs)
        print("FID = {}".format(fid))

        data = {"fid": fid}

        if not os.path.exists(json_filename):
            data = {}
            with open(json_filename, "w") as json_file:
                json.dump(data, json_file)

        with open(json_filename, "r") as file:
            data = json.load(file)
        data[f"Epoch {epoch_id}"] = fid
        data = dict(sorted(data.items(), key=lambda x: int(str(x[0]).split()[1])))
        with open(json_filename, "w") as json_file:
            json.dump(data, json_file)
    else:
        x_t_1 = torch.randn(args.batch_size, args.num_channels, args.image_size, args.image_size).to(device)
        fake_sample = sample_from_model(pos_coeff, netG, args.num_timesteps, x_t_1, T, args)
        fake_sample = to_range_0_1(fake_sample)
        torchvision.utils.save_image(fake_sample, "./samples_{}.jpg".format(args.dataset))


if __name__ == "__main__":
    parser = argparse.ArgumentParser("ddgan parameters")
    parser.add_argument("--seed", type=int, default=1024, help="seed used for initialization")
    parser.add_argument(
        "--compute_fid",
        action="store_true",
        default=False,
        help="whether or not compute FID",
    )
    parser.add_argument("--num_channels", type=int, default=3, help="channel of image")
    parser.add_argument("--centered", action="store_false", default=True, help="-1,1 scale")
    parser.add_argument("--use_geometric", action="store_true", default=False)
    parser.add_argument("--beta_min", type=float, default=0.1, help="beta_min for diffusion")
    parser.add_argument("--beta_max", type=float, default=20.0, help="beta_max for diffusion")

    parser.add_argument(
        "--num_channels_dae",
        type=int,
        default=64,
        help="number of initial channels in denosing model",
    )
    parser.add_argument("--n_mlp", type=int, default=4, help="number of mlp layers for z")
    parser.add_argument("--ch_mult", nargs="+", type=int, help="channel multiplier")

    parser.add_argument(
        "--num_res_blocks",
        type=int,
        default=2,
        help="number of resnet blocks per scale",
    )
    parser.add_argument("--attn_resolutions", default=(16,), help="resolution of applying attention")
    parser.add_argument("--dropout", type=float, default=0.0, help="drop-out rate")
    parser.add_argument(
        "--resamp_with_conv",
        action="store_false",
        default=True,
        help="always up/down sampling with conv",
    )
    parser.add_argument("--conditional", action="store_false", default=True, help="noise conditional")
    parser.add_argument("--fir", action="store_false", default=True, help="FIR")
    parser.add_argument("--fir_kernel", default=[1, 3, 3, 1], help="FIR kernel")
    parser.add_argument("--skip_rescale", action="store_false", default=True, help="skip rescale")
    parser.add_argument(
        "--resblock_type",
        default="biggan",
        help="tyle of resnet block, choice in biggan and ddpm",
    )
    parser.add_argument(
        "--progressive",
        type=str,
        default="none",
        choices=["none", "output_skip", "residual"],
        help="progressive type for output",
    )
    parser.add_argument(
        "--progressive_input",
        type=str,
        default="residual",
        choices=["none", "input_skip", "residual"],
        help="progressive type for input",
    )
    parser.add_argument(
        "--progressive_combine",
        type=str,
        default="sum",
        choices=["sum", "cat"],
        help="progressive combine method.",
    )

    parser.add_argument(
        "--embedding_type",
        type=str,
        default="positional",
        choices=["positional", "fourier"],
        help="type of time embedding",
    )
    parser.add_argument("--fourier_scale", type=float, default=16.0, help="scale of fourier transform")
    parser.add_argument("--not_use_tanh", action="store_true", default=False)

    # geenrator and training
    parser.add_argument("--exp", default="experiment_cifar_default", help="name of experiment")
    parser.add_argument(
        "--real_img_dir",
        default="./pytorch_fid/cifar10_train_stat.npy",
        help="directory to real images for FID computation",
    )

    parser.add_argument("--dataset", default="cifar10", help="name of dataset")
    parser.add_argument("--image_size", type=int, default=32, help="size of image")

    parser.add_argument("--nz", type=int, default=100)
    parser.add_argument("--num_timesteps", type=int, default=4)

    parser.add_argument("--z_emb_dim", type=int, default=256)
    parser.add_argument("--t_emb_dim", type=int, default=256)
    parser.add_argument("--batch_size", type=int, default=100, help="sample generating batch size")

    parser.add_argument("--version", type=str, default="1", help="result version")
    parser.add_argument("--index", type=str, default="1", help="also version, but for the saved results")
    parser.add_argument("--master_port", type=str, default="6020", help="master port number")
    parser.add_argument("--perturb_dataset", default="none", help="name of data added to the dataset")
    parser.add_argument("--perturb_percent", type=float, default=0, help="percentage of perturb_data")

    # Loss configurations
    parser.add_argument("--phi1", type=str, default="softplus", choices=["linear", "kl", "softplus", "chi"])
    parser.add_argument("--phi2", type=str, default="softplus", choices=["linear", "kl", "softplus", "chi"])
    parser.add_argument("--tau", type=float, default=0.001, help="proportion of the cost c")

    parser.add_argument(
        "--schedule",
        type=int,
        default=1800,
        help="cosine scheduler, learning rate 1e-5 until {schedule} epoch",
    )

    parser.add_argument("--epoch_start", type=int, default=1000)
    parser.add_argument("--epoch_end", type=int, default=1000)
    parser.add_argument("--epoch_jump", type=int, default=25)

    args = parser.parse_args()

    for epoch_id in range(args.epoch_start, args.epoch_end + 1, args.epoch_jump):
        sample_and_test(args=args, epoch_id=epoch_id)