

"""simple docstring"""
import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMParallelScheduler,
    DDPMParallelScheduler,
    StableDiffusionParadigmsPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import (
    enable_full_determinism,
    require_torch_gpu,
)

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()

class        _UpperCamelCase	(						lowerCamelCase__   ,				lowerCamelCase__   ,				unittest.TestCase	):


  """simple docstring"""





  snake_case_        =    StableDiffusionParadigmsPipeline
  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



  def 							_UpperCAmelCase      ( self					: List[str]      )	->			str:


         '''simple docstring'''





         torch.manual_seed(0      )
         __magic_name__			:		Union[str, Any]        =     UNetaDConditionModel(
             block_out_channels=(32, 64)       ,       layers_per_block=2       ,       sample_size=32       ,       in_channels=4       ,       out_channels=4       ,       down_block_types=('''DownBlock2D''', '''CrossAttnDownBlock2D''')       ,       up_block_types=('''CrossAttnUpBlock2D''', '''UpBlock2D''')       ,       cross_attention_dim=32       ,       attention_head_dim=(2, 4)       ,       use_linear_projection=snake_case       ,       )
         __magic_name__			:		List[str]        =     DDIMParallelScheduler(
             beta_start=0.0_0085       ,       beta_end=0.012       ,       beta_schedule='''scaled_linear'''       ,       clip_sample=snake_case       ,       set_alpha_to_one=snake_case       ,       )
         torch.manual_seed(0      )
         __magic_name__			:		Tuple        =     AutoencoderKL(
             block_out_channels=[32, 64]       ,       in_channels=3       ,       out_channels=3       ,       down_block_types=['''DownEncoderBlock2D''', '''DownEncoderBlock2D''']       ,       up_block_types=['''UpDecoderBlock2D''', '''UpDecoderBlock2D''']       ,       latent_channels=4       ,       sample_size=128       ,       )
         torch.manual_seed(0      )
         __magic_name__			:		str        =     CLIPTextConfig(
             bos_token_id=0       ,       eos_token_id=2       ,       hidden_size=32       ,       intermediate_size=37       ,       layer_norm_eps=1e-05       ,       num_attention_heads=4       ,       num_hidden_layers=5       ,       pad_token_id=1       ,       vocab_size=1000       ,       hidden_act='''gelu'''       ,       projection_dim=512       ,       )
         __magic_name__			:		Union[str, Any]        =     CLIPTextModel(snake_case      )
         __magic_name__			:		Optional[int]        =     CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''      )

         __magic_name__			:		Tuple        =     {
             '''unet''': unet,
             '''scheduler''': scheduler,
             '''vae''': vae,
             '''text_encoder''': text_encoder,
             '''tokenizer''': tokenizer,
             '''safety_checker''': None,
             '''feature_extractor''': None,
         }
         return components



  def 							_UpperCAmelCase      ( self					: int       ,       snake_case					: Optional[int]       ,       snake_case					: List[str]=0      )	->			Optional[int]:


         '''simple docstring'''





         if str(snake_case      ).startswith('''mps'''      ):
                __magic_name__			:		Dict        =     torch.manual_seed(snake_case      )
         else:
                __magic_name__			:		List[Any]        =     torch.Generator(device=snake_case      ).manual_seed(snake_case      )
         __magic_name__			:		Tuple        =     {
             '''prompt''': '''a photograph of an astronaut riding a horse''',
             '''generator''': generator,
             '''num_inference_steps''': 10,
             '''guidance_scale''': 6.0,
             '''output_type''': '''numpy''',
             '''parallel''': 3,
             '''debug''': True,
         }
         return inputs



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


         '''simple docstring'''





         __magic_name__			:		Optional[Any]        =     '''cpu'''  # ensure determinism for the device-dependent torch.Generator
         __magic_name__			:		List[str]        =     self.get_dummy_components()
         __magic_name__			:		Optional[Any]        =     StableDiffusionParadigmsPipeline(**snake_case      )
         __magic_name__			:		List[Any]        =     sd_pipe.to(snake_case      )
         sd_pipe.set_progress_bar_config(disable=snake_case      )

         __magic_name__			:		Tuple        =     self.get_dummy_inputs(snake_case      )
         __magic_name__			:		List[str]        =     sd_pipe(**snake_case      ).images
         __magic_name__			:		Tuple        =     image[0, -3:, -3:, -1]
         assert image.shape == (1, 64, 64, 3)

         __magic_name__			:		int        =     np.array([0.4773, 0.5417, 0.4723, 0.4925, 0.5631, 0.4752, 0.5240, 0.4935, 0.5023]      )

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



  def 							_UpperCAmelCase      ( self					: List[Any]      )	->			int:


         '''simple docstring'''





         __magic_name__			:		Any        =     '''cpu'''  # ensure determinism for the device-dependent torch.Generator
         __magic_name__			:		Optional[Any]        =     self.get_dummy_components()
         torch.manual_seed(0      )
         __magic_name__			:		Dict        =     DDPMParallelScheduler()
         torch.manual_seed(0      )
         __magic_name__			:		Union[str, Any]        =     StableDiffusionParadigmsPipeline(**snake_case      )
         __magic_name__			:		Union[str, Any]        =     sd_pipe.to(snake_case      )
         sd_pipe.set_progress_bar_config(disable=snake_case      )

         __magic_name__			:		int        =     self.get_dummy_inputs(snake_case      )
         __magic_name__			:		Optional[Any]        =     sd_pipe(**snake_case      ).images
         __magic_name__			:		int        =     image[0, -3:, -3:, -1]
         assert image.shape == (1, 64, 64, 3)

         __magic_name__			:		Union[str, Any]        =     np.array([0.3573, 0.4420, 0.4960, 0.4799, 0.3796, 0.3879, 0.4819, 0.4365, 0.4468]      )

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



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


         '''simple docstring'''





         super().test_inference_batch_consistent(batch_sizes=[1, 2]      )



  def 							_UpperCAmelCase      ( self					: int      )	->			Any:


         '''simple docstring'''





         super().test_inference_batch_single_identical(batch_size=2       ,       expected_max_diff=3e-3      )




  def 							_UpperCAmelCase      ( self					: List[str]      )	->			Any:


         '''simple docstring'''





         __magic_name__			:		Tuple        =     '''cpu'''  # ensure determinism for the device-dependent torch.Generator
         __magic_name__			:		Tuple        =     self.get_dummy_components()
         __magic_name__			:		List[Any]        =     StableDiffusionParadigmsPipeline(**snake_case      )
         __magic_name__			:		str        =     sd_pipe.to(snake_case      )
         sd_pipe.set_progress_bar_config(disable=snake_case      )

         __magic_name__			:		Any        =     self.get_dummy_inputs(snake_case      )
         __magic_name__			:		Union[str, Any]        =     '''french fries'''
         __magic_name__			:		int        =     sd_pipe(**snake_case       ,       negative_prompt=snake_case      )
         __magic_name__			:		Union[str, Any]        =     output.images
         __magic_name__			:		int        =     image[0, -3:, -3:, -1]

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

         __magic_name__			:		Optional[int]        =     np.array([0.4771, 0.5420, 0.4683, 0.4918, 0.5636, 0.4725, 0.5230, 0.4923, 0.5015]      )

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

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


  """simple docstring"""





  def 							_UpperCAmelCase      ( self					: Tuple      )	->			int:


         '''simple docstring'''





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



  def 							_UpperCAmelCase      ( self					: int       ,       snake_case					: Dict=0      )	->			List[str]:


         '''simple docstring'''





         __magic_name__			:		str        =     torch.Generator(device=snake_case      ).manual_seed(snake_case      )
         __magic_name__			:		Dict        =     {
             '''prompt''': '''a photograph of an astronaut riding a horse''',
             '''generator''': generator,
             '''num_inference_steps''': 10,
             '''guidance_scale''': 7.5,
             '''output_type''': '''numpy''',
             '''parallel''': 3,
             '''debug''': True,
         }
         return inputs




  def 							_UpperCAmelCase      ( self					: Dict      )	->			Optional[int]:


         '''simple docstring'''





         __magic_name__			:		int        =     '''stabilityai/stable-diffusion-2-base'''
         __magic_name__			:		int        =     DDIMParallelScheduler.from_pretrained(snake_case       ,       subfolder='''scheduler'''      )
         __magic_name__			:		str        =     StableDiffusionParadigmsPipeline.from_pretrained(snake_case       ,       scheduler=snake_case       ,       safety_checker=snake_case      )
         pipe.to(snake_case      )
         pipe.set_progress_bar_config(disable=snake_case      )
         pipe.enable_attention_slicing()

         __magic_name__			:		Optional[Any]        =     self.get_inputs()
         __magic_name__			:		str        =     pipe(**snake_case      ).images
         __magic_name__			:		List[Any]        =     image[0, -3:, -3:, -1].flatten()

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

         __magic_name__			:		Dict        =     np.array([0.9622, 0.9602, 0.9748, 0.9591, 0.9630, 0.9691, 0.9661, 0.9631, 0.9741]      )

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



"""simple docstring"""
import argparse

from transformers import TaConfig, TaForConditionalGeneration, load_tf_weights_in_ta
from transformers.utils import logging


logging.set_verbosity_info()







def       UpperCamelCase_					(     lowerCamelCase							:					List[str]     , lowerCamelCase							:					Optional[int]     , lowerCamelCase							:					List[str]   )						->     Optional[int]:



       """simple docstring"""







       __magic_name__			:		int        =     TaConfig.from_json_file(lowerCamelCase   )
       print(f"""Building PyTorch model from configuration: {config}"""   )
       __magic_name__			:		Union[str, Any]        =     TaForConditionalGeneration(lowerCamelCase   )

       # Load weights from tf checkpoint
       load_tf_weights_in_ta(lowerCamelCase     , lowerCamelCase     , lowerCamelCase   )

       # Save pytorch-model
       print(f"""Save PyTorch model to {pytorch_dump_path}"""   )
       model.save_pretrained(lowerCamelCase   )


if __name__ == "__main__":
 A									=						argparse.ArgumentParser()
 # Required parameters
 parser.add_argument(
     """--tf_checkpoint_path""", default=None, type=str, required=True, help="""Path to the TensorFlow checkpoint path."""
 )
 parser.add_argument(
     """--config_file""",
     default=None,
     type=str,
     required=True,
     help=(
         """The config json file corresponding to the pre-trained T5 model. \nThis specifies the model architecture."""
     ),
 )
 parser.add_argument(
     """--pytorch_dump_path""", default=None, type=str, required=True, help="""Path to the output PyTorch model."""
 )
 A									=						parser.parse_args()
 convert_tf_checkpoint_to_pytorch(args.tf_checkpoint_path, args.config_file, args.pytorch_dump_path)
