import gc
import unittest

import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTextModelWithProjection, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DDPMScheduler,
    PriorTransformer,
    StableUnCLIPPipeline,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.stable_unclip_image_normalizer import StableUnCLIPImageNormalizer
from diffusers.utils.testing_utils import enable_full_determinism, load_numpy, require_torch_gpu, slow, torch_device

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import (
    PipelineKarrasSchedulerTesterMixin,
    PipelineLatentTesterMixin,
    PipelineTesterMixin,
    assert_mean_pixel_difference,
)


enable_full_determinism()




class 	__snake_case	( __lowerCAmelCase ,  __lowerCAmelCase ,  __lowerCAmelCase ,  unittest.TestCase		):





      '''simple docstring'''






      _snake_case    		=       StableUnCLIPPipeline
      _snake_case    		=       TEXT_TO_IMAGE_PARAMS
      _snake_case    		=       TEXT_TO_IMAGE_BATCH_PARAMS
      _snake_case    		=       TEXT_TO_IMAGE_IMAGE_PARAMS
      _snake_case    		=       TEXT_TO_IMAGE_IMAGE_PARAMS

      # TODO(will) Expected attn_bias.stride(1) == 0 to be true, but got false
      _snake_case    		=       False



      def 	_SCREAMING_SNAKE_CASE    (					self     :				str)				->int:


          """simple docstring"""


          _lowerCamelCase     :		List[str] 						=						32
          _lowerCamelCase     :		Dict 						=						embedder_hidden_size

          # prior components

          torch.manual_seed(0)
          _lowerCamelCase     :		Optional[Any] 						=						CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip""")

          torch.manual_seed(0)
          _lowerCamelCase     :		List[str] 						=						CLIPTextModelWithProjection(
              CLIPTextConfig(
                  bos_token_id=0    ,  eos_token_id=2    ,  hidden_size=__A    ,  projection_dim=__A    ,  intermediate_size=37    ,  layer_norm_eps=1E-0_5    ,  num_attention_heads=4    ,  num_hidden_layers=5    ,  pad_token_id=1    ,  vocab_size=1000    ,  ))

          torch.manual_seed(0)
          _lowerCamelCase     :		int 						=						PriorTransformer(
              num_attention_heads=2    ,  attention_head_dim=12    ,  embedding_dim=__A    ,  num_layers=1    ,  )

          torch.manual_seed(0)
          _lowerCamelCase     :		List[Any] 						=						DDPMScheduler(
              variance_type="""fixed_small_log"""    ,  prediction_type="""sample"""    ,  num_train_timesteps=1000    ,  clip_sample=__A    ,  clip_sample_range=5.0    ,  beta_schedule="""squaredcos_cap_v2"""    ,  )

          # regular denoising components

          torch.manual_seed(0)
          _lowerCamelCase     :		Union[str, Any] 						=						StableUnCLIPImageNormalizer(embedding_dim=__A)
          _lowerCamelCase     :		str 						=						DDPMScheduler(beta_schedule="""squaredcos_cap_v2""")

          torch.manual_seed(0)
          _lowerCamelCase     :		Any 						=						CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip""")

          torch.manual_seed(0)
          _lowerCamelCase     :		Any 						=						CLIPTextModel(
              CLIPTextConfig(
                  bos_token_id=0    ,  eos_token_id=2    ,  hidden_size=__A    ,  projection_dim=32    ,  intermediate_size=37    ,  layer_norm_eps=1E-0_5    ,  num_attention_heads=4    ,  num_hidden_layers=5    ,  pad_token_id=1    ,  vocab_size=1000    ,  ))

          torch.manual_seed(0)
          _lowerCamelCase     :		List[Any] 						=						UNetaDConditionModel(
              sample_size=32    ,  in_channels=4    ,  out_channels=4    ,  down_block_types=("""CrossAttnDownBlock2D""", """DownBlock2D""")    ,  up_block_types=("""UpBlock2D""", """CrossAttnUpBlock2D""")    ,  block_out_channels=(32, 64)    ,  attention_head_dim=(2, 4)    ,  class_embed_type="""projection"""    ,  projection_class_embeddings_input_dim=embedder_projection_dim * 2    ,  cross_attention_dim=__A    ,  layers_per_block=1    ,  upcast_attention=__A    ,  use_linear_projection=__A    ,  )

          torch.manual_seed(0)
          _lowerCamelCase     :		List[Any] 						=						DDIMScheduler(
              beta_schedule="""scaled_linear"""    ,  beta_start=0.0_0_0_8_5    ,  beta_end=0.0_1_2    ,  prediction_type="""v_prediction"""    ,  set_alpha_to_one=__A    ,  steps_offset=1    ,  )

          torch.manual_seed(0)
          _lowerCamelCase     :		Union[str, Any] 						=						AutoencoderKL()

          _lowerCamelCase     :		str 						=						{
              # prior components
              "prior_tokenizer": prior_tokenizer,
              "prior_text_encoder": prior_text_encoder,
              "prior": prior,
              "prior_scheduler": prior_scheduler,
              # image noising components
              "image_normalizer": image_normalizer,
              "image_noising_scheduler": image_noising_scheduler,
              # regular denoising components
              "tokenizer": tokenizer,
              "text_encoder": text_encoder,
              "unet": unet,
              "scheduler": scheduler,
              "vae": vae,
          }

          return components



      def 	_SCREAMING_SNAKE_CASE    (					self     :				List[str]    ,  _UpperCamelCase     :				Any    ,  _UpperCamelCase     :				List[str]=0)				->Tuple:


          """simple docstring"""


          if str(__A).startswith("""mps"""):
              _lowerCamelCase     :		List[str] 						=						torch.manual_seed(__A)
          else:
              _lowerCamelCase     :		List[Any] 						=						torch.Generator(device=__A).manual_seed(__A)
          _lowerCamelCase     :		List[str] 						=						{
              "prompt": "A painting of a squirrel eating a burger",
              "generator": generator,
              "num_inference_steps": 2,
              "prior_num_inference_steps": 2,
              "output_type": "numpy",
          }
          return inputs



      def 	_SCREAMING_SNAKE_CASE    (					self     :				Any)				->str:


          """simple docstring"""


          _lowerCamelCase     :		int 						=						torch_device == "cpu"

          self._test_attention_slicing_forward_pass(test_max_difference=__A)




      def 	_SCREAMING_SNAKE_CASE    (					self     :				int)				->Optional[int]:


          """simple docstring"""


          _lowerCamelCase     :		int 						=						torch_device in ["cpu", "mps"]

          self._test_inference_batch_single_identical(test_max_difference=__A)





@slow
@require_torch_gpu
class 	__snake_case	( unittest.TestCase		):





      '''simple docstring'''






      def 	_SCREAMING_SNAKE_CASE    (					self     :				Any)				->Union[str, Any]:


          """simple docstring"""


          super().tearDown()
          gc.collect()
          torch.cuda.empty_cache()



      def 	_SCREAMING_SNAKE_CASE    (					self     :				str)				->Tuple:


          """simple docstring"""


          _lowerCamelCase     :		Union[str, Any] 						=						load_numpy(
              """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_l_anime_turtle_fp16.npy""")

          _lowerCamelCase     :		List[Any] 						=						StableUnCLIPPipeline.from_pretrained("""fusing/stable-unclip-2-1-l"""    ,  torch_dtype=torch.floataa)
          pipe.to(__A)
          pipe.set_progress_bar_config(disable=__A)
          # stable unclip will oom when integration tests are run on a V100,
          # so turn on memory savings
          pipe.enable_attention_slicing()
          pipe.enable_sequential_cpu_offload()

          _lowerCamelCase     :		int 						=						torch.Generator(device="""cpu""").manual_seed(0)
          _lowerCamelCase     :		Tuple 						=						pipe("""anime turle"""    ,  generator=__A    ,  output_type="""np""")

          _lowerCamelCase     :		Optional[Any] 						=						output.images[0]

          assert image.shape == (768, 768, 3)

          assert_mean_pixel_difference(__A    ,  __A)




      def 	_SCREAMING_SNAKE_CASE    (					self     :				Optional[int])				->Any:


          """simple docstring"""


          torch.cuda.empty_cache()
          torch.cuda.reset_max_memory_allocated()
          torch.cuda.reset_peak_memory_stats()

          _lowerCamelCase     :		Dict 						=						StableUnCLIPPipeline.from_pretrained("""fusing/stable-unclip-2-1-l"""    ,  torch_dtype=torch.floataa)
          _lowerCamelCase     :		Optional[Any] 						=						pipe.to(__A)
          pipe.set_progress_bar_config(disable=__A)
          pipe.enable_attention_slicing()
          pipe.enable_sequential_cpu_offload()

          _lowerCamelCase     :		List[str] 						=						pipe(
              """anime turtle"""    ,  prior_num_inference_steps=2    ,  num_inference_steps=2    ,  output_type="""np"""    ,  )

          _lowerCamelCase     :		Optional[int] 						=						torch.cuda.max_memory_allocated()
          # make sure that less than 7 GB is allocated
          assert mem_bytes < 7 * 10**9





import argparse
from pathlib import Path

import torch
from packaging import version
from torch.onnx import export

from diffusers import AutoencoderKL


lowerCAmelCase			:							Tuple     =version.parse(version.parse(torch.__version__).base_version) < version.parse("1.11")







def 						A__  (					__A   , __A   , __A   , __A   , __A   , __A   , __A   , __A=False   , ):







    '''simple docstring'''
    output_path.parent.mkdir(parents=__A   , exist_ok=__A	)
    # PyTorch deprecated the `enable_onnx_checker` and `use_external_data_format` arguments in v1.11,
    # so we check the torch version for backwards compatibility
    if is_torch_less_than_1_11:
        export(
            __A   , __A   , f=output_path.as_posix()   , input_names=__A   , output_names=__A   , dynamic_axes=__A   , do_constant_folding=__A   , use_external_data_format=__A   , enable_onnx_checker=__A   , opset_version=__A   , )
    else:
        export(
            __A   , __A   , f=output_path.as_posix()   , input_names=__A   , output_names=__A   , dynamic_axes=__A   , do_constant_folding=__A   , opset_version=__A   , )







@torch.no_grad()
def 						A__  (					__A   , __A   , __A   , __A = False	):







    '''simple docstring'''
    _lowerCamelCase     :		Tuple 						=						torch.floataa if fpaa else torch.floataa
    if fpaa and torch.cuda.is_available():
        _lowerCamelCase     :		str 						=						"""cuda"""
    elif fpaa and not torch.cuda.is_available():
        raise ValueError("""`float16` model export is only supported on GPUs with CUDA"""	)
    else:
        _lowerCamelCase     :		List[str] 						=						"""cpu"""
    _lowerCamelCase     :		Dict 						=						Path(__A	)

    # VAE DECODER
    _lowerCamelCase     :		Optional[Any] 						=						AutoencoderKL.from_pretrained(model_path + """/vae"""	)
    _lowerCamelCase     :		List[str] 						=						vae_decoder.config.latent_channels
    # forward only through the decoder part
    _lowerCamelCase     :		Tuple 						=						vae_decoder.decode
    onnx_export(
        __A   , model_args=(
            torch.randn(1   , __A   , 25   , 25	).to(device=__A   , dtype=__A	),
            False,
        )   , output_path=output_path / """vae_decoder""" / """model.onnx"""   , ordered_input_names=["""latent_sample""", """return_dict"""]   , output_names=["""sample"""]   , dynamic_axes={
            """latent_sample""": {0: """batch""", 1: """channels""", 2: """height""", 3: """width"""},
        }   , opset=__A   , )
    del vae_decoder


if __name__ == "__main__":
    lowerCAmelCase			:							Optional[int]     =argparse.ArgumentParser()

    parser.add_argument(
        "--model_path",
        type=str,
        required=True,
        help="Path to the `diffusers` checkpoint to convert (either a local directory or on the Hub).",
    )

    parser.add_argument("--output_path", type=str, required=True, help="Path to the output model.")
    parser.add_argument(
        "--opset",
        default=14,
        type=int,
        help="The version of the ONNX operator set to use.",
    )
    parser.add_argument("--fp16", action="store_true", default=False, help="Export the models in `float16` mode")

    lowerCAmelCase			:							Optional[Any]     =parser.parse_args()
    print(args.output_path)
    convert_models(args.model_path, args.output_path, args.opset, args.fpaa)
    print("SD: Done: ONNX")




