





"""simple docstring"""






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 			__lowercase     (     _UpperCamelCase				, unittest.TestCase							):







					'''simple docstring'''




					__lowerCAmelCase					=					UnCLIPImageVariationPipeline
					__lowerCAmelCase					=					IMAGE_VARIATION_PARAMS - {'''height''', '''width''', '''guidance_scale'''}
					__lowerCAmelCase					=					IMAGE_VARIATION_BATCH_PARAMS

					__lowerCAmelCase					=					[
					    '''generator''',
					    '''return_dict''',
					    '''decoder_num_inference_steps''',
					    '''super_res_num_inference_steps''',
					]
					__lowerCAmelCase					=					False







					@property
					def 	_lowerCamelCase						(  self       ):
									return 32







					@property
					def 	_lowerCamelCase						(  self       ):
									return 32







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







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







					@property
					def 	_lowerCamelCase						(  self       ):
									return 100







					@property
					def 	_lowerCamelCase						(  self       ):
									__a	:  Union[str, Any]         =					CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''       )
									return tokenizer







					@property
					def 	_lowerCamelCase						(  self       ):
									torch.manual_seed(0       )
									__a	:  Optional[int]         =					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-0_5		,     num_attention_heads=4		,     num_hidden_layers=5		,     pad_token_id=1		,     vocab_size=1000		,     )
									return CLIPTextModelWithProjection(_UpperCAmelCase       )







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

									__a	:  List[str]         =					{
									    '''clip_embeddings_dim''': self.text_embedder_hidden_size,
									    '''time_embed_dim''': self.time_embed_dim,
									    '''cross_attention_dim''': self.cross_attention_dim,
									}

									__a	:  Optional[int]         =					UnCLIPTextProjModel(**_UpperCAmelCase       )
									return model







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

									__a	:  Optional[int]         =					{
									    '''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	:  List[Any]         =					UNetaDConditionModel(**_UpperCAmelCase       )
									return model







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

									__a	:  Optional[Any]         =					UNetaDModel(**self.dummy_super_res_kwargs       )
									return model







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

									__a	:  Union[str, Any]         =					UNetaDModel(**self.dummy_super_res_kwargs       )
									return model







					def 	_lowerCamelCase						(  self       ):
									__a	:  Any         =					self.dummy_decoder
									__a	:  List[str]         =					self.dummy_text_proj
									__a	:  str         =					self.dummy_text_encoder
									__a	:  Union[str, Any]         =					self.dummy_tokenizer
									__a	:  Any         =					self.dummy_super_res_first
									__a	:  Tuple         =					self.dummy_super_res_last

									__a	:  Optional[int]         =					UnCLIPScheduler(
									    variance_type='''learned_range'''		,     prediction_type='''epsilon'''		,     num_train_timesteps=1000		,     )

									__a	:  Any         =					UnCLIPScheduler(
									    variance_type='''fixed_small_log'''		,     prediction_type='''epsilon'''		,     num_train_timesteps=1000		,     )

									__a	:  Union[str, Any]         =					CLIPImageProcessor(crop_size=32		,     size=32       )

									__a	:  int         =					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 	_lowerCamelCase						(  self		,     _UpperCAmelCase		,     _UpperCAmelCase=0		,     _UpperCAmelCase=True       ):
									__a	:  List[Any]         =					floats_tensor((1, 3, 32, 32)		,     rng=random.Random(_UpperCAmelCase       )       ).to(_UpperCAmelCase       )
									if str(_UpperCAmelCase       ).startswith('''mps'''       ):
													__a	:  Tuple         =					torch.manual_seed(_UpperCAmelCase       )
									else:
													__a	:  List[str]         =					torch.Generator(device=_UpperCAmelCase       ).manual_seed(_UpperCAmelCase       )

									if pil_image:
													__a	:  Any         =					input_image * 0.5 + 0.5
													__a	:  Union[str, Any]         =					input_image.clamp(0		,     1       )
													__a	:  Dict         =					input_image.cpu().permute(0		,     2		,     3		,     1       ).float().numpy()
													__a	:  Dict         =					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 	_lowerCamelCase						(  self       ):
									__a	:  str         =					'''cpu'''

									__a	:  Union[str, Any]         =					self.get_dummy_components()

									__a	:  List[str]         =					self.pipeline_class(**_UpperCAmelCase       )
									__a	:  List[Any]         =					pipe.to(_UpperCAmelCase       )

									pipe.set_progress_bar_config(disable=_UpperCAmelCase       )

									__a	:  int         =					self.get_dummy_inputs(_UpperCAmelCase		,     pil_image=_UpperCAmelCase       )

									__a	:  Optional[int]         =					pipe(**_UpperCAmelCase       )
									__a	:  Tuple         =					output.images

									__a	:  Any         =					self.get_dummy_inputs(_UpperCAmelCase		,     pil_image=_UpperCAmelCase       )

									__a	:  Optional[Any]         =					pipe(
									    **_UpperCAmelCase		,     return_dict=_UpperCAmelCase		,     )[0]

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

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

									__a	:  Optional[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 	_lowerCamelCase						(  self       ):
									__a	:  List[Any]         =					'''cpu'''

									__a	:  Any         =					self.get_dummy_components()

									__a	:  str         =					self.pipeline_class(**_UpperCAmelCase       )
									__a	:  List[str]         =					pipe.to(_UpperCAmelCase       )

									pipe.set_progress_bar_config(disable=_UpperCAmelCase       )

									__a	:  str         =					self.get_dummy_inputs(_UpperCAmelCase		,     pil_image=_UpperCAmelCase       )

									__a	:  Optional[int]         =					pipe(**_UpperCAmelCase       )
									__a	:  Tuple         =					output.images

									__a	:  List[Any]         =					self.get_dummy_inputs(_UpperCAmelCase		,     pil_image=_UpperCAmelCase       )

									__a	:  Dict         =					pipe(
									    **_UpperCAmelCase		,     return_dict=_UpperCAmelCase		,     )[0]

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

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

									__a	:  Tuple         =					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 	_lowerCamelCase						(  self       ):
									__a	:  str         =					'''cpu'''

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

									__a	:  Optional[Any]         =					self.pipeline_class(**_UpperCAmelCase       )
									__a	:  Any         =					pipe.to(_UpperCAmelCase       )

									pipe.set_progress_bar_config(disable=_UpperCAmelCase       )

									__a	:  str         =					self.get_dummy_inputs(_UpperCAmelCase		,     pil_image=_UpperCAmelCase       )
									__a	:  str         =					[
									    pipeline_inputs['''image'''],
									    pipeline_inputs['''image'''],
									]

									__a	:  int         =					pipe(**_UpperCAmelCase       )
									__a	:  Dict         =					output.images

									__a	:  Union[str, Any]         =					self.get_dummy_inputs(_UpperCAmelCase		,     pil_image=_UpperCAmelCase       )
									__a	:  Any         =					[
									    tuple_pipeline_inputs['''image'''],
									    tuple_pipeline_inputs['''image'''],
									]

									__a	:  Dict         =					pipe(
									    **_UpperCAmelCase		,     return_dict=_UpperCAmelCase		,     )[0]

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

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

									__a	:  Tuple         =					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 	_lowerCamelCase						(  self       ):
									__a	:  str         =					torch.device('''cpu'''       )




									class 			__lowercase     :







														'''simple docstring'''




														__lowerCAmelCase					=					1






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

									__a	:  int         =					self.pipeline_class(**_UpperCAmelCase       )
									__a	:  Optional[int]         =					pipe.to(_UpperCAmelCase       )

									pipe.set_progress_bar_config(disable=_UpperCAmelCase       )

									__a	:  List[Any]         =					torch.Generator(device=_UpperCAmelCase       ).manual_seed(0       )
									__a	:  Any         =					pipe.decoder.dtype
									__a	:  Optional[Any]         =					1

									__a	:  int         =					(
									    batch_size,
									    pipe.decoder.config.in_channels,
									    pipe.decoder.config.sample_size,
									    pipe.decoder.config.sample_size,
									)
									__a	:  Dict         =					pipe.prepare_latents(
									    _UpperCAmelCase		,     dtype=_UpperCAmelCase		,     device=_UpperCAmelCase		,     generator=_UpperCAmelCase		,     latents=_UpperCAmelCase		,     scheduler=DummyScheduler()       )

									__a	:  List[str]         =					(
									    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	:  int         =					pipe.prepare_latents(
									    _UpperCAmelCase		,     dtype=_UpperCAmelCase		,     device=_UpperCAmelCase		,     generator=_UpperCAmelCase		,     latents=_UpperCAmelCase		,     scheduler=DummyScheduler()       )

									__a	:  Optional[int]         =					self.get_dummy_inputs(_UpperCAmelCase		,     pil_image=_UpperCAmelCase       )

									__a	:  int         =					pipe(
									    **_UpperCAmelCase		,     decoder_latents=_UpperCAmelCase		,     super_res_latents=_UpperCAmelCase       ).images

									__a	:  Union[str, Any]         =					self.get_dummy_inputs(_UpperCAmelCase		,     pil_image=_UpperCAmelCase       )
									# Don't pass image, instead pass embedding
									__a	:  str         =					pipeline_inputs.pop('''image'''       )
									__a	:  str         =					pipe.image_encoder(_UpperCAmelCase       ).image_embeds

									__a	:  List[Any]         =					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 	_lowerCamelCase						(  self       ):
									__a	:  Optional[Any]         =					torch_device == '''cpu'''

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

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







					@skip_mps
					def 	_lowerCamelCase						(  self       ):
									__a	:  Optional[Any]         =					torch_device == '''cpu'''
									__a	:  int         =					True
									__a	:  str         =					[
									    '''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 	_lowerCamelCase						(  self       ):
									__a	:  Optional[int]         =					[
									    '''decoder_num_inference_steps''',
									    '''super_res_num_inference_steps''',
									]

									if torch_device == "mps":
													# TODO: MPS errors with larger batch sizes
													__a	:  Dict         =					[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 	_lowerCamelCase						(  self       ):
									return super().test_dict_tuple_outputs_equivalent()







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







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






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







					'''simple docstring'''




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







					def 	_lowerCamelCase						(  self       ):
									__a	:  List[str]         =					load_image(
									    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/unclip/cat.png'''       )
									__a	:  List[str]         =					load_numpy(
									    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
									    '''/unclip/karlo_v1_alpha_cat_variation_fp16.npy'''       )

									__a	:  Union[str, Any]         =					UnCLIPImageVariationPipeline.from_pretrained(
									    '''kakaobrain/karlo-v1-alpha-image-variations'''		,     torch_dtype=torch.floataa       )
									__a	:  Tuple         =					pipeline.to(_UpperCAmelCase       )
									pipeline.set_progress_bar_config(disable=_UpperCAmelCase       )

									__a	:  Optional[int]         =					torch.Generator(device='''cpu'''       ).manual_seed(0       )
									__a	:  List[Any]         =					pipeline(
									    _UpperCAmelCase		,     generator=_UpperCAmelCase		,     output_type='''np'''		,     )

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

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

									assert_mean_pixel_difference(_UpperCAmelCase		,     _UpperCAmelCase		,     15       )






"""simple docstring"""






from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_tokenizers_available, is_torch_available


A	     =					{
    '''configuration_mvp''': ['''MVP_PRETRAINED_CONFIG_ARCHIVE_MAP''', '''MvpConfig''', '''MvpOnnxConfig'''],
    '''tokenization_mvp''': ['''MvpTokenizer'''],
}

try:
		if not is_tokenizers_available():
				raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
		pass
else:
		A	     =					['''MvpTokenizerFast''']

try:
		if not is_torch_available():
				raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
		pass
else:
		A	     =					[
		    '''MVP_PRETRAINED_MODEL_ARCHIVE_LIST''',
		    '''MvpForCausalLM''',
		    '''MvpForConditionalGeneration''',
		    '''MvpForQuestionAnswering''',
		    '''MvpForSequenceClassification''',
		    '''MvpModel''',
		    '''MvpPreTrainedModel''',
		]

if TYPE_CHECKING:
		from .configuration_mvp import MVP_PRETRAINED_CONFIG_ARCHIVE_MAP, MvpConfig, MvpOnnxConfig
		from .tokenization_mvp import MvpTokenizer

		try:
				if not is_tokenizers_available():
						raise OptionalDependencyNotAvailable()
		except OptionalDependencyNotAvailable:
				pass
		else:
				from .tokenization_mvp_fast import MvpTokenizerFast

		try:
				if not is_torch_available():
						raise OptionalDependencyNotAvailable()
		except OptionalDependencyNotAvailable:
				pass
		else:
				from .modeling_mvp import (
				    MVP_PRETRAINED_MODEL_ARCHIVE_LIST,
				    MvpForCausalLM,
				    MvpForConditionalGeneration,
				    MvpForQuestionAnswering,
				    MvpForSequenceClassification,
				    MvpModel,
				    MvpPreTrainedModel,
				)

else:
		import sys

		A	     =					_LazyModule(__name__, globals()['''__file__'''], _import_structure, module_spec=__spec__)