

import gc
import random
import unittest

import numpy as np
import torch
from transformers import (
    CLIPImageProcessor,
    CLIPTextConfig,
    CLIPTextModelWithProjection,
    CLIPTokenizer,
    CLIPVisionConfig,
    CLIPVisionModelWithProjection,
)

from diffusers import (
    DiffusionPipeline,
    UnCLIPImageVariationPipeline,
    UnCLIPScheduler,
    UNetaDConditionModel,
    UNetaDModel,
)
from diffusers.pipelines.unclip.text_proj import UnCLIPTextProjModel
from diffusers.utils import floats_tensor, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, load_image, require_torch_gpu, skip_mps

from ..pipeline_params import IMAGE_VARIATION_BATCH_PARAMS, IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()



class 					_a		(__lowerCamelCase	,     unittest.TestCase       ):


					'''simple docstring'''






					UpperCAmelCase__:     int													= UnCLIPImageVariationPipeline
					UpperCAmelCase__:     Optional[int]													= IMAGE_VARIATION_PARAMS - {'''height''', '''width''', '''guidance_scale'''}
					UpperCAmelCase__:     Union[str, Any]													= IMAGE_VARIATION_BATCH_PARAMS

					UpperCAmelCase__:     List[str]													= [
					    '''generator''',
					    '''return_dict''',
					    '''decoder_num_inference_steps''',
					    '''super_res_num_inference_steps''',
					]
					UpperCAmelCase__:     int													= False






					@property
					def 						__A    (				self    ):
						return 32






					@property
					def 						__A    (				self    ):
						return 32






					@property
					def 						__A    (				self    ):
						return self.time_input_dim






					@property
					def 						__A    (				self    ):
						return self.time_input_dim * 4






					@property
					def 						__A    (				self    ):
						return 100






					@property
					def 						__A    (				self    ):
						A__							:					Any       =	CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""    )
						return tokenizer






					@property
					def 						__A    (				self    ):
						torch.manual_seed(0    )
						A__							:					Union[str, 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(UpperCAmelCase_    )






					@property
					def 						__A    (				self    ):
						torch.manual_seed(0    )
						A__							:					Tuple       =	CLIPVisionConfig(
						    hidden_size=self.text_embedder_hidden_size					, projection_dim=self.text_embedder_hidden_size					, num_hidden_layers=5					, num_attention_heads=4					, image_size=32					, intermediate_size=37					, patch_size=1					, )
						return CLIPVisionModelWithProjection(UpperCAmelCase_    )






					@property
					def 						__A    (				self    ):
						torch.manual_seed(0    )

						A__							:					Dict       =	{
						    'clip_embeddings_dim': self.text_embedder_hidden_size,
						    'time_embed_dim': self.time_embed_dim,
						    'cross_attention_dim': self.cross_attention_dim,
						}

						A__							:					int       =	UnCLIPTextProjModel(**UpperCAmelCase_    )
						return model






					@property
					def 						__A    (				self    ):
						torch.manual_seed(0    )

						A__							:					str       =	{
						    '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',
						}

						A__							:					Optional[Any]       =	UNetaDConditionModel(**UpperCAmelCase_    )
						return model






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

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






					@property
					def 						__A    (				self    ):
						# seeded differently to get different unet than `self.dummy_super_res_first`
						torch.manual_seed(1    )

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






					def 						__A    (				self    ):
						A__							:					Optional[int]       =	self.dummy_decoder
						A__							:					Dict       =	self.dummy_text_proj
						A__							:					int       =	self.dummy_text_encoder
						A__							:					Tuple       =	self.dummy_tokenizer
						A__							:					Optional[Any]       =	self.dummy_super_res_first
						A__							:					List[str]       =	self.dummy_super_res_last

						A__							:					Optional[Any]       =	UnCLIPScheduler(
						    variance_type="""learned_range"""					, prediction_type="""epsilon"""					, num_train_timesteps=1000					, )

						A__							:					Tuple       =	UnCLIPScheduler(
						    variance_type="""fixed_small_log"""					, prediction_type="""epsilon"""					, num_train_timesteps=1000					, )

						A__							:					List[Any]       =	CLIPImageProcessor(crop_size=32					, size=32    )

						A__							:					Optional[Any]       =	self.dummy_image_encoder

						return {
						    "decoder": decoder,
						    "text_encoder": text_encoder,
						    "tokenizer": tokenizer,
						    "text_proj": text_proj,
						    "feature_extractor": feature_extractor,
						    "image_encoder": image_encoder,
						    "super_res_first": super_res_first,
						    "super_res_last": super_res_last,
						    "decoder_scheduler": decoder_scheduler,
						    "super_res_scheduler": super_res_scheduler,
						}






					def 						__A    (				self					, A__					, A__=0					, A__=True    ):
						A__							:					Optional[int]       =	floats_tensor((1, 3, 32, 32)					, rng=random.Random(UpperCAmelCase_    )    ).to(UpperCAmelCase_    )
						if str(UpperCAmelCase_    ).startswith("""mps"""    ):
							A__							:					Optional[int]       =	torch.manual_seed(UpperCAmelCase_    )
						else:
							A__							:					str       =	torch.Generator(device=UpperCAmelCase_    ).manual_seed(UpperCAmelCase_    )

						if pil_image:
							A__							:					List[Any]       =	input_image * 0.5 + 0.5
							A__							:					str       =	input_image.clamp(0					, 1    )
							A__							:					Optional[Any]       =	input_image.cpu().permute(0					, 2					, 3					, 1    ).float().numpy()
							A__							:					int       =	DiffusionPipeline.numpy_to_pil(UpperCAmelCase_    )[0]

						return {
						    "image": input_image,
						    "generator": generator,
						    "decoder_num_inference_steps": 2,
						    "super_res_num_inference_steps": 2,
						    "output_type": "np",
						}






					def 						__A    (				self    ):
						A__							:					Any       =	'cpu'

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

						A__							:					Dict       =	self.pipeline_class(**UpperCAmelCase_    )
						A__							:					int       =	pipe.to(UpperCAmelCase_    )

						pipe.set_progress_bar_config(disable=UpperCAmelCase_    )

						A__							:					Optional[int]       =	self.get_dummy_inputs(UpperCAmelCase_					, pil_image=UpperCAmelCase_    )

						A__							:					Any       =	pipe(**UpperCAmelCase_    )
						A__							:					int       =	output.images

						A__							:					Dict       =	self.get_dummy_inputs(UpperCAmelCase_					, pil_image=UpperCAmelCase_    )

						A__							:					List[str]       =	pipe(
						    **UpperCAmelCase_					, return_dict=UpperCAmelCase_					, )[0]

						A__							:					int       =	image[0, -3:, -3:, -1]
						A__							:					Dict       =	image_from_tuple[0, -3:, -3:, -1]

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

						A__							:					int       =	np.array(
						    [
						        0.9_9_9_7,
						        0.0_0_0_2,
						        0.9_9_9_7,
						        0.9_9_9_7,
						        0.9_9_6_9,
						        0.0_0_2_3,
						        0.9_9_9_7,
						        0.9_9_6_9,
						        0.9_9_7_0,
						    ]    )

						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 						__A    (				self    ):
						A__							:					Tuple       =	'cpu'

						A__							:					Dict       =	self.get_dummy_components()

						A__							:					Any       =	self.pipeline_class(**UpperCAmelCase_    )
						A__							:					Any       =	pipe.to(UpperCAmelCase_    )

						pipe.set_progress_bar_config(disable=UpperCAmelCase_    )

						A__							:					List[str]       =	self.get_dummy_inputs(UpperCAmelCase_					, pil_image=UpperCAmelCase_    )

						A__							:					Union[str, Any]       =	pipe(**UpperCAmelCase_    )
						A__							:					Optional[Any]       =	output.images

						A__							:					Union[str, Any]       =	self.get_dummy_inputs(UpperCAmelCase_					, pil_image=UpperCAmelCase_    )

						A__							:					Dict       =	pipe(
						    **UpperCAmelCase_					, return_dict=UpperCAmelCase_					, )[0]

						A__							:					Union[str, Any]       =	image[0, -3:, -3:, -1]
						A__							:					str       =	image_from_tuple[0, -3:, -3:, -1]

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

						A__							:					Dict       =	np.array([0.9_9_9_7, 0.0_0_0_3, 0.9_9_9_7, 0.9_9_9_7, 0.9_9_7_0, 0.0_0_2_4, 0.9_9_9_7, 0.9_9_7_1, 0.9_9_7_1]    )

						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 						__A    (				self    ):
						A__							:					int       =	'cpu'

						A__							:					int       =	self.get_dummy_components()

						A__							:					Tuple       =	self.pipeline_class(**UpperCAmelCase_    )
						A__							:					Dict       =	pipe.to(UpperCAmelCase_    )

						pipe.set_progress_bar_config(disable=UpperCAmelCase_    )

						A__							:					Dict       =	self.get_dummy_inputs(UpperCAmelCase_					, pil_image=UpperCAmelCase_    )
						A__							:					Dict       =	[
						    pipeline_inputs['image'],
						    pipeline_inputs['image'],
						]

						A__							:					int       =	pipe(**UpperCAmelCase_    )
						A__							:					Optional[int]       =	output.images

						A__							:					Tuple       =	self.get_dummy_inputs(UpperCAmelCase_					, pil_image=UpperCAmelCase_    )
						A__							:					List[str]       =	[
						    tuple_pipeline_inputs['image'],
						    tuple_pipeline_inputs['image'],
						]

						A__							:					str       =	pipe(
						    **UpperCAmelCase_					, return_dict=UpperCAmelCase_					, )[0]

						A__							:					str       =	image[0, -3:, -3:, -1]
						A__							:					Optional[int]       =	image_from_tuple[0, -3:, -3:, -1]

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

						A__							:					Any       =	np.array(
						    [
						        0.9_9_9_7,
						        0.9_9_8_9,
						        0.0_0_0_8,
						        0.0_0_2_1,
						        0.9_9_6_0,
						        0.0_0_1_8,
						        0.0_0_1_4,
						        0.0_0_0_2,
						        0.9_9_3_3,
						    ]    )

						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 						__A    (				self    ):
						A__							:					List[Any]       =	torch.device("""cpu"""    )



						class 					_a		:


											'''simple docstring'''






											UpperCAmelCase__:     Any													= 1



						A__							:					List[Any]       =	self.get_dummy_components()

						A__							:					Dict       =	self.pipeline_class(**UpperCAmelCase_    )
						A__							:					str       =	pipe.to(UpperCAmelCase_    )

						pipe.set_progress_bar_config(disable=UpperCAmelCase_    )

						A__							:					Dict       =	torch.Generator(device=UpperCAmelCase_    ).manual_seed(0    )
						A__							:					Union[str, Any]       =	pipe.decoder.dtype
						A__							:					Optional[int]       =	1

						A__							:					Optional[Any]       =	(
						    batch_size,
						    pipe.decoder.config.in_channels,
						    pipe.decoder.config.sample_size,
						    pipe.decoder.config.sample_size,
						)
						A__							:					Optional[int]       =	pipe.prepare_latents(
						    UpperCAmelCase_					, dtype=UpperCAmelCase_					, device=UpperCAmelCase_					, generator=UpperCAmelCase_					, latents=UpperCAmelCase_					, scheduler=DummyScheduler()    )

						A__							:					Any       =	(
						    batch_size,
						    pipe.super_res_first.config.in_channels // 2,
						    pipe.super_res_first.config.sample_size,
						    pipe.super_res_first.config.sample_size,
						)
						A__							:					str       =	pipe.prepare_latents(
						    UpperCAmelCase_					, dtype=UpperCAmelCase_					, device=UpperCAmelCase_					, generator=UpperCAmelCase_					, latents=UpperCAmelCase_					, scheduler=DummyScheduler()    )

						A__							:					List[str]       =	self.get_dummy_inputs(UpperCAmelCase_					, pil_image=UpperCAmelCase_    )

						A__							:					Union[str, Any]       =	pipe(
						    **UpperCAmelCase_					, decoder_latents=UpperCAmelCase_					, super_res_latents=UpperCAmelCase_    ).images

						A__							:					int       =	self.get_dummy_inputs(UpperCAmelCase_					, pil_image=UpperCAmelCase_    )
						# Don't pass image, instead pass embedding
						A__							:					Dict       =	pipeline_inputs.pop("""image"""    )
						A__							:					int       =	pipe.image_encoder(UpperCAmelCase_    ).image_embeds

						A__							:					int       =	pipe(
						    **UpperCAmelCase_					, decoder_latents=UpperCAmelCase_					, super_res_latents=UpperCAmelCase_					, image_embeddings=UpperCAmelCase_					, ).images

						# make sure passing text embeddings manually is identical
						assert np.abs(img_out_a - img_out_a    ).max() < 1e-4






					@skip_mps
					def 						__A    (				self    ):
						A__							:					Optional[int]       =	torch_device == 'cpu'

						# Check is relaxed because there is not a torch 2.0 sliced attention added kv processor
						A__							:					Optional[Any]       =	1e-2

						self._test_attention_slicing_forward_pass(
						    test_max_difference=UpperCAmelCase_					, expected_max_diff=UpperCAmelCase_    )






					@skip_mps
					def 						__A    (				self    ):
						A__							:					List[Any]       =	torch_device == 'cpu'
						A__							:					Dict       =	True
						A__							:					Optional[int]       =	[
						    'decoder_num_inference_steps',
						    'super_res_num_inference_steps',
						]

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






					def 						__A    (				self    ):
						A__							:					List[Any]       =	[
						    'decoder_num_inference_steps',
						    'super_res_num_inference_steps',
						]

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






					@skip_mps
					def 						__A    (				self    ):
						return super().test_dict_tuple_outputs_equivalent()






					@skip_mps
					def 						__A    (				self    ):
						return super().test_save_load_local()






					@skip_mps
					def 						__A    (				self    ):
						return super().test_save_load_optional_components()



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


					'''simple docstring'''






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






					def 						__A    (				self    ):
						A__							:					Optional[int]       =	load_image(
						    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/unclip/cat.png"""    )
						A__							:					Dict       =	load_numpy(
						    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
						    """/unclip/karlo_v1_alpha_cat_variation_fp16.npy"""    )

						A__							:					Any       =	UnCLIPImageVariationPipeline.from_pretrained(
						    """kakaobrain/karlo-v1-alpha-image-variations"""					, torch_dtype=torch.floataa    )
						A__							:					Dict       =	pipeline.to(UpperCAmelCase_    )
						pipeline.set_progress_bar_config(disable=UpperCAmelCase_    )

						A__							:					List[Any]       =	torch.Generator(device="""cpu"""    ).manual_seed(0    )
						A__							:					Optional[int]       =	pipeline(
						    UpperCAmelCase_					, generator=UpperCAmelCase_					, output_type="""np"""					, )

						A__							:					List[Any]       =	output.images[0]

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

						assert_mean_pixel_difference(UpperCAmelCase_					, UpperCAmelCase_					, 15    )






def 		UpperCamelCase      (lowercase_:				int      )  ->   int:
	if not isinstance(lowercase_					,   lowercase_      ):
		raise TypeError("""Input value must be an 'int' type"""      )

	A__							:					int       =	0
	while number:
		position += 1
		number >>= 1

	return position


if __name__ == "__main__":
	import doctest

	doctest.testmod()



