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

					@property
					def        _lowerCAmelCase	(			self      :					Optional[int]					):
											return 32

					@property
					def        _lowerCAmelCase	(			self      :					Optional[int]					):
											return 32

					@property
					def        _lowerCAmelCase	(			self      :					Optional[int]					):
											return self.time_input_dim

					@property
					def        _lowerCAmelCase	(			self      :					Optional[int]					):
											return self.time_input_dim * 4

					@property
					def        _lowerCAmelCase	(			self      :					Optional[Any]					):
											return 100

					@property
					def        _lowerCAmelCase	(			self      :					Optional[int]					):
											lowercase       :    Optional[int]											=  CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''					)
											return tokenizer

					@property
					def        _lowerCAmelCase	(			self      :					List[str]					):
											torch.manual_seed(0					)
											lowercase       :    Optional[Any]											=  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        _lowerCAmelCase	(			self      :					Union[str, Any]					):
											torch.manual_seed(0					)

											lowercase       :    int											=  {
											    '''num_attention_heads''': 2,
											    '''attention_head_dim''': 12,
											    '''embedding_dim''': self.text_embedder_hidden_size,
											    '''num_layers''': 1,
											}

											lowercase       :    Union[str, Any]											=  PriorTransformer(**lowerCAmelCase					)
											return model

					@property
					def        _lowerCAmelCase	(			self      :					Any					):
											torch.manual_seed(0					)

											lowercase       :    Optional[int]											=  {
											    '''clip_embeddings_dim''': self.text_embedder_hidden_size,
											    '''time_embed_dim''': self.time_embed_dim,
											    '''cross_attention_dim''': self.cross_attention_dim,
											}

											lowercase       :    Optional[int]											=  UnCLIPTextProjModel(**lowerCAmelCase					)
											return model

					@property
					def        _lowerCAmelCase	(			self      :					List[str]					):
											torch.manual_seed(0					)

											lowercase       :    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''',
											}

											lowercase       :    List[Any]											=  UNetaDConditionModel(**lowerCAmelCase					)
											return model

					@property
					def        _lowerCAmelCase	(			self      :					Dict					):
											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        _lowerCAmelCase	(			self      :					str					):
											torch.manual_seed(0					)

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

					@property
					def        _lowerCAmelCase	(			self      :					List[str]					):
											# seeded differently to get different unet than `self.dummy_super_res_first`
											torch.manual_seed(1					)

											lowercase       :    Any											=  UNetaDModel(**self.dummy_super_res_kwargs					)
											return model

					def        _lowerCAmelCase	(			self      :					Tuple					):
											lowercase       :    int											=  self.dummy_prior
											lowercase       :    Union[str, Any]											=  self.dummy_decoder
											lowercase       :    Optional[int]											=  self.dummy_text_proj
											lowercase       :    str											=  self.dummy_text_encoder
											lowercase       :    List[str]											=  self.dummy_tokenizer
											lowercase       :    List[Any]											=  self.dummy_super_res_first
											lowercase       :    int											=  self.dummy_super_res_last

											lowercase       :    List[Any]											=  UnCLIPScheduler(
											    variance_type='''fixed_small_log''' ,      prediction_type='''sample''' ,      num_train_timesteps=1000 ,      clip_sample_range=5.0 ,      )

											lowercase       :    List[str]											=  UnCLIPScheduler(
											    variance_type='''learned_range''' ,      prediction_type='''epsilon''' ,      num_train_timesteps=1000 ,      )

											lowercase       :    str											=  UnCLIPScheduler(
											    variance_type='''fixed_small_log''' ,      prediction_type='''epsilon''' ,      num_train_timesteps=1000 ,      )

											lowercase       :    Tuple											=  {
											    '''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        _lowerCAmelCase	(			self      :					List[Any] ,      lowerCAmelCase      :					Optional[Any] ,      lowerCAmelCase      :					List[Any]=0					):
											if str(lowerCAmelCase					).startswith('''mps'''					):
																	lowercase       :    int											=  torch.manual_seed(lowerCAmelCase					)
											else:
																	lowercase       :    List[Any]											=  torch.Generator(device=lowerCAmelCase					).manual_seed(lowerCAmelCase					)
											lowercase       :    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        _lowerCAmelCase	(			self      :					int					):
											lowercase       :    Union[str, Any]											=  '''cpu'''

											lowercase       :    Optional[Any]											=  self.get_dummy_components()

											lowercase       :    Tuple											=  self.pipeline_class(**lowerCAmelCase					)
											lowercase       :    Union[str, Any]											=  pipe.to(lowerCAmelCase					)

											pipe.set_progress_bar_config(disable=lowerCAmelCase					)

											lowercase       :    Optional[int]											=  pipe(**self.get_dummy_inputs(lowerCAmelCase					)					)
											lowercase       :    int											=  output.images

											lowercase       :    List[str]											=  pipe(
											    **self.get_dummy_inputs(lowerCAmelCase					) ,      return_dict=lowerCAmelCase ,      )[0]

											lowercase       :    List[Any]											=  image[0, -3:, -3:, -1]
											lowercase       :    Dict											=  image_from_tuple[0, -3:, -3:, -1]

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

											lowercase       :    Dict											=  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        _lowerCAmelCase	(			self      :					Optional[Any]					):
											lowercase       :    Dict											=  torch.device('''cpu'''					)







											class 							UpperCAmelCase	:
																a__: List[Any]			  =				1





											lowercase       :    Any											=  self.get_dummy_components()

											lowercase       :    Optional[Any]											=  self.pipeline_class(**lowerCAmelCase					)
											lowercase       :    Union[str, Any]											=  pipe.to(lowerCAmelCase					)

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

											lowercase       :    Dict											=  torch.Generator(device=lowerCAmelCase					).manual_seed(0					)
											lowercase       :    List[Any]											=  prior.dtype
											lowercase       :    List[str]											=  1

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

											lowercase       :    Union[str, Any]											=  (
											    batch_size,
											    super_res_first.config.in_channels // 2,
											    super_res_first.config.sample_size,
											    super_res_first.config.sample_size,
											)
											lowercase       :    Union[str, Any]											=  pipe.prepare_latents(
											    lowerCAmelCase ,      dtype=lowerCAmelCase ,      device=lowerCAmelCase ,      generator=lowerCAmelCase ,      latents=lowerCAmelCase ,      scheduler=DummyScheduler()					)

											pipe.set_progress_bar_config(disable=lowerCAmelCase					)

											lowercase       :    Any											=  '''this is a prompt example'''

											lowercase       :    Union[str, Any]											=  torch.Generator(device=lowerCAmelCase					).manual_seed(0					)
											lowercase       :    Optional[int]											=  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''' ,      )
											lowercase       :    Union[str, Any]											=  output.images

											lowercase       :    Dict											=  tokenizer(
											    lowerCAmelCase ,      padding='''max_length''' ,      max_length=tokenizer.model_max_length ,      return_tensors='''pt''' ,      )
											lowercase       :    Tuple											=  text_encoder(text_inputs.input_ids					)
											lowercase       :    Any											=  text_inputs.attention_mask

											lowercase       :    Dict											=  torch.Generator(device=lowerCAmelCase					).manual_seed(0					)
											lowercase       :    List[Any]											=  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        _lowerCAmelCase	(			self      :					Union[str, Any]					):
											lowercase       :    int											=  torch_device == '''cpu'''

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

					@skip_mps
					def        _lowerCAmelCase	(			self      :					Optional[int]					):
											lowercase       :    int											=  torch_device == '''cpu'''
											lowercase       :    str											=  True
											lowercase       :    List[Any]											=  [
											    '''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        _lowerCAmelCase	(			self      :					int					):
											lowercase       :    Union[str, 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
																	lowercase       :    str											=  [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        _lowerCAmelCase	(			self      :					List[Any]					):
											return super().test_dict_tuple_outputs_equivalent()

					@skip_mps
					def        _lowerCAmelCase	(			self      :					Optional[int]					):
											return super().test_save_load_local()




					@skip_mps
					def        _lowerCAmelCase	(			self      :					str					):
											return super().test_save_load_optional_components()







@nightly
class 							UpperCAmelCase	(   unittest.TestCase					):

					def        _lowerCAmelCase	(			self      :					Tuple					):
											# clean up the VRAM after each test
											super().tearDown()
											gc.collect()
											torch.cuda.empty_cache()




					def        _lowerCAmelCase	(			self      :					Dict					):
											lowercase       :    Optional[Any]											=  load_numpy(
											    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
											    '''/unclip/karlo_v1_alpha_horse_cpu.npy'''					)

											lowercase       :    Optional[int]											=  UnCLIPPipeline.from_pretrained('''kakaobrain/karlo-v1-alpha'''					)
											pipeline.set_progress_bar_config(disable=lowerCAmelCase					)

											lowercase       :    Optional[int]											=  torch.manual_seed(0					)
											lowercase       :    List[Any]											=  pipeline(
											    '''horse''' ,      num_images_per_prompt=1 ,      generator=lowerCAmelCase ,      output_type='''np''' ,      )

											lowercase       :    List[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					):

					def        _lowerCAmelCase	(			self      :					int					):
											# clean up the VRAM after each test
											super().tearDown()
											gc.collect()
											torch.cuda.empty_cache()

					def        _lowerCAmelCase	(			self      :					int					):
											lowercase       :    Optional[Any]											=  load_numpy(
											    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
											    '''/unclip/karlo_v1_alpha_horse_fp16.npy'''					)

											lowercase       :    Any											=  UnCLIPPipeline.from_pretrained('''kakaobrain/karlo-v1-alpha''' ,      torch_dtype=torch.floataa					)
											lowercase       :    str											=  pipeline.to(lowerCAmelCase					)
											pipeline.set_progress_bar_config(disable=lowerCAmelCase					)

											lowercase       :    int											=  torch.Generator(device='''cpu'''					).manual_seed(0					)
											lowercase       :    Dict											=  pipeline(
											    '''horse''' ,      generator=lowerCAmelCase ,      output_type='''np''' ,      )

											lowercase       :    List[str]											=  output.images[0]

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

											assert_mean_pixel_difference(lowerCAmelCase ,      lowerCAmelCase					)




					def        _lowerCAmelCase	(			self      :					Tuple					):
											torch.cuda.empty_cache()
											torch.cuda.reset_max_memory_allocated()
											torch.cuda.reset_peak_memory_stats()

											lowercase       :    Tuple											=  UnCLIPPipeline.from_pretrained('''kakaobrain/karlo-v1-alpha''' ,      torch_dtype=torch.floataa					)
											lowercase       :    int											=  pipe.to(lowerCAmelCase					)
											pipe.set_progress_bar_config(disable=lowerCAmelCase					)
											pipe.enable_attention_slicing()
											pipe.enable_sequential_cpu_offload()

											lowercase       :    int											=  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''' ,      )

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




from itertools import combinations_with_replacement

def        lowerCamelCase_      (			UpperCAmelCase_							:				int = 1_00       ):
						lowercase       :    list[int]											=  [*list(range(1							,			21       )       ), 25]
						lowercase       :    list[int]											=  [2 * x for x in range(1							,			21       )] + [50]
						lowercase       :    list[int]											=  [3 * x for x in range(1							,			21       )]
						lowercase       :    list[int]											=  singles + doubles + triples + [0]

						lowercase       :    int											=  0
						lowercase       :    int
						lowercase       :    int
						lowercase       :    int
						lowercase       :    int

						for double in doubles:
												for throwa, throwa in combinations_with_replacement(UpperCAmelCase_							,			2       ):
																		lowercase       :    Union[str, Any]											=  double + throwa + throwa
																		if checkout_total < limit:
																								num_checkouts += 1

						return num_checkouts


if __name__ == "__main__":
				print(F'{solution() = }')


