





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 			A__   (							__UpperCAmelCase			,  __UpperCAmelCase			,  __UpperCAmelCase			,  unittest.TestCase     ):




   """simple docstring"""



   __A       :    int	  =  StableUnCLIPPipeline
   __A       :    Union[str, Any]	  =  TEXT_TO_IMAGE_PARAMS
   __A       :    Union[str, Any]	  =  TEXT_TO_IMAGE_BATCH_PARAMS
   __A       :    List[str]	  =  TEXT_TO_IMAGE_IMAGE_PARAMS
   __A       :    Dict	  =  TEXT_TO_IMAGE_IMAGE_PARAMS

   # TODO(will) Expected attn_bias.stride(1) == 0 to be true, but got false
   __A       :    List[str]	  =  False

   def 							__lowercase		( self)      ->						Any:







          '''simple docstring'''

          a__						:     Optional[Any]			=    32
          a__						:     str			=    embedder_hidden_size

          # prior components

          torch.manual_seed(0)
          a__						:     Tuple			=    CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip')

          torch.manual_seed(0)
          a__						:     List[Any]			=    CLIPTextModelWithProjection(
              CLIPTextConfig(
                  bos_token_id=0 ,  eos_token_id=2 ,  hidden_size=lowercase ,  projection_dim=lowercase ,  intermediate_size=37 ,  layer_norm_eps=1e-05 ,  num_attention_heads=4 ,  num_hidden_layers=5 ,  pad_token_id=1 ,  vocab_size=1000 ,  ))

          torch.manual_seed(0)
          a__						:     Optional[Any]			=    PriorTransformer(
              num_attention_heads=2 ,  attention_head_dim=12 ,  embedding_dim=lowercase ,  num_layers=1 ,  )

          torch.manual_seed(0)
          a__						:     str			=    DDPMScheduler(
              variance_type='fixed_small_log' ,  prediction_type='sample' ,  num_train_timesteps=1000 ,  clip_sample=lowercase ,  clip_sample_range=5.0 ,  beta_schedule='squaredcos_cap_v2' ,  )

          # regular denoising components

          torch.manual_seed(0)
          a__						:     int			=    StableUnCLIPImageNormalizer(embedding_dim=lowercase)
          a__						:     List[Any]			=    DDPMScheduler(beta_schedule='squaredcos_cap_v2')

          torch.manual_seed(0)
          a__						:     Optional[int]			=    CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip')

          torch.manual_seed(0)
          a__						:     Optional[int]			=    CLIPTextModel(
              CLIPTextConfig(
                  bos_token_id=0 ,  eos_token_id=2 ,  hidden_size=lowercase ,  projection_dim=32 ,  intermediate_size=37 ,  layer_norm_eps=1e-05 ,  num_attention_heads=4 ,  num_hidden_layers=5 ,  pad_token_id=1 ,  vocab_size=1000 ,  ))

          torch.manual_seed(0)
          a__						:     str			=    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=lowercase ,  layers_per_block=1 ,  upcast_attention=lowercase ,  use_linear_projection=lowercase ,  )

          torch.manual_seed(0)
          a__						:     int			=    DDIMScheduler(
              beta_schedule='scaled_linear' ,  beta_start=0.0_00_85 ,  beta_end=0.0_12 ,  prediction_type='v_prediction' ,  set_alpha_to_one=lowercase ,  steps_offset=1 ,  )

          torch.manual_seed(0)
          a__						:     Dict			=    AutoencoderKL()

          a__						:     Optional[Any]			=    {
              # 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 							__lowercase		( self ,  lowercase ,  lowercase=0)      ->						Tuple:







          '''simple docstring'''

          if str(lowercase).startswith('mps'):
                 a__						:     Tuple			=    torch.manual_seed(lowercase)
          else:
                 a__						:     Dict			=    torch.Generator(device=lowercase).manual_seed(lowercase)
          a__						:     Union[str, Any]			=    {
              '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 							__lowercase		( self)      ->						str:







          '''simple docstring'''

          a__						:     Dict			=    torch_device == 'cpu'

          self._test_attention_slicing_forward_pass(test_max_difference=lowercase)

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







          '''simple docstring'''

          a__						:     str			=    torch_device in ['cpu', 'mps']

          self._test_inference_batch_single_identical(test_max_difference=lowercase)





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




   """simple docstring"""



   def 							__lowercase		( self)      ->						Optional[Any]:







          '''simple docstring'''

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

   def 							__lowercase		( self)      ->						str:







          '''simple docstring'''

          a__						:     Tuple			=    load_numpy(
              'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_l_anime_turtle_fp16.npy')

          a__						:     int			=    StableUnCLIPPipeline.from_pretrained('fusing/stable-unclip-2-1-l' ,  torch_dtype=torch.floataa)
          pipe.to(lowercase)
          pipe.set_progress_bar_config(disable=lowercase)
          # 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()

          a__						:     Any			=    torch.Generator(device='cpu').manual_seed(0)
          a__						:     Dict			=    pipe('anime turle' ,  generator=lowercase ,  output_type='np')

          a__						:     Any			=    output.images[0]

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

          assert_mean_pixel_difference(lowercase ,  lowercase)

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







          '''simple docstring'''

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

          a__						:     Any			=    StableUnCLIPPipeline.from_pretrained('fusing/stable-unclip-2-1-l' ,  torch_dtype=torch.floataa)
          a__						:     Any			=    pipe.to(lowercase)
          pipe.set_progress_bar_config(disable=lowercase)
          pipe.enable_attention_slicing()
          pipe.enable_sequential_cpu_offload()

          a__						:     int			=    pipe(
              'anime turtle' ,  prior_num_inference_steps=2 ,  num_inference_steps=2 ,  output_type='np' ,  )

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









import argparse
import os
import pickle
import sys

import torch

from transformers import TransfoXLConfig, TransfoXLLMHeadModel, load_tf_weights_in_transfo_xl
from transformers.models.transfo_xl import tokenization_transfo_xl as data_utils
from transformers.models.transfo_xl.tokenization_transfo_xl import CORPUS_NAME, VOCAB_FILES_NAMES
from transformers.utils import CONFIG_NAME, WEIGHTS_NAME, logging


logging.set_verbosity_info()

# We do this to be able to load python 2 datasets pickles
# See e.g. https://stackoverflow.com/questions/2121874/python-pickling-after-changing-a-modules-directory/2121918#2121918
lowercase				:	Union[str, Any]      						=  data_utils.TransfoXLTokenizer
lowercase				:	Optional[int]      						=  data_utils.TransfoXLCorpus
lowercase				:	List[Any]      						=  data_utils
lowercase				:	Tuple      						=  data_utils



def A_     (		A__		,				A__		,				A__		,				A__    )  ->		Optional[Any]:
       if transfo_xl_dataset_file:
              # Convert a pre-processed corpus (see original TensorFlow repo)
              with open(A__		,				'rb'    ) as fp:
                     a__						:     int			=    pickle.load(A__		,				encoding='latin1'    )
              # Save vocabulary and dataset cache as Dictionaries (should be better than pickles for the long-term)
              a__						:     int			=    pytorch_dump_folder_path + '/' + VOCAB_FILES_NAMES['pretrained_vocab_file']
              print(F'Save vocabulary to {pytorch_vocab_dump_path}'    )
              a__						:     List[Any]			=    corpus.vocab.__dict__
              torch.save(A__		,				A__    )

              a__						:     Dict			=    corpus.__dict__
              corpus_dict_no_vocab.pop('vocab'		,				A__    )
              a__						:     Optional[int]			=    pytorch_dump_folder_path + '/' + CORPUS_NAME
              print(F'Save dataset to {pytorch_dataset_dump_path}'    )
              torch.save(A__		,				A__    )

       if tf_checkpoint_path:
              # Convert a pre-trained TensorFlow model
              a__						:     Union[str, Any]			=    os.path.abspath(A__    )
              a__						:     Optional[Any]			=    os.path.abspath(A__    )

              print(F'Converting Transformer XL checkpoint from {tf_path} with config at {config_path}.'    )
              # Initialise PyTorch model
              if transfo_xl_config_file == "":
                     a__						:     Dict			=    TransfoXLConfig()
              else:
                     a__						:     Dict			=    TransfoXLConfig.from_json_file(A__    )
              print(F'Building PyTorch model from configuration: {config}'    )
              a__						:     Optional[int]			=    TransfoXLLMHeadModel(A__    )

              a__						:     int			=    load_tf_weights_in_transfo_xl(A__		,				A__		,				A__    )
              # Save pytorch-model
              a__						:     Any			=    os.path.join(A__		,				A__    )
              a__						:     Dict			=    os.path.join(A__		,				A__    )
              print(F'Save PyTorch model to {os.path.abspath(A__    )}'    )
              torch.save(model.state_dict()		,				A__    )
              print(F'Save configuration file to {os.path.abspath(A__    )}'    )
              with open(A__		,				'w'		,				encoding='utf-8'    ) as f:
                     f.write(config.to_json_string()    )


if __name__ == "__main__":
    lowercase				:	List[Any]      						=  argparse.ArgumentParser()
    parser.add_argument(
        """--pytorch_dump_folder_path""",
        default=None,
        type=str,
        required=True,
        help="""Path to the folder to store the PyTorch model or dataset/vocab.""",
    )
    parser.add_argument(
        """--tf_checkpoint_path""",
        default="""""",
        type=str,
        help="""An optional path to a TensorFlow checkpoint path to be converted.""",
    )
    parser.add_argument(
        """--transfo_xl_config_file""",
        default="""""",
        type=str,
        help=(
            """An optional config json file corresponding to the pre-trained BERT model. \n"""
            """This specifies the model architecture."""
        ),
    )
    parser.add_argument(
        """--transfo_xl_dataset_file""",
        default="""""",
        type=str,
        help="""An optional dataset file to be converted in a vocabulary.""",
    )
    lowercase				:	Any      						=  parser.parse_args()
    convert_transfo_xl_checkpoint_to_pytorch(
        args.tf_checkpoint_path,
        args.transfo_xl_config_file,
        args.pytorch_dump_folder_path,
        args.transfo_xl_dataset_file,
    )


