
'''simple docstring'''

import argparse

import torch

from diffusers.pipelines.stable_diffusion.convert_from_ckpt import download_from_original_stable_diffusion_ckpt


if __name__ == "__main__":
    __snake_case:  Any								=			argparse.ArgumentParser()

    parser.add_argument(
        "--checkpoint_path", default=None, type=str, required=True, help="Path to the checkpoint to convert."
    )
    # !wget https://raw.githubusercontent.com/CompVis/stable-diffusion/main/configs/stable-diffusion/v1-inference.yaml
    parser.add_argument(
        "--original_config_file",
        default=None,
        type=str,
        help="The YAML config file corresponding to the original architecture.",
    )
    parser.add_argument(
        "--num_in_channels",
        default=None,
        type=int,
        help="The number of input channels. If `None` number of input channels will be automatically inferred.",
    )
    parser.add_argument(
        "--scheduler_type",
        default="pndm",
        type=str,
        help="Type of scheduler to use. Should be one of ['pndm', 'lms', 'ddim', 'euler', 'euler-ancestral', 'dpm']",
    )
    parser.add_argument(
        "--pipeline_type",
        default=None,
        type=str,
        help=(
            "The pipeline type. One of 'FrozenOpenCLIPEmbedder', 'FrozenCLIPEmbedder', 'PaintByExample'"
            ". If `None` pipeline will be automatically inferred."
        ),
    )
    parser.add_argument(
        "--image_size",
        default=None,
        type=int,
        help=(
            "The image size that the model was trained on. Use 512 for Stable Diffusion v1.X and Stable Siffusion v2"
            " Base. Use 768 for Stable Diffusion v2."
        ),
    )
    parser.add_argument(
        "--prediction_type",
        default=None,
        type=str,
        help=(
            "The prediction type that the model was trained on. Use 'epsilon' for Stable Diffusion v1.X and Stable"
            " Diffusion v2 Base. Use 'v_prediction' for Stable Diffusion v2."
        ),
    )
    parser.add_argument(
        "--extract_ema",
        action="store_true",
        help=(
            "Only relevant for checkpoints that have both EMA and non-EMA weights. Whether to extract the EMA weights"
            " or not. Defaults to `False`. Add `--extract_ema` to extract the EMA weights. EMA weights usually yield"
            " higher quality images for inference. Non-EMA weights are usually better to continue fine-tuning."
        ),
    )
    parser.add_argument(
        "--upcast_attention",
        action="store_true",
        help=(
            "Whether the attention computation should always be upcasted. This is necessary when running stable"
            " diffusion 2.1."
        ),
    )
    parser.add_argument(
        "--from_safetensors",
        action="store_true",
        help="If `--checkpoint_path` is in `safetensors` format, load checkpoint with safetensors instead of PyTorch.",
    )
    parser.add_argument(
        "--to_safetensors",
        action="store_true",
        help="Whether to store pipeline in safetensors format or not.",
    )
    parser.add_argument("--dump_path", default=None, type=str, required=True, help="Path to the output model.")
    parser.add_argument("--device", type=str, help="Device to use (e.g. cpu, cuda:0, cuda:1, etc.)")
    parser.add_argument(
        "--stable_unclip",
        type=str,
        default=None,
        required=False,
        help="Set if this is a stable unCLIP model. One of 'txt2img' or 'img2img'.",
    )
    parser.add_argument(
        "--stable_unclip_prior",
        type=str,
        default=None,
        required=False,
        help="Set if this is a stable unCLIP txt2img model. Selects which prior to use. If `--stable_unclip` is set to `txt2img`, the karlo prior (https://huggingface.co/kakaobrain/karlo-v1-alpha/tree/main/prior) is selected by default.",
    )
    parser.add_argument(
        "--clip_stats_path",
        type=str,
        help="Path to the clip stats file. Only required if the stable unclip model's config specifies `model.params.noise_aug_config.params.clip_stats_path`.",
        required=False,
    )
    parser.add_argument(
        "--controlnet", action="store_true", default=None, help="Set flag if this is a controlnet checkpoint."
    )
    parser.add_argument("--half", action="store_true", help="Save weights in half precision.")
    parser.add_argument(
        "--vae_path",
        type=str,
        default=None,
        required=False,
        help="Set to a path, hub id to an already converted vae to not convert it again.",
    )
    __snake_case:  Optional[int]								=			parser.parse_args()

    __snake_case:  Any								=			download_from_original_stable_diffusion_ckpt(
        checkpoint_path=args.checkpoint_path,
        original_config_file=args.original_config_file,
        image_size=args.image_size,
        prediction_type=args.prediction_type,
        model_type=args.pipeline_type,
        extract_ema=args.extract_ema,
        scheduler_type=args.scheduler_type,
        num_in_channels=args.num_in_channels,
        upcast_attention=args.upcast_attention,
        from_safetensors=args.from_safetensors,
        device=args.device,
        stable_unclip=args.stable_unclip,
        stable_unclip_prior=args.stable_unclip_prior,
        clip_stats_path=args.clip_stats_path,
        controlnet=args.controlnet,
        vae_path=args.vae_path,
    )

    if args.half:
        pipe.to(torch_dtype=torch.floataa)

    if args.controlnet:
        # only save the controlnet model
        pipe.controlnet.save_pretrained(args.dump_path, safe_serialization=args.to_safetensors)
    else:
        pipe.save_pretrained(args.dump_path, safe_serialization=args.to_safetensors)




'''simple docstring'''

import math
from dataclasses import dataclass
from typing import Optional, Tuple, Union

import numpy as np
import torch

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput, randn_tensor
from .scheduling_utils import SchedulerMixin


@dataclass
# Copied from diffusers.schedulers.scheduling_ddpm.DDPMSchedulerOutput with DDPM->UnCLIP
class  _UpperCAmelCase ( lowerCAmelCase__    ):

     """simple docstring"""

     a_    	=					42
     a_    	=					None







def 	_snake_case						(      A_       :    int     ,		A_       :    List[str]=0.999     ,		A_       :    List[Any]="cosine"     ,		):
  """simple docstring"""







  if alpha_transform_type == "cosine":

    def alpha_bar_fn(A_       :    List[Any]   ):
      return math.cos((t + 0.008) / 1.008 * math.pi / 2   ) ** 2

  elif alpha_transform_type == "exp":

    def alpha_bar_fn(A_       :    List[str]   ):
      return math.exp(t * -12.0   )

  else:
    raise ValueError(f'''Unsupported alpha_tranform_type: {alpha_transform_type}'''   )

  a_      :	int      =							[]
  for i in range(A_   ):
    a_      :	Optional[Any]      =							i / num_diffusion_timesteps
    a_      :	List[Any]      =							(i + 1) / num_diffusion_timesteps
    betas.append(min(1 - alpha_bar_fn(A_   ) / alpha_bar_fn(A_   )     ,		A_   )   )
  return torch.tensor(A_     ,		dtype=torch.floataa   )




class  _UpperCAmelCase ( lowerCAmelCase__	,lowerCAmelCase__    ):

     """simple docstring"""

     @register_to_config
     def __init__(      self							,		lowerCAmelCase_ = 10_00							,		lowerCAmelCase_ = "fixed_small_log"							,		lowerCAmelCase_ = True							,		lowerCAmelCase_ = 1.0							,		lowerCAmelCase_ = "epsilon"							,		lowerCAmelCase_ = "squaredcos_cap_v2"							,		):

       '''simple docstring'''

       if beta_schedule != "squaredcos_cap_v2":
         raise ValueError("""UnCLIPScheduler only supports `beta_schedule`: 'squaredcos_cap_v2'""" )

       a_      :	Optional[Any]      =							betas_for_alpha_bar(lowerCAmelCase_ )

       a_      :	Optional[Any]      =							1.0 - self.betas
       a_      :	List[str]      =							torch.cumprod(self.alphas							,		dim=0 )
       a_      :	Any      =							torch.tensor(1.0 )

       # standard deviation of the initial noise distribution
       a_      :	Any      =							1.0

       # setable values
       a_      :	List[Any]      =							None
       a_      :	Any      =							torch.from_numpy(np.arange(0							,		lowerCAmelCase_ )[::-1].copy() )

       a_      :	Any      =							variance_type


     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_							,		lowerCAmelCase_ = None ):

       '''simple docstring'''

       return sample


     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_							,		lowerCAmelCase_ = None ):

       '''simple docstring'''

       a_      :	List[str]      =							num_inference_steps
       a_      :	Dict      =							(self.config.num_train_timesteps - 1) / (self.num_inference_steps - 1)
       a_      :	Optional[Any]      =							(np.arange(0							,		lowerCAmelCase_ ) * step_ratio).round()[::-1].copy().astype(np.intaa )
       a_      :	List[Any]      =							torch.from_numpy(lowerCAmelCase_ ).to(lowerCAmelCase_ )


     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_							,		lowerCAmelCase_=None							,		lowerCAmelCase_=None							,		lowerCAmelCase_=None ):

       '''simple docstring'''

       if prev_timestep is None:
         a_      :	Tuple      =							t - 1

       a_      :	Optional[Any]      =							self.alphas_cumprod[t]
       a_      :	Union[str, Any]      =							self.alphas_cumprod[prev_timestep] if prev_timestep >= 0 else self.one
       a_      :	Optional[Any]      =							1 - alpha_prod_t
       a_      :	str      =							1 - alpha_prod_t_prev

       if prev_timestep == t - 1:
         a_      :	int      =							self.betas[t]
       else:
         a_      :	Tuple      =							1 - alpha_prod_t / alpha_prod_t_prev

       # For t > 0, compute predicted variance βt (see formula (6) and (7) from https://arxiv.org/pdf/2006.11239.pdf)
       # and sample from it to get previous sample
       # x_{t-1} ~ N(pred_prev_sample, variance) == add variance to pred_sample
       a_      :	str      =							beta_prod_t_prev / beta_prod_t * beta

       if variance_type is None:
         a_      :	int      =							self.config.variance_type

       # hacks - were probably added for training stability
       if variance_type == "fixed_small_log":
         a_      :	List[Any]      =							torch.log(torch.clamp(lowerCAmelCase_							,		min=1E-2_0 ) )
         a_      :	List[Any]      =							torch.exp(0.5 * variance )
       elif variance_type == "learned_range":
         # NOTE difference with DDPM scheduler
         a_      :	Tuple      =							variance.log()
         a_      :	int      =							beta.log()

         a_      :	List[Any]      =							(predicted_variance + 1) / 2
         a_      :	Dict      =							frac * max_log + (1 - frac) * min_log

       return variance


     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_							,		lowerCAmelCase_							,		lowerCAmelCase_							,		lowerCAmelCase_ = None							,		lowerCAmelCase_=None							,		lowerCAmelCase_ = True							,		):

       '''simple docstring'''

       a_      :	Tuple      =							timestep

       if model_output.shape[1] == sample.shape[1] * 2 and self.variance_type == "learned_range":
         a_      , a_      :	Optional[Any]      =							torch.split(lowerCAmelCase_							,		sample.shape[1]							,		dim=1 )
       else:
         a_      :	Optional[int]      =							None

       # 1. compute alphas, betas
       if prev_timestep is None:
         a_      :	List[Any]      =							t - 1

       a_      :	int      =							self.alphas_cumprod[t]
       a_      :	List[str]      =							self.alphas_cumprod[prev_timestep] if prev_timestep >= 0 else self.one
       a_      :	Any      =							1 - alpha_prod_t
       a_      :	Any      =							1 - alpha_prod_t_prev

       if prev_timestep == t - 1:
         a_      :	Optional[int]      =							self.betas[t]
         a_      :	Any      =							self.alphas[t]
       else:
         a_      :	List[Any]      =							1 - alpha_prod_t / alpha_prod_t_prev
         a_      :	List[str]      =							1 - beta

       # 2. compute predicted original sample from predicted noise also called
       # "predicted x_0" of formula (15) from https://arxiv.org/pdf/2006.11239.pdf
       if self.config.prediction_type == "epsilon":
         a_      :	List[str]      =							(sample - beta_prod_t ** 0.5 * model_output) / alpha_prod_t ** 0.5
       elif self.config.prediction_type == "sample":
         a_      :	Optional[int]      =							model_output
       else:
         raise ValueError(
             f'''prediction_type given as {self.config.prediction_type} must be one of `epsilon` or `sample`'''
             """ for the UnCLIPScheduler.""" )

       # 3. Clip "predicted x_0"
       if self.config.clip_sample:
         a_      :	Any      =							torch.clamp(
             lowerCAmelCase_							,		-self.config.clip_sample_range							,		self.config.clip_sample_range )

       # 4. Compute coefficients for pred_original_sample x_0 and current sample x_t
       # See formula (7) from https://arxiv.org/pdf/2006.11239.pdf
       a_      :	str      =							(alpha_prod_t_prev ** 0.5 * beta) / beta_prod_t
       a_      :	Dict      =							alpha ** 0.5 * beta_prod_t_prev / beta_prod_t

       # 5. Compute predicted previous sample µ_t
       # See formula (7) from https://arxiv.org/pdf/2006.11239.pdf
       a_      :	Any      =							pred_original_sample_coeff * pred_original_sample + current_sample_coeff * sample

       # 6. Add noise
       a_      :	List[Any]      =							0
       if t > 0:
         a_      :	List[Any]      =							randn_tensor(
             model_output.shape							,		dtype=model_output.dtype							,		generator=lowerCAmelCase_							,		device=model_output.device )

         a_      :	List[Any]      =							self._get_variance(
             lowerCAmelCase_							,		predicted_variance=lowerCAmelCase_							,		prev_timestep=lowerCAmelCase_							,		)

         if self.variance_type == "fixed_small_log":
           a_      :	Dict      =							variance
         elif self.variance_type == "learned_range":
           a_      :	Any      =							(0.5 * variance).exp()
         else:
           raise ValueError(
               f'''variance_type given as {self.variance_type} must be one of `fixed_small_log` or `learned_range`'''
               """ for the UnCLIPScheduler.""" )

         a_      :	List[str]      =							variance * variance_noise

       a_      :	Optional[Any]      =							pred_prev_sample + variance

       if not return_dict:
         return (pred_prev_sample,)

       return UnCLIPSchedulerOutput(prev_sample=lowerCAmelCase_							,		pred_original_sample=lowerCAmelCase_ )






     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_							,		lowerCAmelCase_							,		lowerCAmelCase_							,		):

       '''simple docstring'''

       a_      :	List[str]      =							self.alphas_cumprod.to(device=original_samples.device							,		dtype=original_samples.dtype )
       a_      :	Union[str, Any]      =							timesteps.to(original_samples.device )

       a_      :	List[Any]      =							alphas_cumprod[timesteps] ** 0.5
       a_      :	Optional[Any]      =							sqrt_alpha_prod.flatten()
       while len(sqrt_alpha_prod.shape ) < len(original_samples.shape ):
         a_      :	int      =							sqrt_alpha_prod.unsqueeze(-1 )

       a_      :	Optional[Any]      =							(1 - alphas_cumprod[timesteps]) ** 0.5
       a_      :	str      =							sqrt_one_minus_alpha_prod.flatten()
       while len(sqrt_one_minus_alpha_prod.shape ) < len(original_samples.shape ):
         a_      :	str      =							sqrt_one_minus_alpha_prod.unsqueeze(-1 )

       a_      :	int      =							sqrt_alpha_prod * original_samples + sqrt_one_minus_alpha_prod * noise
       return noisy_samples


