
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 	lowerCamelCase						(    A_				,					unittest.TestCase    ):
     UpperCAmelCase__							:				int 	=						UnCLIPPipeline
     UpperCAmelCase__							:				str 	=						TEXT_TO_IMAGE_PARAMS - {
         "negative_prompt",
         "height",
         "width",
         "negative_prompt_embeds",
         "guidance_scale",
         "prompt_embeds",
         "cross_attention_kwargs",
     }
     UpperCAmelCase__							:				Dict 	=						TEXT_TO_IMAGE_BATCH_PARAMS
     UpperCAmelCase__							:				Dict 	=						[
         "generator",
         "return_dict",
         "prior_num_inference_steps",
         "decoder_num_inference_steps",
         "super_res_num_inference_steps",
     ]
     UpperCAmelCase__							:				Any 	=						False





     @property
     def 						UpperCAmelCase(self :       Optional[Any]  )		->		List[Any]:
         return 3_2





     @property
     def 						UpperCAmelCase(self :       int  )		->		str:
         return 3_2





     @property
     def 						UpperCAmelCase(self :       int  )		->		List[Any]:
         return self.time_input_dim





     @property
     def 						UpperCAmelCase(self :       Optional[Any]  )		->		Optional[int]:
         return self.time_input_dim * 4





     @property
     def 						UpperCAmelCase(self :       int  )		->		List[str]:
         return 1_0_0





     @property
     def 						UpperCAmelCase(self :       Tuple  )		->		Optional[Any]:
         snake_case						=   CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"  )
         return tokenizer





     @property
     def 						UpperCAmelCase(self :       Dict  )		->		Union[str, Any]:
         torch.manual_seed(0  )
         snake_case						=   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=3_7							,       layer_norm_eps=1E-05							,       num_attention_heads=4							,       num_hidden_layers=5							,       pad_token_id=1							,       vocab_size=1_0_0_0							,       )
         return CLIPTextModelWithProjection(_A  )





     @property
     def 						UpperCAmelCase(self :       int  )		->		Dict:
         torch.manual_seed(0  )

         snake_case						=   {
             "num_attention_heads": 2,
             "attention_head_dim": 1_2,
             "embedding_dim": self.text_embedder_hidden_size,
             "num_layers": 1,
         }

         snake_case						=   PriorTransformer(**_A  )
         return model





     @property
     def 						UpperCAmelCase(self :       Dict  )		->		Union[str, Any]:
         torch.manual_seed(0  )

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

         snake_case						=   UnCLIPTextProjModel(**_A  )
         return model





     @property
     def 						UpperCAmelCase(self :       Optional[Any]  )		->		Any:
         torch.manual_seed(0  )

         snake_case						=   {
             "sample_size": 3_2,
             # 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",
         }

         snake_case						=   UNetaDConditionModel(**_A  )
         return model





     @property
     def 						UpperCAmelCase(self :       List[Any]  )		->		Optional[Any]:
         return {
             "sample_size": 6_4,
             "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 						UpperCAmelCase(self :       str  )		->		Optional[int]:
         torch.manual_seed(0  )

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





     @property
     def 						UpperCAmelCase(self :       Dict  )		->		Optional[Any]:
         torch.manual_seed(1  )

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





     def 						UpperCAmelCase(self :       Optional[Any]  )		->		Any:
         snake_case						=   self.dummy_prior
         snake_case						=   self.dummy_decoder
         snake_case						=   self.dummy_text_proj
         snake_case						=   self.dummy_text_encoder
         snake_case						=   self.dummy_tokenizer
         snake_case						=   self.dummy_super_res_first
         snake_case						=   self.dummy_super_res_last

         snake_case						=   UnCLIPScheduler(
             variance_type="fixed_small_log"							,       prediction_type="sample"							,       num_train_timesteps=1_0_0_0							,       clip_sample_range=5.0							,       )

         snake_case						=   UnCLIPScheduler(
             variance_type="learned_range"							,       prediction_type="epsilon"							,       num_train_timesteps=1_0_0_0							,       )

         snake_case						=   UnCLIPScheduler(
             variance_type="fixed_small_log"							,       prediction_type="epsilon"							,       num_train_timesteps=1_0_0_0							,       )

         snake_case						=   {
             "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 						UpperCAmelCase(self :       List[str]							,       _A :       Dict							,       _A :       Dict=0  )		->		Dict:







         if str(_A  ).startswith("mps"  ):
             snake_case						=   torch.manual_seed(_A  )
         else:
             snake_case						=   torch.Generator(device=_A  ).manual_seed(_A  )
         snake_case						=   {
             "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 						UpperCAmelCase(self :       Optional[int]  )		->		Optional[Any]:
         snake_case						=   "cpu"

         snake_case						=   self.get_dummy_components()

         snake_case						=   self.pipeline_class(**_A  )
         snake_case						=   pipe.to(_A  )

         pipe.set_progress_bar_config(disable=_A  )

         snake_case						=   pipe(**self.get_dummy_inputs(_A  )  )
         snake_case						=   output.images

         snake_case						=   pipe(
             **self.get_dummy_inputs(_A  )							,       return_dict=_A							,       )[0]

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

         assert image.shape == (1, 6_4, 6_4, 3)

         snake_case						=   np.array(
             [
                 0.99_97,
                 0.99_88,
                 0.00_28,
                 0.99_97,
                 0.99_84,
                 0.99_65,
                 0.00_29,
                 0.99_86,
                 0.00_25,
             ]  )

         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 						UpperCAmelCase(self :       int  )		->		Dict:
         snake_case						=   torch.device("cpu"  )







         class 	lowerCamelCase						:
              UpperCAmelCase__							:				Dict 	=						1


         snake_case						=   self.get_dummy_components()

         snake_case						=   self.pipeline_class(**_A  )
         snake_case						=   pipe.to(_A  )

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

         snake_case						=   torch.Generator(device=_A  ).manual_seed(0  )
         snake_case						=   prior.dtype
         snake_case						=   1

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

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

         pipe.set_progress_bar_config(disable=_A  )

         snake_case						=   "this is a prompt example"

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

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

         snake_case						=   torch.Generator(device=_A  ).manual_seed(0  )
         snake_case						=   pipe(
             generator=_A							,       prior_num_inference_steps=2							,       decoder_num_inference_steps=2							,       super_res_num_inference_steps=2							,       prior_latents=_A							,       decoder_latents=_A							,       super_res_latents=_A							,       text_model_output=_A							,       text_attention_mask=_A							,       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 						UpperCAmelCase(self :       Optional[Any]  )		->		List[str]:
         snake_case						=   torch_device == "cpu"

         self._test_attention_slicing_forward_pass(test_max_difference=_A							,       expected_max_diff=0.01  )





     @skip_mps
     def 						UpperCAmelCase(self :       Dict  )		->		List[str]:
         snake_case						=   torch_device == "cpu"
         snake_case						=   True
         snake_case						=   [
             "prior_num_inference_steps",
             "decoder_num_inference_steps",
             "super_res_num_inference_steps",
         ]

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





     def 						UpperCAmelCase(self :       Optional[int]  )		->		str:
         snake_case						=   [
             "prior_num_inference_steps",
             "decoder_num_inference_steps",
             "super_res_num_inference_steps",
         ]

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





     @skip_mps
     def 						UpperCAmelCase(self :       Any  )		->		Dict:
         return super().test_dict_tuple_outputs_equivalent()





     @skip_mps
     def 						UpperCAmelCase(self :       List[Any]  )		->		Optional[Any]:
         return super().test_save_load_local()





     @skip_mps
     def 						UpperCAmelCase(self :       List[str]  )		->		Optional[Any]:
         return super().test_save_load_optional_components()







@nightly
class 	lowerCamelCase						(    unittest.TestCase    ):





     def 						UpperCAmelCase(self :       Tuple  )		->		Dict:
         super().tearDown()
         gc.collect()
         torch.cuda.empty_cache()





     def 						UpperCAmelCase(self :       Optional[Any]  )		->		Optional[int]:
         snake_case						=   load_numpy(
             "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
             "/unclip/karlo_v1_alpha_horse_cpu.npy"  )

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

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

         snake_case						=   output.images[0]

         assert image.shape == (2_5_6, 2_5_6, 3)
         assert np.abs(expected_image - image  ).max() < 1E-1







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





     def 						UpperCAmelCase(self :       Optional[int]  )		->		List[str]:
         super().tearDown()
         gc.collect()
         torch.cuda.empty_cache()





     def 						UpperCAmelCase(self :       List[Any]  )		->		Tuple:
         snake_case						=   load_numpy(
             "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
             "/unclip/karlo_v1_alpha_horse_fp16.npy"  )

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

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

         snake_case						=   output.images[0]

         assert image.shape == (2_5_6, 2_5_6, 3)

         assert_mean_pixel_difference(_A							,       _A  )





     def 						UpperCAmelCase(self :       Union[str, Any]  )		->		Optional[int]:
         torch.cuda.empty_cache()
         torch.cuda.reset_max_memory_allocated()
         torch.cuda.reset_peak_memory_stats()

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

         snake_case						=   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"							,       )

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









'''simple docstring'''


def 						__magic_name__(			_A       ):
						'''simple docstring'''





						UpperCamelCase__       =			len(_A       )

						# Set the gap value to be decreased by a factor of 1.3
						# after each iteration
						UpperCamelCase__       =			1.3

						# Continue sorting until the gap is 1
						while gap > 1:
												# Decrease the gap value
												UpperCamelCase__       =			int(gap / shrink       )

												# Sort the elements using insertion sort
												for i in range(_A    ,   len(_A       )       ):
																		UpperCamelCase__       =			collection[i]
																		UpperCamelCase__       =			i
																		while j >= gap and collection[j - gap] > temp:
																								UpperCamelCase__       =			collection[j - gap]
																								j -= gap
																		UpperCamelCase__       =			temp

						return collection


if __name__ == "__main__":
					import doctest

					doctest.testmod()
