


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  UpperCAmelCase_						(				_a,  unittest.TestCase):
       '''simple docstring'''
       __UpperCamelCase   :  Any				=       UnCLIPPipeline
       __UpperCamelCase   :  Optional[Any]				=       TEXT_TO_IMAGE_PARAMS - {
           "negative_prompt",
           "height",
           "width",
           "negative_prompt_embeds",
           "guidance_scale",
           "prompt_embeds",
           "cross_attention_kwargs",
       }
       __UpperCamelCase   :  str				=       TEXT_TO_IMAGE_BATCH_PARAMS
       __UpperCamelCase   :  Optional[int]				=       [
           "generator",
           "return_dict",
           "prior_num_inference_steps",
           "decoder_num_inference_steps",
           "super_res_num_inference_steps",
       ]
       __UpperCamelCase   :  Union[str, Any]				=       False






       @property
       def 			_lowercase (  self    ):




              """simple docstring"""




              return 32






       @property
       def 			_lowercase (  self    ):




              """simple docstring"""




              return 32






       @property
       def 			_lowercase (  self    ):




              """simple docstring"""




              return self.time_input_dim






       @property
       def 			_lowercase (  self    ):




              """simple docstring"""




              return self.time_input_dim * 4






       @property
       def 			_lowercase (  self    ):




              """simple docstring"""




              return 100






       @property
       def 			_lowercase (  self    ):




              """simple docstring"""




              UpperCamelCase     :						Any 						=		CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''    )
              return tokenizer






       @property
       def 			_lowercase (  self    ):




              """simple docstring"""




              torch.manual_seed(0    )
              UpperCamelCase     :						Dict 						=		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=1_000			,       )
              return CLIPTextModelWithProjection(__SCREAMING_SNAKE_CASE    )






       @property
       def 			_lowercase (  self    ):




              """simple docstring"""




              torch.manual_seed(0    )

              UpperCamelCase     :						Union[str, Any] 						=		{
                  '''num_attention_heads''': 2,
                  '''attention_head_dim''': 12,
                  '''embedding_dim''': self.text_embedder_hidden_size,
                  '''num_layers''': 1,
              }

              UpperCamelCase     :						Tuple 						=		PriorTransformer(**__SCREAMING_SNAKE_CASE    )
              return model






       @property
       def 			_lowercase (  self    ):




              """simple docstring"""




              torch.manual_seed(0    )

              UpperCamelCase     :						Tuple 						=		{
                  '''clip_embeddings_dim''': self.text_embedder_hidden_size,
                  '''time_embed_dim''': self.time_embed_dim,
                  '''cross_attention_dim''': self.cross_attention_dim,
              }

              UpperCamelCase     :						Optional[Any] 						=		UnCLIPTextProjModel(**__SCREAMING_SNAKE_CASE    )
              return model






       @property
       def 			_lowercase (  self    ):




              """simple docstring"""




              torch.manual_seed(0    )

              UpperCamelCase     :						Union[str, Any] 						=		{
                  '''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     :						Tuple 						=		UNetaDConditionModel(**__SCREAMING_SNAKE_CASE    )
              return model






       @property
       def 			_lowercase (  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 			_lowercase (  self    ):




              """simple docstring"""




              torch.manual_seed(0    )

              UpperCamelCase     :						Tuple 						=		UNetaDModel(**self.dummy_super_res_kwargs    )
              return model






       @property
       def 			_lowercase (  self    ):




              """simple docstring"""




              torch.manual_seed(1    )

              UpperCamelCase     :						Optional[int] 						=		UNetaDModel(**self.dummy_super_res_kwargs    )
              return model






       def 			_lowercase (  self    ):




              """simple docstring"""




              UpperCamelCase     :						Any 						=		self.dummy_prior
              UpperCamelCase     :						List[str] 						=		self.dummy_decoder
              UpperCamelCase     :						Any 						=		self.dummy_text_proj
              UpperCamelCase     :						Dict 						=		self.dummy_text_encoder
              UpperCamelCase     :						Union[str, Any] 						=		self.dummy_tokenizer
              UpperCamelCase     :						Tuple 						=		self.dummy_super_res_first
              UpperCamelCase     :						Optional[Any] 						=		self.dummy_super_res_last

              UpperCamelCase     :						Optional[int] 						=		UnCLIPScheduler(
                  variance_type='''fixed_small_log'''			,       prediction_type='''sample'''			,       num_train_timesteps=1_000			,       clip_sample_range=5.0			,       )

              UpperCamelCase     :						Dict 						=		UnCLIPScheduler(
                  variance_type='''learned_range'''			,       prediction_type='''epsilon'''			,       num_train_timesteps=1_000			,       )

              UpperCamelCase     :						Union[str, Any] 						=		UnCLIPScheduler(
                  variance_type='''fixed_small_log'''			,       prediction_type='''epsilon'''			,       num_train_timesteps=1_000			,       )

              UpperCamelCase     :						List[Any] 						=		{
                  '''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 			_lowercase (  self			,       __SCREAMING_SNAKE_CASE			,       __SCREAMING_SNAKE_CASE=0    ):




              """simple docstring"""




              if str(__SCREAMING_SNAKE_CASE    ).startswith('''mps'''    ):
                     UpperCamelCase     :						Optional[Any] 						=		torch.manual_seed(__SCREAMING_SNAKE_CASE    )
              else:
                     UpperCamelCase     :						Optional[int] 						=		torch.Generator(device=__SCREAMING_SNAKE_CASE    ).manual_seed(__SCREAMING_SNAKE_CASE    )
              UpperCamelCase     :						List[str] 						=		{
                  '''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 			_lowercase (  self    ):




              """simple docstring"""




              UpperCamelCase     :						Optional[Any] 						=		'''cpu'''

              UpperCamelCase     :						Tuple 						=		self.get_dummy_components()

              UpperCamelCase     :						Any 						=		self.pipeline_class(**__SCREAMING_SNAKE_CASE    )
              UpperCamelCase     :						Optional[Any] 						=		pipe.to(__SCREAMING_SNAKE_CASE    )

              pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE    )

              UpperCamelCase     :						Optional[Any] 						=		pipe(**self.get_dummy_inputs(__SCREAMING_SNAKE_CASE    )    )
              UpperCamelCase     :						Dict 						=		output.images

              UpperCamelCase     :						str 						=		pipe(
                  **self.get_dummy_inputs(__SCREAMING_SNAKE_CASE    )			,       return_dict=__SCREAMING_SNAKE_CASE			,       )[0]

              UpperCamelCase     :						List[str] 						=		image[0, -3:, -3:, -1]
              UpperCamelCase     :						Optional[Any] 						=		image_from_tuple[0, -3:, -3:, -1]

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

              UpperCamelCase     :						str 						=		np.array(
                  [
                      0.9_997,
                      0.9_988,
                      0.0_028,
                      0.9_997,
                      0.9_984,
                      0.9_965,
                      0.0_029,
                      0.9_986,
                      0.0_025,
                  ]    )

              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 			_lowercase (  self    ):




              """simple docstring"""




              UpperCamelCase     :						List[str] 						=		torch.device('''cpu'''    )



              class  UpperCAmelCase_						:
                     '''simple docstring'''
                     __UpperCamelCase   :  List[Any]				=       1




              UpperCamelCase     :						Optional[int] 						=		self.get_dummy_components()

              UpperCamelCase     :						Any 						=		self.pipeline_class(**__SCREAMING_SNAKE_CASE    )
              UpperCamelCase     :						Any 						=		pipe.to(__SCREAMING_SNAKE_CASE    )

              UpperCamelCase     :						List[str] 						=		components['''prior''']
              UpperCamelCase     :						List[Any] 						=		components['''decoder''']
              UpperCamelCase     :						Optional[Any] 						=		components['''super_res_first''']
              UpperCamelCase     :						int 						=		components['''tokenizer''']
              UpperCamelCase     :						Optional[Any] 						=		components['''text_encoder''']

              UpperCamelCase     :						int 						=		torch.Generator(device=__SCREAMING_SNAKE_CASE    ).manual_seed(0    )
              UpperCamelCase     :						int 						=		prior.dtype
              UpperCamelCase     :						Optional[int] 						=		1

              UpperCamelCase     :						List[str] 						=		(batch_size, prior.config.embedding_dim)
              UpperCamelCase     :						int 						=		pipe.prepare_latents(
                  __SCREAMING_SNAKE_CASE			,       dtype=__SCREAMING_SNAKE_CASE			,       device=__SCREAMING_SNAKE_CASE			,       generator=__SCREAMING_SNAKE_CASE			,       latents=__SCREAMING_SNAKE_CASE			,       scheduler=DummyScheduler()    )
              UpperCamelCase     :						Any 						=		(batch_size, decoder.config.in_channels, decoder.config.sample_size, decoder.config.sample_size)
              UpperCamelCase     :						Optional[int] 						=		pipe.prepare_latents(
                  __SCREAMING_SNAKE_CASE			,       dtype=__SCREAMING_SNAKE_CASE			,       device=__SCREAMING_SNAKE_CASE			,       generator=__SCREAMING_SNAKE_CASE			,       latents=__SCREAMING_SNAKE_CASE			,       scheduler=DummyScheduler()    )

              UpperCamelCase     :						Optional[Any] 						=		(
                  batch_size,
                  super_res_first.config.in_channels // 2,
                  super_res_first.config.sample_size,
                  super_res_first.config.sample_size,
              )
              UpperCamelCase     :						List[str] 						=		pipe.prepare_latents(
                  __SCREAMING_SNAKE_CASE			,       dtype=__SCREAMING_SNAKE_CASE			,       device=__SCREAMING_SNAKE_CASE			,       generator=__SCREAMING_SNAKE_CASE			,       latents=__SCREAMING_SNAKE_CASE			,       scheduler=DummyScheduler()    )

              pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE    )

              UpperCamelCase     :						str 						=		'''this is a prompt example'''

              UpperCamelCase     :						List[Any] 						=		torch.Generator(device=__SCREAMING_SNAKE_CASE    ).manual_seed(0    )
              UpperCamelCase     :						Union[str, Any] 						=		pipe(
                  [prompt]			,       generator=__SCREAMING_SNAKE_CASE			,       prior_num_inference_steps=2			,       decoder_num_inference_steps=2			,       super_res_num_inference_steps=2			,       prior_latents=__SCREAMING_SNAKE_CASE			,       decoder_latents=__SCREAMING_SNAKE_CASE			,       super_res_latents=__SCREAMING_SNAKE_CASE			,       output_type='''np'''			,       )
              UpperCamelCase     :						List[Any] 						=		output.images

              UpperCamelCase     :						Dict 						=		tokenizer(
                  __SCREAMING_SNAKE_CASE			,       padding='''max_length'''			,       max_length=tokenizer.model_max_length			,       return_tensors='''pt'''			,       )
              UpperCamelCase     :						Optional[Any] 						=		text_encoder(text_inputs.input_ids    )
              UpperCamelCase     :						Dict 						=		text_inputs.attention_mask

              UpperCamelCase     :						List[Any] 						=		torch.Generator(device=__SCREAMING_SNAKE_CASE    ).manual_seed(0    )
              UpperCamelCase     :						Union[str, Any] 						=		pipe(
                  generator=__SCREAMING_SNAKE_CASE			,       prior_num_inference_steps=2			,       decoder_num_inference_steps=2			,       super_res_num_inference_steps=2			,       prior_latents=__SCREAMING_SNAKE_CASE			,       decoder_latents=__SCREAMING_SNAKE_CASE			,       super_res_latents=__SCREAMING_SNAKE_CASE			,       text_model_output=__SCREAMING_SNAKE_CASE			,       text_attention_mask=__SCREAMING_SNAKE_CASE			,       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 			_lowercase (  self    ):




              """simple docstring"""




              UpperCamelCase     :						Any 						=		torch_device == '''cpu'''

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






       @skip_mps
       def 			_lowercase (  self    ):




              """simple docstring"""




              UpperCamelCase     :						List[Any] 						=		torch_device == '''cpu'''
              UpperCamelCase     :						Optional[Any] 						=		True
              UpperCamelCase     :						Dict 						=		[
                  '''prior_num_inference_steps''',
                  '''decoder_num_inference_steps''',
                  '''super_res_num_inference_steps''',
              ]

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






       def 			_lowercase (  self    ):




              """simple docstring"""




              UpperCamelCase     :						Any 						=		[
                  '''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     :						Tuple 						=		[2, 3]
                     self._test_inference_batch_consistent(
                         batch_sizes=__SCREAMING_SNAKE_CASE			,       additional_params_copy_to_batched_inputs=__SCREAMING_SNAKE_CASE			,       )
              else:
                     self._test_inference_batch_consistent(
                         additional_params_copy_to_batched_inputs=__SCREAMING_SNAKE_CASE    )






       @skip_mps
       def 			_lowercase (  self    ):




              """simple docstring"""




              return super().test_dict_tuple_outputs_equivalent()






       @skip_mps
       def 			_lowercase (  self    ):




              """simple docstring"""




              return super().test_save_load_local()






       @skip_mps
       def 			_lowercase (  self    ):




              """simple docstring"""




              return super().test_save_load_optional_components()




@nightly
class  UpperCAmelCase_						(				unittest.TestCase):
       '''simple docstring'''






       def 			_lowercase (  self    ):




              """simple docstring"""




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






       def 			_lowercase (  self    ):




              """simple docstring"""




              UpperCamelCase     :						Optional[int] 						=		load_numpy(
                  '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
                  '''/unclip/karlo_v1_alpha_horse_cpu.npy'''    )

              UpperCamelCase     :						Tuple 						=		UnCLIPPipeline.from_pretrained('''kakaobrain/karlo-v1-alpha'''    )
              pipeline.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE    )

              UpperCamelCase     :						Any 						=		torch.manual_seed(0    )
              UpperCamelCase     :						Any 						=		pipeline(
                  '''horse'''			,       num_images_per_prompt=1			,       generator=__SCREAMING_SNAKE_CASE			,       output_type='''np'''			,       )

              UpperCamelCase     :						Any 						=		output.images[0]

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




@slow
@require_torch_gpu
class  UpperCAmelCase_						(				unittest.TestCase):
       '''simple docstring'''






       def 			_lowercase (  self    ):




              """simple docstring"""




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






       def 			_lowercase (  self    ):




              """simple docstring"""




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

              UpperCamelCase     :						Any 						=		UnCLIPPipeline.from_pretrained('''kakaobrain/karlo-v1-alpha'''			,       torch_dtype=torch.floataa    )
              UpperCamelCase     :						Dict 						=		pipeline.to(__SCREAMING_SNAKE_CASE    )
              pipeline.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE    )

              UpperCamelCase     :						Tuple 						=		torch.Generator(device='''cpu'''    ).manual_seed(0    )
              UpperCamelCase     :						List[Any] 						=		pipeline(
                  '''horse'''			,       generator=__SCREAMING_SNAKE_CASE			,       output_type='''np'''			,       )

              UpperCamelCase     :						Union[str, Any] 						=		output.images[0]

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

              assert_mean_pixel_difference(__SCREAMING_SNAKE_CASE			,       __SCREAMING_SNAKE_CASE    )






       def 			_lowercase (  self    ):




              """simple docstring"""




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

              UpperCamelCase     :						Optional[Any] 						=		UnCLIPPipeline.from_pretrained('''kakaobrain/karlo-v1-alpha'''			,       torch_dtype=torch.floataa    )
              UpperCamelCase     :						Union[str, Any] 						=		pipe.to(__SCREAMING_SNAKE_CASE    )
              pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE    )
              pipe.enable_attention_slicing()
              pipe.enable_sequential_cpu_offload()

              UpperCamelCase     :						str 						=		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     :						Optional[Any] 						=		torch.cuda.max_memory_allocated()
              # make sure that less than 7 GB is allocated
              assert mem_bytes < 7 * 10**9





def   a						(				SCREAMING_SNAKE_CASE_   :		Any							):






       """simple docstring"""





       UpperCamelCase     :						int 						=		len(SCREAMING_SNAKE_CASE_							)
       for i in range(length - 1							):
              UpperCamelCase     :						Optional[int] 						=		False
              for j in range(length - 1 - i							):
                     if collection[j] > collection[j + 1]:
                            UpperCamelCase     :						Optional[int] 						=		True
                            UpperCamelCase						,			UpperCamelCase     :						Union[str, Any] 						=		collection[j + 1], collection[j]
              if not swapped:
                     break  # Stop iteration if the collection is sorted.
       return collection


if __name__ == "__main__":
  import doctest
  import time

  doctest.testmod()

  __UpperCAmelCase			:			Any			      =					input("Enter numbers separated by a comma:").strip()
  __UpperCAmelCase			:			List[str]			      =					[int(item) for item in user_input.split(",")]
  __UpperCAmelCase			:			List[Any]			      =					time.process_time()
  print(*bubble_sort(unsorted), sep=",")
  print(f'''Processing time: {(time.process_time() - start)%1E9 + 7}''')

