
"""simple docstring"""







import argparse
import json
from pathlib import Path

import requests
import torch
from datasets import load_dataset
from huggingface_hub import hf_hub_download
from PIL import Image

from transformers import (
    BeitConfig,
    BeitForImageClassification,
    BeitForMaskedImageModeling,
    BeitForSemanticSegmentation,
    BeitImageProcessor,
)
from transformers.image_utils import PILImageResampling
from transformers.utils import logging


logging.set_verbosity_info()
SCREAMING_SNAKE_CASE_  					= logging.get_logger(__name__)
def       __UpperCAmelCase		(		a_      :		Optional[Any]     ,	a_      :		Optional[Any]=False     ,	a_      :		Any=False					)  ->							List[str]:
   _UpperCAmelCase 						=       "backbone." if is_semantic else ""

   _UpperCAmelCase 						=       []
   for i in range(config.num_hidden_layers					):
      # encoder layers: output projection, 2 feedforward neural networks and 2 layernorms
      rename_keys.append((F'''{prefix}blocks.{i}.norm1.weight''', F'''beit.encoder.layer.{i}.layernorm_before.weight''')					)
      rename_keys.append((F'''{prefix}blocks.{i}.norm1.bias''', F'''beit.encoder.layer.{i}.layernorm_before.bias''')					)
      rename_keys.append(
          (F'''{prefix}blocks.{i}.attn.proj.weight''', F'''beit.encoder.layer.{i}.attention.output.dense.weight''')					)
      rename_keys.append(
          (F'''{prefix}blocks.{i}.attn.proj.bias''', F'''beit.encoder.layer.{i}.attention.output.dense.bias''')					)
      rename_keys.append((F'''{prefix}blocks.{i}.norm2.weight''', F'''beit.encoder.layer.{i}.layernorm_after.weight''')					)
      rename_keys.append((F'''{prefix}blocks.{i}.norm2.bias''', F'''beit.encoder.layer.{i}.layernorm_after.bias''')					)
      rename_keys.append((F'''{prefix}blocks.{i}.mlp.fc1.weight''', F'''beit.encoder.layer.{i}.intermediate.dense.weight''')					)
      rename_keys.append((F'''{prefix}blocks.{i}.mlp.fc1.bias''', F'''beit.encoder.layer.{i}.intermediate.dense.bias''')					)
      rename_keys.append((F'''{prefix}blocks.{i}.mlp.fc2.weight''', F'''beit.encoder.layer.{i}.output.dense.weight''')					)
      rename_keys.append((F'''{prefix}blocks.{i}.mlp.fc2.bias''', F'''beit.encoder.layer.{i}.output.dense.bias''')					)

   # projection layer + position embeddings
   rename_keys.extend(
       [
           (F'''{prefix}cls_token''', "beit.embeddings.cls_token"),
           (F'''{prefix}patch_embed.proj.weight''', "beit.embeddings.patch_embeddings.projection.weight"),
           (F'''{prefix}patch_embed.proj.bias''', "beit.embeddings.patch_embeddings.projection.bias"),
       ]					)

   if has_lm_head:
      # mask token + shared relative position bias + layernorm
      rename_keys.extend(
          [
              ("mask_token", "beit.embeddings.mask_token"),
              (
                  "rel_pos_bias.relative_position_bias_table",
                  "beit.encoder.relative_position_bias.relative_position_bias_table",
              ),
              (
                  "rel_pos_bias.relative_position_index",
                  "beit.encoder.relative_position_bias.relative_position_index",
              ),
              ("norm.weight", "layernorm.weight"),
              ("norm.bias", "layernorm.bias"),
          ]					)
   elif is_semantic:
      # semantic segmentation classification heads
      rename_keys.extend(
          [
              ("decode_head.conv_seg.weight", "decode_head.classifier.weight"),
              ("decode_head.conv_seg.bias", "decode_head.classifier.bias"),
              ("auxiliary_head.conv_seg.weight", "auxiliary_head.classifier.weight"),
              ("auxiliary_head.conv_seg.bias", "auxiliary_head.classifier.bias"),
          ]					)
   else:
      # layernorm + classification head
      rename_keys.extend(
          [
              ("fc_norm.weight", "beit.pooler.layernorm.weight"),
              ("fc_norm.bias", "beit.pooler.layernorm.bias"),
              ("head.weight", "classifier.weight"),
              ("head.bias", "classifier.bias"),
          ]					)

   return rename_keys
def       __UpperCAmelCase		(		a_      :		List[str]     ,	a_      :		Optional[Any]     ,	a_      :		Tuple=False     ,	a_      :		Optional[int]=False					)  ->							Optional[Any]:
   for i in range(config.num_hidden_layers					):
      _UpperCAmelCase 						=       "backbone." if is_semantic else ""
      # queries, keys and values
      _UpperCAmelCase 						=       state_dict.pop(F'''{prefix}blocks.{i}.attn.qkv.weight'''					)
      _UpperCAmelCase 						=       state_dict.pop(F'''{prefix}blocks.{i}.attn.q_bias'''					)
      _UpperCAmelCase 						=       state_dict.pop(F'''{prefix}blocks.{i}.attn.v_bias'''					)

      _UpperCAmelCase 						=       in_proj_weight[
          : config.hidden_size, :
      ]
      _UpperCAmelCase 						=       q_bias
      _UpperCAmelCase 						=       in_proj_weight[
          config.hidden_size : config.hidden_size * 2, :
      ]
      _UpperCAmelCase 						=       in_proj_weight[
          -config.hidden_size :, :
      ]
      _UpperCAmelCase 						=       v_bias

      # gamma_1 and gamma_2
      # we call them lambda because otherwise they are renamed when using .from_pretrained
      _UpperCAmelCase 						=       state_dict.pop(F'''{prefix}blocks.{i}.gamma_1'''					)
      _UpperCAmelCase 						=       state_dict.pop(F'''{prefix}blocks.{i}.gamma_2'''					)

      _UpperCAmelCase 						=       gamma_a
      _UpperCAmelCase 						=       gamma_a

      # relative_position bias table + index
      if not has_lm_head:
         # each layer has its own relative position bias
         _UpperCAmelCase 						=       state_dict.pop(F'''{prefix}blocks.{i}.attn.relative_position_bias_table'''					)
         _UpperCAmelCase 						=       state_dict.pop(F'''{prefix}blocks.{i}.attn.relative_position_index'''					)

         _UpperCAmelCase 						=       table
         _UpperCAmelCase 						=       index
def       __UpperCAmelCase		(		a_      :		Optional[Any]     ,	a_      :		Dict     ,	a_      :		Any					)  ->							str:
   _UpperCAmelCase 						=       dct.pop(a_					)
   _UpperCAmelCase 						=       val
def       __UpperCAmelCase		(		)  ->							Any:
   _UpperCAmelCase 						=       "http://images.cocodataset.org/val2017/000000039769.jpg"
   _UpperCAmelCase 						=       Image.open(requests.get(a_     ,	stream=a_					).raw					)
   return im
@torch.no_grad()
def       __UpperCAmelCase		(		a_      :		List[str]     ,	a_      :		List[str]					)  ->							List[str]:
   _UpperCAmelCase 						=       BeitConfig()
   _UpperCAmelCase 						=       False
   _UpperCAmelCase 						=       False
   _UpperCAmelCase 						=       "huggingface/label-files"
   # set config parameters based on URL
   if checkpoint_url[-9:-4] == "pt22k":
      # masked image modeling
      _UpperCAmelCase 						=       True
      _UpperCAmelCase 						=       True
      _UpperCAmelCase 						=       True
   elif checkpoint_url[-9:-4] == "ft22k":
      # intermediate fine-tuning on ImageNet-22k
      _UpperCAmelCase 						=       True
      _UpperCAmelCase 						=       2_18_41
      _UpperCAmelCase 						=       "imagenet-22k-id2label.json"
      _UpperCAmelCase 						=       json.load(open(hf_hub_download(a_     ,	a_     ,	repo_type="dataset"					)     ,	"r"					)					)
      _UpperCAmelCase 						=       {int(a_					): v for k, v in idalabel.items()}
      # this dataset contains 21843 labels but the model only has 21841
      # we delete the classes as mentioned in https://github.com/google-research/big_transfer/issues/18
      del idalabel[92_05]
      del idalabel[1_50_27]
      _UpperCAmelCase 						=       idalabel
      _UpperCAmelCase 						=       {v: k for k, v in idalabel.items()}
   elif checkpoint_url[-8:-4] == "to1k":
      # fine-tuning on ImageNet-1k
      _UpperCAmelCase 						=       True
      _UpperCAmelCase 						=       10_00
      _UpperCAmelCase 						=       "imagenet-1k-id2label.json"
      _UpperCAmelCase 						=       json.load(open(hf_hub_download(a_     ,	a_     ,	repo_type="dataset"					)     ,	"r"					)					)
      _UpperCAmelCase 						=       {int(a_					): v for k, v in idalabel.items()}
      _UpperCAmelCase 						=       idalabel
      _UpperCAmelCase 						=       {v: k for k, v in idalabel.items()}
      if "384" in checkpoint_url:
         _UpperCAmelCase 						=       3_84
      if "512" in checkpoint_url:
         _UpperCAmelCase 						=       5_12
   elif "ade20k" in checkpoint_url:
      # fine-tuning
      _UpperCAmelCase 						=       True
      _UpperCAmelCase 						=       1_50
      _UpperCAmelCase 						=       "ade20k-id2label.json"
      _UpperCAmelCase 						=       json.load(open(hf_hub_download(a_     ,	a_     ,	repo_type="dataset"					)     ,	"r"					)					)
      _UpperCAmelCase 						=       {int(a_					): v for k, v in idalabel.items()}
      _UpperCAmelCase 						=       idalabel
      _UpperCAmelCase 						=       {v: k for k, v in idalabel.items()}
      _UpperCAmelCase 						=       6_40
      _UpperCAmelCase 						=       True
   else:
      raise ValueError("Checkpoint not supported, URL should either end with 'pt22k', 'ft22k', 'to1k' or 'ade20k'"					)

   # size of the architecture
   if "base" in checkpoint_url:
      pass
   elif "large" in checkpoint_url:
      _UpperCAmelCase 						=       10_24
      _UpperCAmelCase 						=       40_96
      _UpperCAmelCase 						=       24
      _UpperCAmelCase 						=       16
      if "ade20k" in checkpoint_url:
         _UpperCAmelCase 						=       6_40
         _UpperCAmelCase 						=       [7, 11, 15, 23]
   else:
      raise ValueError("Should either find 'base' or 'large' in checkpoint URL"					)

   # load state_dict of original model, remove and rename some keys
   _UpperCAmelCase 						=       torch.hub.load_state_dict_from_url(a_     ,	map_location="cpu"     ,	check_hash=a_					)
   _UpperCAmelCase 						=       state_dict["model"] if "ade20k" not in checkpoint_url else state_dict["state_dict"]

   _UpperCAmelCase 						=       create_rename_keys(a_     ,	has_lm_head=a_     ,	is_semantic=a_					)
   for src, dest in rename_keys:
      rename_key(a_     ,	a_     ,	a_					)
   read_in_q_k_v(a_     ,	a_     ,	has_lm_head=a_     ,	is_semantic=a_					)
   if is_semantic:
      # add prefix to decoder keys
      for key, val in state_dict.copy().items():
         _UpperCAmelCase 						=       state_dict.pop(a_					)
         if key.startswith("backbone.fpn"					):
            _UpperCAmelCase 						=       key.replace("backbone.fpn"     ,	"fpn"					)
         _UpperCAmelCase 						=       val

    # load HuggingFace model
   if checkpoint_url[-9:-4] == "pt22k":
      _UpperCAmelCase 						=       BeitForMaskedImageModeling(a_					)
   elif "ade20k" in checkpoint_url:
      _UpperCAmelCase 						=       BeitForSemanticSegmentation(a_					)
   else:
      _UpperCAmelCase 						=       BeitForImageClassification(a_					)
   model.eval()
   model.load_state_dict(a_					)

   # Check outputs on an image
   if is_semantic:
      _UpperCAmelCase 						=       BeitImageProcessor(size=config.image_size     ,	do_center_crop=a_					)
      _UpperCAmelCase 						=       load_dataset("hf-internal-testing/fixtures_ade20k"     ,	split="test"					)
      _UpperCAmelCase 						=       Image.open(ds[0]["file"]					)
   else:
      _UpperCAmelCase 						=       BeitImageProcessor(
          size=config.image_size     ,	resample=PILImageResampling.BILINEAR     ,	do_center_crop=a_					)
      _UpperCAmelCase 						=       prepare_img()

   _UpperCAmelCase 						=       image_processor(images=a_     ,	return_tensors="pt"					)
   _UpperCAmelCase 						=       encoding["pixel_values"]

   _UpperCAmelCase 						=       model(a_					)
   _UpperCAmelCase 						=       outputs.logits

   # verify logits
   _UpperCAmelCase 						=       torch.Size([1, 10_00]					)
   if checkpoint_url[:-4].endswith("beit_base_patch16_224_pt22k"					):
      _UpperCAmelCase 						=       torch.Size([1, 1_96, 81_92]					)
   elif checkpoint_url[:-4].endswith("beit_large_patch16_224_pt22k"					):
      _UpperCAmelCase 						=       torch.Size([1, 1_96, 81_92]					)
   elif checkpoint_url[:-4].endswith("beit_base_patch16_224_pt22k_ft22k"					):
      _UpperCAmelCase 						=       torch.Size([1, 2_18_41]					)
      _UpperCAmelCase 						=       torch.tensor([2.2_288, 2.4_671, 0.7_395]					)
      _UpperCAmelCase 						=       23_97
   elif checkpoint_url[:-4].endswith("beit_large_patch16_224_pt22k_ft22k"					):
      _UpperCAmelCase 						=       torch.Size([1, 2_18_41]					)
      _UpperCAmelCase 						=       torch.tensor([1.6_881, -0.2_787, 0.5_901]					)
      _UpperCAmelCase 						=       23_96
   elif checkpoint_url[:-4].endswith("beit_base_patch16_224_pt22k_ft1k"					):
      _UpperCAmelCase 						=       torch.tensor([0.1_241, 0.0_798, -0.6_569]					)
      _UpperCAmelCase 						=       2_85
   elif checkpoint_url[:-4].endswith("beit_base_patch16_224_pt22k_ft22kto1k"					):
      _UpperCAmelCase 						=       torch.tensor([-1.2_385, -1.0_987, -1.0_108]					)
      _UpperCAmelCase 						=       2_81
   elif checkpoint_url[:-4].endswith("beit_base_patch16_384_pt22k_ft22kto1k"					):
      _UpperCAmelCase 						=       torch.tensor([-1.5_303, -0.9_484, -0.3_147]					)
      _UpperCAmelCase 						=       7_61
   elif checkpoint_url[:-4].endswith("beit_large_patch16_224_pt22k_ft1k"					):
      _UpperCAmelCase 						=       torch.tensor([0.4_610, -0.0_928, 0.2_086]					)
      _UpperCAmelCase 						=       7_61
   elif checkpoint_url[:-4].endswith("beit_large_patch16_224_pt22k_ft22kto1k"					):
      _UpperCAmelCase 						=       torch.tensor([-0.4_804, 0.6_257, -0.1_837]					)
      _UpperCAmelCase 						=       7_61
   elif checkpoint_url[:-4].endswith("beit_large_patch16_384_pt22k_ft22kto1k"					):
      _UpperCAmelCase 						=       torch.tensor([[-0.5_122, 0.5_117, -0.2_113]]					)
      _UpperCAmelCase 						=       7_61
   elif checkpoint_url[:-4].endswith("beit_large_patch16_512_pt22k_ft22kto1k"					):
      _UpperCAmelCase 						=       torch.tensor([-0.3_062, 0.7_261, 0.4_852]					)
      _UpperCAmelCase 						=       7_61
   elif checkpoint_url[:-4].endswith("beit_base_patch16_640_pt22k_ft22ktoade20k"					):
      _UpperCAmelCase 						=       (1, 1_50, 1_60, 1_60)
      _UpperCAmelCase 						=       torch.tensor(
          [
              [[-4.9_225, -2.3_954, -3.0_522], [-2.8_822, -1.0_046, -1.7_561], [-2.9_549, -1.3_228, -2.1_347]],
              [[-5.8_168, -3.4_129, -4.0_778], [-3.8_651, -2.2_214, -3.0_277], [-3.8_356, -2.4_643, -3.3_535]],
              [[-0.0_078, 3.9_952, 4.0_754], [2.9_856, 4.6_944, 5.0_035], [3.2_413, 4.7_813, 4.9_969]],
          ]					)
   elif checkpoint_url[:-4].endswith("beit_large_patch16_640_pt22k_ft22ktoade20k"					):
      _UpperCAmelCase 						=       (1, 1_50, 1_60, 1_60)
      _UpperCAmelCase 						=       torch.tensor(
          [
              [[-4.3_305, -2.3_049, -3.0_161], [-2.9_591, -1.5_305, -2.2_251], [-3.4_198, -1.8_004, -2.9_062]],
              [[-5.8_922, -3.7_435, -4.3_978], [-4.2_063, -2.7_872, -3.4_755], [-4.2_791, -3.1_874, -4.1_681]],
              [[0.9_895, 4.3_467, 4.7_663], [4.2_476, 5.6_830, 6.1_518], [4.5_550, 6.2_495, 6.5_154]],
          ]					)
   else:
      raise ValueError("Can't verify logits as model is not supported"					)

   assert logits.shape == expected_shape, "Shape of logits not as expected"
   if not has_lm_head:
      if is_semantic:
         assert torch.allclose(
             logits[0, :3, :3, :3]     ,	a_     ,	atol=1e-3					), "First elements of logits not as expected"
      else:
         print("Predicted class idx:"     ,	logits.argmax(-1					).item()					)
         assert torch.allclose(
             logits[0, :3]     ,	a_     ,	atol=1e-3					), "First elements of logits not as expected"
         assert logits.argmax(-1					).item() == expected_class_idx, "Predicted class index not as expected"

   Path(a_					).mkdir(exist_ok=a_					)
   print(F'''Saving model to {pytorch_dump_folder_path}'''					)
   model.save_pretrained(a_					)
   print(F'''Saving image processor to {pytorch_dump_folder_path}'''					)
   image_processor.save_pretrained(a_					)


if __name__ == "__main__":
  SCREAMING_SNAKE_CASE_  					= argparse.ArgumentParser()

  parser.add_argument(
      '--checkpoint_url',
      default='https://conversationhub.blob.core.windows.net/beit-share-public/beit/beit_base_patch16_224_pt22k_ft22kto1k.pth',
      type=str,
      help='URL to the original PyTorch checkpoint (.pth file).',
  )
  parser.add_argument(
      '--pytorch_dump_folder_path', default=None, type=str, help='Path to the folder to output PyTorch model.'
  )
  SCREAMING_SNAKE_CASE_  					= parser.parse_args()
  convert_beit_checkpoint(args.checkpoint_url, args.pytorch_dump_folder_path)



"""simple docstring"""







import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModelWithProjection, CLIPTokenizer

from diffusers import PriorTransformer, UnCLIPPipeline, UnCLIPScheduler, UNetaDConditionModel, UNetaDModel
from diffusers.pipelines.unclip.text_proj import UnCLIPTextProjModel
from diffusers.utils import load_numpy, nightly, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu, skip_mps

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()






class      __a      (       lowerCAmelCase_					,						unittest.TestCase      ):
       _lowerCamelCase				:				int   			=      UnCLIPPipeline
       _lowerCamelCase				:				Any   			=      TEXT_TO_IMAGE_PARAMS - {
           'negative_prompt',
           'height',
           'width',
           'negative_prompt_embeds',
           'guidance_scale',
           'prompt_embeds',
           'cross_attention_kwargs',
       }
       _lowerCamelCase				:				Optional[int]   			=      TEXT_TO_IMAGE_BATCH_PARAMS
       _lowerCamelCase				:				Dict   			=      [
           'generator',
           'return_dict',
           'prior_num_inference_steps',
           'decoder_num_inference_steps',
           'super_res_num_inference_steps',
       ]
       _lowerCamelCase				:				Any   			=      False







       @property
       def  _snake_case      (       self		):



          '''simple docstring'''

          return 32







       @property
       def  _snake_case      (       self		):



          '''simple docstring'''

          return 32







       @property
       def  _snake_case      (       self		):



          '''simple docstring'''

          return self.time_input_dim







       @property
       def  _snake_case      (       self		):



          '''simple docstring'''

          return self.time_input_dim * 4







       @property
       def  _snake_case      (       self		):



          '''simple docstring'''

          return 100







       @property
       def  _snake_case      (       self		):



          '''simple docstring'''

          _UpperCAmelCase 						=       CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"		)
          return tokenizer







       @property
       def  _snake_case      (       self		):



          '''simple docstring'''

          torch.manual_seed(0		)
          _UpperCAmelCase 						=       CLIPTextConfig(
              bos_token_id=0				,eos_token_id=2				,hidden_size=self.text_embedder_hidden_size				,projection_dim=self.text_embedder_hidden_size				,intermediate_size=37				,layer_norm_eps=1e-05				,num_attention_heads=4				,num_hidden_layers=5				,pad_token_id=1				,vocab_size=1000				,)
          return CLIPTextModelWithProjection(lowerCamelCase_		)







       @property
       def  _snake_case      (       self		):



          '''simple docstring'''

          torch.manual_seed(0		)

          _UpperCAmelCase 						=       {
              "num_attention_heads": 2,
              "attention_head_dim": 12,
              "embedding_dim": self.text_embedder_hidden_size,
              "num_layers": 1,
          }

          _UpperCAmelCase 						=       PriorTransformer(**lowerCamelCase_		)
          return model







       @property
       def  _snake_case      (       self		):



          '''simple docstring'''

          torch.manual_seed(0		)

          _UpperCAmelCase 						=       {
              "clip_embeddings_dim": self.text_embedder_hidden_size,
              "time_embed_dim": self.time_embed_dim,
              "cross_attention_dim": self.cross_attention_dim,
          }

          _UpperCAmelCase 						=       UnCLIPTextProjModel(**lowerCamelCase_		)
          return model







       @property
       def  _snake_case      (       self		):



          '''simple docstring'''

          torch.manual_seed(0		)

          _UpperCAmelCase 						=       {
              "sample_size": 32,
              # RGB in channels
              "in_channels": 3,
              # Out channels is double in channels because predicts mean and variance
              "out_channels": 6,
              "down_block_types": ("ResnetDownsampleBlock2D", "SimpleCrossAttnDownBlock2D"),
              "up_block_types": ("SimpleCrossAttnUpBlock2D", "ResnetUpsampleBlock2D"),
              "mid_block_type": "UNetMidBlock2DSimpleCrossAttn",
              "block_out_channels": (self.block_out_channels_a, self.block_out_channels_a * 2),
              "layers_per_block": 1,
              "cross_attention_dim": self.cross_attention_dim,
              "attention_head_dim": 4,
              "resnet_time_scale_shift": "scale_shift",
              "class_embed_type": "identity",
          }

          _UpperCAmelCase 						=       UNetaDConditionModel(**lowerCamelCase_		)
          return model







       @property
       def  _snake_case      (       self		):



          '''simple docstring'''

          return {
              "sample_size": 64,
              "layers_per_block": 1,
              "down_block_types": ("ResnetDownsampleBlock2D", "ResnetDownsampleBlock2D"),
              "up_block_types": ("ResnetUpsampleBlock2D", "ResnetUpsampleBlock2D"),
              "block_out_channels": (self.block_out_channels_a, self.block_out_channels_a * 2),
              "in_channels": 6,
              "out_channels": 3,
          }







       @property
       def  _snake_case      (       self		):



          '''simple docstring'''

          torch.manual_seed(0		)

          _UpperCAmelCase 						=       UNetaDModel(**self.dummy_super_res_kwargs		)
          return model







       @property
       def  _snake_case      (       self		):



          '''simple docstring'''

          torch.manual_seed(1		)

          _UpperCAmelCase 						=       UNetaDModel(**self.dummy_super_res_kwargs		)
          return model







       def  _snake_case      (       self		):



          '''simple docstring'''

          _UpperCAmelCase 						=       self.dummy_prior
          _UpperCAmelCase 						=       self.dummy_decoder
          _UpperCAmelCase 						=       self.dummy_text_proj
          _UpperCAmelCase 						=       self.dummy_text_encoder
          _UpperCAmelCase 						=       self.dummy_tokenizer
          _UpperCAmelCase 						=       self.dummy_super_res_first
          _UpperCAmelCase 						=       self.dummy_super_res_last

          _UpperCAmelCase 						=       UnCLIPScheduler(
              variance_type="fixed_small_log"				,prediction_type="sample"				,num_train_timesteps=1000				,clip_sample_range=5.0				,)

          _UpperCAmelCase 						=       UnCLIPScheduler(
              variance_type="learned_range"				,prediction_type="epsilon"				,num_train_timesteps=1000				,)

          _UpperCAmelCase 						=       UnCLIPScheduler(
              variance_type="fixed_small_log"				,prediction_type="epsilon"				,num_train_timesteps=1000				,)

          _UpperCAmelCase 						=       {
              "prior": prior,
              "decoder": decoder,
              "text_proj": text_proj,
              "text_encoder": text_encoder,
              "tokenizer": tokenizer,
              "super_res_first": super_res_first,
              "super_res_last": super_res_last,
              "prior_scheduler": prior_scheduler,
              "decoder_scheduler": decoder_scheduler,
              "super_res_scheduler": super_res_scheduler,
          }

          return components







       def  _snake_case      (       self				,lowerCamelCase_				,lowerCamelCase_=0		):



          '''simple docstring'''

          if str(lowerCamelCase_		).startswith("mps"		):
             _UpperCAmelCase 						=       torch.manual_seed(lowerCamelCase_		)
          else:
             _UpperCAmelCase 						=       torch.Generator(device=lowerCamelCase_		).manual_seed(lowerCamelCase_		)
          _UpperCAmelCase 						=       {
              "prompt": "horse",
              "generator": generator,
              "prior_num_inference_steps": 2,
              "decoder_num_inference_steps": 2,
              "super_res_num_inference_steps": 2,
              "output_type": "numpy",
          }
          return inputs







       def  _snake_case      (       self		):



          '''simple docstring'''

          _UpperCAmelCase 						=       "cpu"

          _UpperCAmelCase 						=       self.get_dummy_components()

          _UpperCAmelCase 						=       self.pipeline_class(**lowerCamelCase_		)
          _UpperCAmelCase 						=       pipe.to(lowerCamelCase_		)

          pipe.set_progress_bar_config(disable=lowerCamelCase_		)

          _UpperCAmelCase 						=       pipe(**self.get_dummy_inputs(lowerCamelCase_		)		)
          _UpperCAmelCase 						=       output.images

          _UpperCAmelCase 						=       pipe(
              **self.get_dummy_inputs(lowerCamelCase_		)				,return_dict=lowerCamelCase_				,)[0]

          _UpperCAmelCase 						=       image[0, -3:, -3:, -1]
          _UpperCAmelCase 						=       image_from_tuple[0, -3:, -3:, -1]

          assert image.shape == (1, 64, 64, 3)

          _UpperCAmelCase 						=       np.array(
              [
                  0.9_9_9_7,
                  0.9_9_8_8,
                  0.0_0_2_8,
                  0.9_9_9_7,
                  0.9_9_8_4,
                  0.9_9_6_5,
                  0.0_0_2_9,
                  0.9_9_8_6,
                  0.0_0_2_5,
              ]		)

          assert np.abs(image_slice.flatten() - expected_slice		).max() < 1e-2
          assert np.abs(image_from_tuple_slice.flatten() - expected_slice		).max() < 1e-2







       def  _snake_case      (       self		):



          '''simple docstring'''

          _UpperCAmelCase 						=       torch.device("cpu"		)






          class      __a      :
                 _lowerCamelCase				:				Optional[Any]   			=      1



          _UpperCAmelCase 						=       self.get_dummy_components()

          _UpperCAmelCase 						=       self.pipeline_class(**lowerCamelCase_		)
          _UpperCAmelCase 						=       pipe.to(lowerCamelCase_		)

          _UpperCAmelCase 						=       components["prior"]
          _UpperCAmelCase 						=       components["decoder"]
          _UpperCAmelCase 						=       components["super_res_first"]
          _UpperCAmelCase 						=       components["tokenizer"]
          _UpperCAmelCase 						=       components["text_encoder"]

          _UpperCAmelCase 						=       torch.Generator(device=lowerCamelCase_		).manual_seed(0		)
          _UpperCAmelCase 						=       prior.dtype
          _UpperCAmelCase 						=       1

          _UpperCAmelCase 						=       (batch_size, prior.config.embedding_dim)
          _UpperCAmelCase 						=       pipe.prepare_latents(
              lowerCamelCase_				,dtype=lowerCamelCase_				,device=lowerCamelCase_				,generator=lowerCamelCase_				,latents=lowerCamelCase_				,scheduler=DummyScheduler()		)
          _UpperCAmelCase 						=       (batch_size, decoder.config.in_channels, decoder.config.sample_size, decoder.config.sample_size)
          _UpperCAmelCase 						=       pipe.prepare_latents(
              lowerCamelCase_				,dtype=lowerCamelCase_				,device=lowerCamelCase_				,generator=lowerCamelCase_				,latents=lowerCamelCase_				,scheduler=DummyScheduler()		)

          _UpperCAmelCase 						=       (
              batch_size,
              super_res_first.config.in_channels // 2,
              super_res_first.config.sample_size,
              super_res_first.config.sample_size,
          )
          _UpperCAmelCase 						=       pipe.prepare_latents(
              lowerCamelCase_				,dtype=lowerCamelCase_				,device=lowerCamelCase_				,generator=lowerCamelCase_				,latents=lowerCamelCase_				,scheduler=DummyScheduler()		)

          pipe.set_progress_bar_config(disable=lowerCamelCase_		)

          _UpperCAmelCase 						=       "this is a prompt example"

          _UpperCAmelCase 						=       torch.Generator(device=lowerCamelCase_		).manual_seed(0		)
          _UpperCAmelCase 						=       pipe(
              [prompt]				,generator=lowerCamelCase_				,prior_num_inference_steps=2				,decoder_num_inference_steps=2				,super_res_num_inference_steps=2				,prior_latents=lowerCamelCase_				,decoder_latents=lowerCamelCase_				,super_res_latents=lowerCamelCase_				,output_type="np"				,)
          _UpperCAmelCase 						=       output.images

          _UpperCAmelCase 						=       tokenizer(
              lowerCamelCase_				,padding="max_length"				,max_length=tokenizer.model_max_length				,return_tensors="pt"				,)
          _UpperCAmelCase 						=       text_encoder(text_inputs.input_ids		)
          _UpperCAmelCase 						=       text_inputs.attention_mask

          _UpperCAmelCase 						=       torch.Generator(device=lowerCamelCase_		).manual_seed(0		)
          _UpperCAmelCase 						=       pipe(
              generator=lowerCamelCase_				,prior_num_inference_steps=2				,decoder_num_inference_steps=2				,super_res_num_inference_steps=2				,prior_latents=lowerCamelCase_				,decoder_latents=lowerCamelCase_				,super_res_latents=lowerCamelCase_				,text_model_output=lowerCamelCase_				,text_attention_mask=lowerCamelCase_				,output_type="np"				,)[0]

          # make sure passing text embeddings manually is identical
          assert np.abs(image - image_from_text		).max() < 1e-4







       @skip_mps
       def  _snake_case      (       self		):



          '''simple docstring'''

          _UpperCAmelCase 						=       torch_device == "cpu"

          self._test_attention_slicing_forward_pass(test_max_difference=lowerCamelCase_				,expected_max_diff=0.0_1		)







       @skip_mps
       def  _snake_case      (       self		):



          '''simple docstring'''

          _UpperCAmelCase 						=       torch_device == "cpu"
          _UpperCAmelCase 						=       True
          _UpperCAmelCase 						=       [
              "prior_num_inference_steps",
              "decoder_num_inference_steps",
              "super_res_num_inference_steps",
          ]

          self._test_inference_batch_single_identical(
              test_max_difference=lowerCamelCase_				,relax_max_difference=lowerCamelCase_				,additional_params_copy_to_batched_inputs=lowerCamelCase_				,)







       def  _snake_case      (       self		):



          '''simple docstring'''

          _UpperCAmelCase 						=       [
              "prior_num_inference_steps",
              "decoder_num_inference_steps",
              "super_res_num_inference_steps",
          ]

          if torch_device == "mps":
             # TODO: MPS errors with larger batch sizes
             _UpperCAmelCase 						=       [2, 3]
             self._test_inference_batch_consistent(
                 batch_sizes=lowerCamelCase_				,additional_params_copy_to_batched_inputs=lowerCamelCase_				,)
          else:
             self._test_inference_batch_consistent(
                 additional_params_copy_to_batched_inputs=lowerCamelCase_		)







       @skip_mps
       def  _snake_case      (       self		):



          '''simple docstring'''

          return super().test_dict_tuple_outputs_equivalent()







       @skip_mps
       def  _snake_case      (       self		):



          '''simple docstring'''

          return super().test_save_load_local()







       @skip_mps
       def  _snake_case      (       self		):



          '''simple docstring'''

          return super().test_save_load_optional_components()






@nightly
class      __a      (       unittest.TestCase      ):







       def  _snake_case      (       self		):



          '''simple docstring'''

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







       def  _snake_case      (       self		):



          '''simple docstring'''

          _UpperCAmelCase 						=       load_numpy(
              "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
              "/unclip/karlo_v1_alpha_horse_cpu.npy"		)

          _UpperCAmelCase 						=       UnCLIPPipeline.from_pretrained("kakaobrain/karlo-v1-alpha"		)
          pipeline.set_progress_bar_config(disable=lowerCamelCase_		)

          _UpperCAmelCase 						=       torch.manual_seed(0		)
          _UpperCAmelCase 						=       pipeline(
              "horse"				,num_images_per_prompt=1				,generator=lowerCamelCase_				,output_type="np"				,)

          _UpperCAmelCase 						=       output.images[0]

          assert image.shape == (256, 256, 3)
          assert np.abs(expected_image - image		).max() < 1e-1






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







       def  _snake_case      (       self		):



          '''simple docstring'''

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







       def  _snake_case      (       self		):



          '''simple docstring'''

          _UpperCAmelCase 						=       load_numpy(
              "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
              "/unclip/karlo_v1_alpha_horse_fp16.npy"		)

          _UpperCAmelCase 						=       UnCLIPPipeline.from_pretrained("kakaobrain/karlo-v1-alpha"				,torch_dtype=torch.floataa		)
          _UpperCAmelCase 						=       pipeline.to(lowerCamelCase_		)
          pipeline.set_progress_bar_config(disable=lowerCamelCase_		)

          _UpperCAmelCase 						=       torch.Generator(device="cpu"		).manual_seed(0		)
          _UpperCAmelCase 						=       pipeline(
              "horse"				,generator=lowerCamelCase_				,output_type="np"				,)

          _UpperCAmelCase 						=       output.images[0]

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

          assert_mean_pixel_difference(lowerCamelCase_				,lowerCamelCase_		)







       def  _snake_case      (       self		):



          '''simple docstring'''

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

          _UpperCAmelCase 						=       UnCLIPPipeline.from_pretrained("kakaobrain/karlo-v1-alpha"				,torch_dtype=torch.floataa		)
          _UpperCAmelCase 						=       pipe.to(lowerCamelCase_		)
          pipe.set_progress_bar_config(disable=lowerCamelCase_		)
          pipe.enable_attention_slicing()
          pipe.enable_sequential_cpu_offload()

          _UpperCAmelCase 						=       pipe(
              "horse"				,num_images_per_prompt=1				,prior_num_inference_steps=2				,decoder_num_inference_steps=2				,super_res_num_inference_steps=2				,output_type="np"				,)

          _UpperCAmelCase 						=       torch.cuda.max_memory_allocated()
          # make sure that less than 7 GB is allocated
          assert mem_bytes < 7 * 10**9

