guided_diffusion/script_util.py

import argparse
import inspect

from pdb import set_trace as st

from cldm.cldm import ControlledUnetModel, ControlNet

from . import gaussian_diffusion as gd
from .respace import SpacedDiffusion, space_timesteps
# from .unet_old import SuperResModel, UNetModel, EncoderUNetModel # , UNetModelWithHint
from .unet import SuperResModel, UNetModel, EncoderUNetModel # , UNetModelWithHint
import torch as th
from dit.dit_models_xformers import DiT_models
if th.cuda.is_available():
    from xformers.triton import FusedLayerNorm as LayerNorm

NUM_CLASSES = 1000


def diffusion_defaults():
    """
    Defaults for image and classifier training.
    """
    return dict(
        learn_sigma=False,
        diffusion_steps=1000,
        noise_schedule="linear",
        standarization_xt=False,
        timestep_respacing="",
        use_kl=False,
        predict_xstart=False,
        predict_v=False,
        rescale_timesteps=False,
        rescale_learned_sigmas=False,
        mixed_prediction=False,  # ! to assign later
    )


def classifier_defaults():
    """
    Defaults for classifier models.
    """
    return dict(
        image_size=64,
        classifier_use_fp16=False,
        classifier_width=128,
        classifier_depth=2,
        classifier_attention_resolutions="32,16,8",  # 16
        classifier_use_scale_shift_norm=True,  # False
        classifier_resblock_updown=True,  # False
        classifier_pool="attention",
    )


def control_net_defaults():
    res = dict(
        only_mid_control=False,  # TODO
        control_key='img',
        normalize_clip_encoding=False,  # zero-shot text inference
        scale_clip_encoding=1.0,
        cfg_dropout_prob=0.0,  # dropout condition for CFG training
        # cond_key='caption',
    )
    return res


def continuous_diffusion_defaults():
    # NVlabs/LSGM/train_vada.py
    res = dict(
        sde_time_eps=1e-2,
        sde_beta_start=0.1,
        sde_beta_end=20.0,
        sde_sde_type='vpsde',
        sde_sigma2_0=0.0,  # ?
        iw_sample_p='drop_sigma2t_iw',
        iw_sample_q='ll_iw',
        iw_subvp_like_vp_sde=False,
        train_vae=True,
        pred_type='eps',  # [x0, eps]
        # joint_train=False,
        p_rendering_loss=False,
        unfix_logit=False,
        loss_type='eps',
        loss_weight='simple',  # snr snr_sqrt sigmoid_snr
        # train_vae_denoise_rendering=False,
        diffusion_ce_anneal=True,
        enable_mixing_normal=True,
    )

    return res


def model_and_diffusion_defaults():
    """
    Defaults for image training.
    """
    res = dict(
        # image_size=64,
        diffusion_input_size=224,
        num_channels=128,
        num_res_blocks=2,
        num_heads=4,
        num_heads_upsample=-1,
        num_head_channels=-1,
        attention_resolutions="16,8",
        channel_mult="",
        dropout=0.0,
        class_cond=False,
        use_checkpoint=False,
        use_scale_shift_norm=True,
        resblock_updown=False,
        use_fp16=False,
        use_new_attention_order=False,
        denoise_in_channels=3,
        denoise_out_channels=3,
        # ! controlnet args
        create_controlnet=False,
        create_dit=False,
        create_unet_with_hint=False,
        dit_model_arch='DiT-L/2',
        # ! ldm unet support
        use_spatial_transformer=False,  # custom transformer support
        transformer_depth=1,  # custom transformer support
        context_dim=-1,  # custom transformer support
        roll_out=False,  # whether concat in batch, not channel
        n_embed=
        None,  # custom support for prediction of discrete ids into codebook of first stage vq model
        legacy=True,
        mixing_logit_init=-6,
        hint_channels=3,
        # unconditional_guidance_scale=1.0,
        # normalize_clip_encoding=False, # for zero-shot conditioning
    )
    res.update(diffusion_defaults())
    # res.update(continuous_diffusion_defaults())
    return res


def classifier_and_diffusion_defaults():
    res = classifier_defaults()
    res.update(diffusion_defaults())
    return res


def create_model_and_diffusion(
    # image_size,
    diffusion_input_size,
    class_cond,
    learn_sigma,
    num_channels,
    num_res_blocks,
    channel_mult,
    num_heads,
    num_head_channels,
    num_heads_upsample,
    attention_resolutions,
    dropout,
    diffusion_steps,
    noise_schedule,
    timestep_respacing,
    use_kl,
    predict_xstart,
    predict_v,
    rescale_timesteps,
    rescale_learned_sigmas,
    use_checkpoint,
    use_scale_shift_norm,
    resblock_updown,
    use_fp16,
    use_new_attention_order,
    denoise_in_channels,
    denoise_out_channels,
    standarization_xt,
    mixed_prediction,
    # controlnet
    create_controlnet,
    # only_mid_control,
    # control_key,
    use_spatial_transformer,
    transformer_depth,
    context_dim,
    n_embed,
    legacy,
    mixing_logit_init,
    create_dit,
    create_unet_with_hint,
    dit_model_arch,
    roll_out,
    hint_channels,
    # unconditional_guidance_scale,
    # normalize_clip_encoding,
):
    model = create_model(
        diffusion_input_size,
        num_channels,
        num_res_blocks,
        channel_mult=channel_mult,
        learn_sigma=learn_sigma,
        class_cond=class_cond,
        use_checkpoint=use_checkpoint,
        attention_resolutions=attention_resolutions,
        num_heads=num_heads,
        num_head_channels=num_head_channels,
        num_heads_upsample=num_heads_upsample,
        use_scale_shift_norm=use_scale_shift_norm,
        dropout=dropout,
        resblock_updown=resblock_updown,
        use_fp16=use_fp16,
        use_new_attention_order=use_new_attention_order,
        denoise_in_channels=denoise_in_channels,
        denoise_out_channels=denoise_out_channels,
        mixed_prediction=mixed_prediction,
        create_controlnet=create_controlnet,
        # only_mid_control=only_mid_control,
        # control_key=control_key,
        use_spatial_transformer=use_spatial_transformer,
        transformer_depth=transformer_depth,
        context_dim=context_dim,
        n_embed=n_embed,
        legacy=legacy,
        mixing_logit_init=mixing_logit_init,
        create_dit=create_dit,
        create_unet_with_hint=create_unet_with_hint,
        dit_model_arch=dit_model_arch,
        roll_out=roll_out,
        hint_channels=hint_channels,
        # normalize_clip_encoding=normalize_clip_encoding,
    )
    diffusion = create_gaussian_diffusion(
        diffusion_steps=diffusion_steps,
        learn_sigma=learn_sigma,
        noise_schedule=noise_schedule,
        use_kl=use_kl,
        predict_xstart=predict_xstart,
        predict_v=predict_v,
        rescale_timesteps=rescale_timesteps,
        rescale_learned_sigmas=rescale_learned_sigmas,
        timestep_respacing=timestep_respacing,
        standarization_xt=standarization_xt,
    )
    return model, diffusion


def create_model(
    image_size,
    num_channels,
    num_res_blocks,
    channel_mult="",
    learn_sigma=False,
    class_cond=False,
    use_checkpoint=False,
    attention_resolutions="16",
    num_heads=1,
    num_head_channels=-1,
    num_heads_upsample=-1,
    use_scale_shift_norm=False,
    dropout=0,
    resblock_updown=False,
    use_fp16=False,
    use_new_attention_order=False,
    # denoise_in_channels=3,
    denoise_in_channels=-1,
    denoise_out_channels=3,
    mixed_prediction=False,
    create_controlnet=False,
    create_dit=False,
    create_unet_with_hint=False,
    dit_model_arch='DiT-L/2',
    hint_channels=3,
    use_spatial_transformer=False,  # custom transformer support
    transformer_depth=1,  # custom transformer support
    context_dim=None,  # custom transformer support
    n_embed=None,  # custom support for prediction of discrete ids into codebook of first stage vq model
    legacy=True,
    mixing_logit_init=-6,
    roll_out=False,
    # normalize_clip_encoding=False,
):
    if channel_mult == "":
        if image_size == 512:
            channel_mult = (0.5, 1, 1, 2, 2, 4, 4)
        elif image_size == 448:
            channel_mult = (0.5, 1, 1, 2, 2, 4, 4)
        elif image_size == 320:  # ffhq
            channel_mult = (0.5, 1, 1, 2, 2, 4, 4)
        elif image_size == 224 and denoise_in_channels == 144:  # ffhq
            channel_mult = (1, 1, 2, 3, 4, 4)
        elif image_size == 224:
            channel_mult = (1, 1, 2, 2, 4, 4)
        elif image_size == 256:
            channel_mult = (1, 1, 2, 2, 4, 4)
        elif image_size == 128:
            channel_mult = (1, 1, 2, 3, 4)
        elif image_size == 64:
            channel_mult = (1, 2, 3, 4)

        elif image_size == 32:  # https://github.com/CompVis/latent-diffusion/blob/a506df5756472e2ebaf9078affdde2c4f1502cd4/configs/latent-diffusion/lsun_churches-ldm-kl-8.yaml#L37
            channel_mult = (1, 2, 4, 4)

        elif image_size == 16:  # B,12,16,16. just for baseline check. not good performance.
            channel_mult = (1, 2, 3, 4)
        else:
            raise ValueError(f"unsupported image size: {image_size}")
    else:
        channel_mult = tuple(
            int(ch_mult) for ch_mult in channel_mult.split(","))

    attention_ds = []
    for res in attention_resolutions.split(","):
        attention_ds.append(image_size // int(res))

    if create_controlnet:

        controlledUnetModel = ControlledUnetModel(
            image_size=image_size,
            in_channels=denoise_in_channels,
            model_channels=num_channels,
            # out_channels=(3 if not learn_sigma else 6),
            out_channels=(denoise_out_channels
                          if not learn_sigma else denoise_out_channels * 2),
            num_res_blocks=num_res_blocks,
            attention_resolutions=tuple(attention_ds),
            dropout=dropout,
            channel_mult=channel_mult,
            num_classes=(NUM_CLASSES if class_cond else None),
            use_checkpoint=use_checkpoint,
            use_fp16=use_fp16,
            num_heads=num_heads,
            num_head_channels=num_head_channels,
            num_heads_upsample=num_heads_upsample,
            use_scale_shift_norm=use_scale_shift_norm,
            resblock_updown=resblock_updown,
            use_new_attention_order=use_new_attention_order,
            mixed_prediction=mixed_prediction,
            # ldm support
            use_spatial_transformer=use_spatial_transformer,
            transformer_depth=transformer_depth,
            context_dim=context_dim,
            n_embed=n_embed,
            legacy=legacy,
            mixing_logit_init=mixing_logit_init,
            roll_out=roll_out
            )

        controlNet = ControlNet(
            image_size=image_size,
            in_channels=denoise_in_channels,
            model_channels=num_channels,
            # ! condition channels
            hint_channels=hint_channels,
            # out_channels=(3 if not learn_sigma else 6),
            # out_channels=(denoise_out_channels
            #             if not learn_sigma else denoise_out_channels * 2),
            num_res_blocks=num_res_blocks,
            attention_resolutions=tuple(attention_ds),
            dropout=dropout,
            channel_mult=channel_mult,
            # num_classes=(NUM_CLASSES if class_cond else None),
            use_checkpoint=use_checkpoint,
            use_fp16=use_fp16,
            num_heads=num_heads,
            num_head_channels=num_head_channels,
            num_heads_upsample=num_heads_upsample,
            use_scale_shift_norm=use_scale_shift_norm,
            resblock_updown=resblock_updown,
            use_new_attention_order=use_new_attention_order,
            roll_out=roll_out
        )
        # mixed_prediction=mixed_prediction)

        return controlledUnetModel, controlNet

    elif create_dit:
        return DiT_models[dit_model_arch](
            input_size=image_size,
            num_classes=0,
            learn_sigma=learn_sigma,
            in_channels=denoise_in_channels,
            context_dim=context_dim,  # add CLIP text embedding
            roll_out=roll_out)
    else:

        # if create_unet_with_hint:
        #     unet_cls = UNetModelWithHint
        # else:
        unet_cls = UNetModel

        # st()
        return unet_cls(
            image_size=image_size,
            in_channels=denoise_in_channels,
            model_channels=num_channels,
            # out_channels=(3 if not learn_sigma else 6),
            out_channels=(denoise_out_channels
                          if not learn_sigma else denoise_out_channels * 2),
            num_res_blocks=num_res_blocks,
            attention_resolutions=tuple(attention_ds),
            dropout=dropout,
            channel_mult=channel_mult,
            num_classes=(NUM_CLASSES if class_cond else None),
            use_checkpoint=use_checkpoint,
            use_fp16=use_fp16,
            num_heads=num_heads,
            num_head_channels=num_head_channels,
            num_heads_upsample=num_heads_upsample,
            use_scale_shift_norm=use_scale_shift_norm,
            resblock_updown=resblock_updown,
            use_new_attention_order=use_new_attention_order,
            mixed_prediction=mixed_prediction,
            # ldm support
            use_spatial_transformer=use_spatial_transformer,
            transformer_depth=transformer_depth,
            context_dim=context_dim,
            n_embed=n_embed,
            legacy=legacy,
            mixing_logit_init=mixing_logit_init,
            roll_out=roll_out,
            hint_channels=hint_channels,
            # normalize_clip_encoding=normalize_clip_encoding,
        )


def create_classifier_and_diffusion(
    image_size,
    classifier_use_fp16,
    classifier_width,
    classifier_depth,
    classifier_attention_resolutions,
    classifier_use_scale_shift_norm,
    classifier_resblock_updown,
    classifier_pool,
    learn_sigma,
    diffusion_steps,
    noise_schedule,
    timestep_respacing,
    use_kl,
    predict_xstart,
    rescale_timesteps,
    rescale_learned_sigmas,
):
    classifier = create_classifier(
        image_size,
        classifier_use_fp16,
        classifier_width,
        classifier_depth,
        classifier_attention_resolutions,
        classifier_use_scale_shift_norm,
        classifier_resblock_updown,
        classifier_pool,
    )
    diffusion = create_gaussian_diffusion(
        steps=diffusion_steps,
        learn_sigma=learn_sigma,
        noise_schedule=noise_schedule,
        use_kl=use_kl,
        predict_xstart=predict_xstart,
        rescale_timesteps=rescale_timesteps,
        rescale_learned_sigmas=rescale_learned_sigmas,
        timestep_respacing=timestep_respacing,
    )
    return classifier, diffusion


def create_classifier(
    image_size,
    classifier_use_fp16,
    classifier_width,
    classifier_depth,
    classifier_attention_resolutions,
    classifier_use_scale_shift_norm,
    classifier_resblock_updown,
    classifier_pool,
):
    if image_size == 512:
        channel_mult = (0.5, 1, 1, 2, 2, 4, 4)
    elif image_size == 256:
        channel_mult = (1, 1, 2, 2, 4, 4)
    elif image_size == 128:
        channel_mult = (1, 1, 2, 3, 4)
    elif image_size == 64:
        channel_mult = (1, 2, 3, 4)
    else:
        raise ValueError(f"unsupported image size: {image_size}")

    attention_ds = []
    for res in classifier_attention_resolutions.split(","):
        attention_ds.append(image_size // int(res))

    return EncoderUNetModel(
        image_size=image_size,
        in_channels=3,
        model_channels=classifier_width,
        out_channels=1000,
        num_res_blocks=classifier_depth,
        attention_resolutions=tuple(attention_ds),
        channel_mult=channel_mult,
        use_fp16=classifier_use_fp16,
        num_head_channels=64,
        use_scale_shift_norm=classifier_use_scale_shift_norm,
        resblock_updown=classifier_resblock_updown,
        pool=classifier_pool,
    )


def sr_model_and_diffusion_defaults():
    res = model_and_diffusion_defaults()
    res["large_size"] = 256
    res["small_size"] = 64
    arg_names = inspect.getfullargspec(sr_create_model_and_diffusion)[0]
    for k in res.copy().keys():
        if k not in arg_names:
            del res[k]
    return res


def sr_create_model_and_diffusion(
    large_size,
    small_size,
    class_cond,
    learn_sigma,
    num_channels,
    num_res_blocks,
    num_heads,
    num_head_channels,
    num_heads_upsample,
    attention_resolutions,
    dropout,
    diffusion_steps,
    noise_schedule,
    timestep_respacing,
    use_kl,
    predict_xstart,
    rescale_timesteps,
    rescale_learned_sigmas,
    use_checkpoint,
    use_scale_shift_norm,
    resblock_updown,
    use_fp16,
):
    model = sr_create_model(
        large_size,
        small_size,
        num_channels,
        num_res_blocks,
        learn_sigma=learn_sigma,
        class_cond=class_cond,
        use_checkpoint=use_checkpoint,
        attention_resolutions=attention_resolutions,
        num_heads=num_heads,
        num_head_channels=num_head_channels,
        num_heads_upsample=num_heads_upsample,
        use_scale_shift_norm=use_scale_shift_norm,
        dropout=dropout,
        resblock_updown=resblock_updown,
        use_fp16=use_fp16,
    )
    diffusion = create_gaussian_diffusion(
        steps=diffusion_steps,
        learn_sigma=learn_sigma,
        noise_schedule=noise_schedule,
        use_kl=use_kl,
        predict_xstart=predict_xstart,
        rescale_timesteps=rescale_timesteps,
        rescale_learned_sigmas=rescale_learned_sigmas,
        timestep_respacing=timestep_respacing,
    )
    return model, diffusion


def sr_create_model(
    large_size,
    small_size,
    num_channels,
    num_res_blocks,
    learn_sigma,
    class_cond,
    use_checkpoint,
    attention_resolutions,
    num_heads,
    num_head_channels,
    num_heads_upsample,
    use_scale_shift_norm,
    dropout,
    resblock_updown,
    use_fp16,
):
    _ = small_size  # hack to prevent unused variable

    if large_size == 512:
        channel_mult = (1, 1, 2, 2, 4, 4)
    elif large_size == 256:
        channel_mult = (1, 1, 2, 2, 4, 4)
    elif large_size == 64:
        channel_mult = (1, 2, 3, 4)
    else:
        raise ValueError(f"unsupported large size: {large_size}")

    attention_ds = []
    for res in attention_resolutions.split(","):
        attention_ds.append(large_size // int(res))

    return SuperResModel(
        image_size=large_size,
        in_channels=3,
        model_channels=num_channels,
        out_channels=(3 if not learn_sigma else 6),
        num_res_blocks=num_res_blocks,
        attention_resolutions=tuple(attention_ds),
        dropout=dropout,
        channel_mult=channel_mult,
        num_classes=(NUM_CLASSES if class_cond else None),
        use_checkpoint=use_checkpoint,
        num_heads=num_heads,
        num_head_channels=num_head_channels,
        num_heads_upsample=num_heads_upsample,
        use_scale_shift_norm=use_scale_shift_norm,
        resblock_updown=resblock_updown,
        use_fp16=use_fp16,
    )


def create_gaussian_diffusion(
    *,
    diffusion_steps=1000,
    learn_sigma=False,
    sigma_small=False,
    noise_schedule="linear",
    use_kl=False,
    predict_xstart=False,
    predict_v=False,
    rescale_timesteps=False,
    rescale_learned_sigmas=False,
    timestep_respacing="",
    standarization_xt=False,
):
    betas = gd.get_named_beta_schedule(noise_schedule, diffusion_steps)
    if use_kl:
        loss_type = gd.LossType.RESCALED_KL
    elif rescale_learned_sigmas:
        loss_type = gd.LossType.RESCALED_MSE
    else:
        loss_type = gd.LossType.MSE  # * used here.
    if not timestep_respacing:
        timestep_respacing = [diffusion_steps]

    if predict_xstart:
        model_mean_type = gd.ModelMeanType.START_X
    elif predict_v:
        model_mean_type = gd.ModelMeanType.V
    else:
        model_mean_type = gd.ModelMeanType.EPSILON

        # model_mean_type=(
        #     gd.ModelMeanType.EPSILON if not predict_xstart else
        #     gd.ModelMeanType.START_X  # * used gd.ModelMeanType.EPSILON
        # ),

    return SpacedDiffusion(
        use_timesteps=space_timesteps(diffusion_steps, timestep_respacing),
        betas=betas,
        model_mean_type=model_mean_type,
        # (
        #     gd.ModelMeanType.EPSILON if not predict_xstart else
        #     gd.ModelMeanType.START_X  # * used gd.ModelMeanType.EPSILON
        # ),
        model_var_type=((
            gd.ModelVarType.FIXED_LARGE  # * used here
            if not sigma_small else gd.ModelVarType.FIXED_SMALL)
                        if not learn_sigma else gd.ModelVarType.LEARNED_RANGE),
        loss_type=loss_type,
        rescale_timesteps=rescale_timesteps,
        standarization_xt=standarization_xt,
    )


def add_dict_to_argparser(parser, default_dict):
    for k, v in default_dict.items():
        v_type = type(v)
        if v is None:
            v_type = str
        elif isinstance(v, bool):
            v_type = str2bool
        parser.add_argument(f"--{k}", default=v, type=v_type)


def args_to_dict(args, keys):
    return {k: getattr(args, k) for k in keys}


def str2bool(v):
    """
    https://stackoverflow.com/questions/15008758/parsing-boolean-values-with-argparse
    """
    if isinstance(v, bool):
        return v
    if v.lower() in ("yes", "true", "t", "y", "1"):
        return True
    elif v.lower() in ("no", "false", "f", "n", "0"):
        return False
    else:
        raise argparse.ArgumentTypeError("boolean value expected")