



'''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 					UpperCAmelCase__	(		UpperCamelCase__							,       unittest.TestCase ):
					a      :     List[str]        =							UnCLIPImageVariationPipeline
					a      :     str        =							IMAGE_VARIATION_PARAMS - {"""height""", """width""", """guidance_scale"""}
					a      :     List[str]        =							IMAGE_VARIATION_BATCH_PARAMS

					a      :     int        =							[
					    """generator""",
					    """return_dict""",
					    """decoder_num_inference_steps""",
					    """super_res_num_inference_steps""",
					]
					a      :     int        =							False



					@property
					def        UpperCAmelCase_ (    self     )     ->			Dict:
						return 32



					@property
					def        UpperCAmelCase_ (    self     )     ->			str:
						return 32



					@property
					def        UpperCAmelCase_ (    self     )     ->			Union[str, Any]:
						return self.time_input_dim



					@property
					def        UpperCAmelCase_ (    self     )     ->			int:
						return self.time_input_dim * 4



					@property
					def        UpperCAmelCase_ (    self     )     ->			List[Any]:
						return 100



					@property
					def        UpperCAmelCase_ (    self     )     ->			Tuple:
						__lowerCAmelCase												=  CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"     )
						return tokenizer



					@property
					def        UpperCAmelCase_ (    self     )     ->			int:
						torch.manual_seed(0     )
						__lowerCAmelCase												=  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        UpperCAmelCase_ (    self     )     ->			str:
						torch.manual_seed(0     )
						__lowerCAmelCase												=  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        UpperCAmelCase_ (    self     )     ->			Optional[Any]:
						torch.manual_seed(0     )

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

						__lowerCAmelCase												=  UnCLIPTextProjModel(**UpperCamelCase     )
						return model



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

						__lowerCAmelCase												=  {
						    "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",
						}

						__lowerCAmelCase												=  UNetaDConditionModel(**UpperCamelCase     )
						return model



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

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



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

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



					def        UpperCAmelCase_ (    self     )     ->			Optional[int]:
						__lowerCAmelCase												=  self.dummy_decoder
						__lowerCAmelCase												=  self.dummy_text_proj
						__lowerCAmelCase												=  self.dummy_text_encoder
						__lowerCAmelCase												=  self.dummy_tokenizer
						__lowerCAmelCase												=  self.dummy_super_res_first
						__lowerCAmelCase												=  self.dummy_super_res_last

						__lowerCAmelCase												=  UnCLIPScheduler(
						    variance_type="learned_range"			,				prediction_type="epsilon"			,				num_train_timesteps=1000			,				)

						__lowerCAmelCase												=  UnCLIPScheduler(
						    variance_type="fixed_small_log"			,				prediction_type="epsilon"			,				num_train_timesteps=1000			,				)

						__lowerCAmelCase												=  CLIPImageProcessor(crop_size=32			,				size=32     )

						__lowerCAmelCase												=  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        UpperCAmelCase_ (    self			,				UpperCamelCase			,				UpperCamelCase=0			,				UpperCamelCase=True     )     ->			Optional[int]:
						__lowerCAmelCase												=  floats_tensor((1, 3, 32, 32)			,				rng=random.Random(UpperCamelCase     )     ).to(UpperCamelCase     )
						if str(UpperCamelCase     ).startswith("mps"     ):
							__lowerCAmelCase												=  torch.manual_seed(UpperCamelCase     )
						else:
							__lowerCAmelCase												=  torch.Generator(device=UpperCamelCase     ).manual_seed(UpperCamelCase     )

						if pil_image:
							__lowerCAmelCase												=  input_image * 0.5 + 0.5
							__lowerCAmelCase												=  input_image.clamp(0			,				1     )
							__lowerCAmelCase												=  input_image.cpu().permute(0			,				2			,				3			,				1     ).float().numpy()
							__lowerCAmelCase												=  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        UpperCAmelCase_ (    self     )     ->			str:
						__lowerCAmelCase												=  "cpu"

						__lowerCAmelCase												=  self.get_dummy_components()

						__lowerCAmelCase												=  self.pipeline_class(**UpperCamelCase     )
						__lowerCAmelCase												=  pipe.to(UpperCamelCase     )

						pipe.set_progress_bar_config(disable=UpperCamelCase     )

						__lowerCAmelCase												=  self.get_dummy_inputs(UpperCamelCase			,				pil_image=UpperCamelCase     )

						__lowerCAmelCase												=  pipe(**UpperCamelCase     )
						__lowerCAmelCase												=  output.images

						__lowerCAmelCase												=  self.get_dummy_inputs(UpperCamelCase			,				pil_image=UpperCamelCase     )

						__lowerCAmelCase												=  pipe(
						    **UpperCamelCase			,				return_dict=UpperCamelCase			,				)[0]

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

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

						__lowerCAmelCase												=  np.array(
						    [
						        0.99_97,
						        0.00_02,
						        0.99_97,
						        0.99_97,
						        0.99_69,
						        0.00_23,
						        0.99_97,
						        0.99_69,
						        0.99_70,
						    ]     )

						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     )     ->			Union[str, Any]:
						__lowerCAmelCase												=  "cpu"

						__lowerCAmelCase												=  self.get_dummy_components()

						__lowerCAmelCase												=  self.pipeline_class(**UpperCamelCase     )
						__lowerCAmelCase												=  pipe.to(UpperCamelCase     )

						pipe.set_progress_bar_config(disable=UpperCamelCase     )

						__lowerCAmelCase												=  self.get_dummy_inputs(UpperCamelCase			,				pil_image=UpperCamelCase     )

						__lowerCAmelCase												=  pipe(**UpperCamelCase     )
						__lowerCAmelCase												=  output.images

						__lowerCAmelCase												=  self.get_dummy_inputs(UpperCamelCase			,				pil_image=UpperCamelCase     )

						__lowerCAmelCase												=  pipe(
						    **UpperCamelCase			,				return_dict=UpperCamelCase			,				)[0]

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

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

						__lowerCAmelCase												=  np.array([0.99_97, 0.00_03, 0.99_97, 0.99_97, 0.99_70, 0.00_24, 0.99_97, 0.99_71, 0.99_71]     )

						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     )     ->			Optional[Any]:
						__lowerCAmelCase												=  "cpu"

						__lowerCAmelCase												=  self.get_dummy_components()

						__lowerCAmelCase												=  self.pipeline_class(**UpperCamelCase     )
						__lowerCAmelCase												=  pipe.to(UpperCamelCase     )

						pipe.set_progress_bar_config(disable=UpperCamelCase     )

						__lowerCAmelCase												=  self.get_dummy_inputs(UpperCamelCase			,				pil_image=UpperCamelCase     )
						__lowerCAmelCase												=  [
						    pipeline_inputs["image"],
						    pipeline_inputs["image"],
						]

						__lowerCAmelCase												=  pipe(**UpperCamelCase     )
						__lowerCAmelCase												=  output.images

						__lowerCAmelCase												=  self.get_dummy_inputs(UpperCamelCase			,				pil_image=UpperCamelCase     )
						__lowerCAmelCase												=  [
						    tuple_pipeline_inputs["image"],
						    tuple_pipeline_inputs["image"],
						]

						__lowerCAmelCase												=  pipe(
						    **UpperCamelCase			,				return_dict=UpperCamelCase			,				)[0]

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

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

						__lowerCAmelCase												=  np.array(
						    [
						        0.99_97,
						        0.99_89,
						        0.00_08,
						        0.00_21,
						        0.99_60,
						        0.00_18,
						        0.00_14,
						        0.00_02,
						        0.99_33,
						    ]     )

						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     )     ->			Tuple:
						__lowerCAmelCase												=  torch.device("cpu"     )






						class 					UpperCAmelCase__	:
											a      :     int        =							1


						__lowerCAmelCase												=  self.get_dummy_components()

						__lowerCAmelCase												=  self.pipeline_class(**UpperCamelCase     )
						__lowerCAmelCase												=  pipe.to(UpperCamelCase     )

						pipe.set_progress_bar_config(disable=UpperCamelCase     )

						__lowerCAmelCase												=  torch.Generator(device=UpperCamelCase     ).manual_seed(0     )
						__lowerCAmelCase												=  pipe.decoder.dtype
						__lowerCAmelCase												=  1

						__lowerCAmelCase												=  (
						    batch_size,
						    pipe.decoder.config.in_channels,
						    pipe.decoder.config.sample_size,
						    pipe.decoder.config.sample_size,
						)
						__lowerCAmelCase												=  pipe.prepare_latents(
						    UpperCamelCase			,				dtype=UpperCamelCase			,				device=UpperCamelCase			,				generator=UpperCamelCase			,				latents=UpperCamelCase			,				scheduler=DummyScheduler()     )

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

						__lowerCAmelCase												=  self.get_dummy_inputs(UpperCamelCase			,				pil_image=UpperCamelCase     )

						__lowerCAmelCase												=  pipe(
						    **UpperCamelCase			,				decoder_latents=UpperCamelCase			,				super_res_latents=UpperCamelCase     ).images

						__lowerCAmelCase												=  self.get_dummy_inputs(UpperCamelCase			,				pil_image=UpperCamelCase     )
						# Don't pass image, instead pass embedding
						__lowerCAmelCase												=  pipeline_inputs.pop("image"     )
						__lowerCAmelCase												=  pipe.image_encoder(UpperCamelCase     ).image_embeds

						__lowerCAmelCase												=  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        UpperCAmelCase_ (    self     )     ->			str:
						__lowerCAmelCase												=  torch_device == "cpu"

						# Check is relaxed because there is not a torch 2.0 sliced attention added kv processor
						__lowerCAmelCase												=  1E-2

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



					@skip_mps
					def        UpperCAmelCase_ (    self     )     ->			Any:
						__lowerCAmelCase												=  torch_device == "cpu"
						__lowerCAmelCase												=  True
						__lowerCAmelCase												=  [
						    "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        UpperCAmelCase_ (    self     )     ->			Dict:
						__lowerCAmelCase												=  [
						    "decoder_num_inference_steps",
						    "super_res_num_inference_steps",
						]

						if torch_device == "mps":
							# TODO: MPS errors with larger batch sizes
							__lowerCAmelCase												=  [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        UpperCAmelCase_ (    self     )     ->			List[str]:
						return super().test_dict_tuple_outputs_equivalent()



					@skip_mps
					def        UpperCAmelCase_ (    self     )     ->			Tuple:
						return super().test_save_load_local()



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






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



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



					def        UpperCAmelCase_ (    self     )     ->			Optional[Any]:
						__lowerCAmelCase												=  load_image(
						    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/unclip/cat.png"     )
						__lowerCAmelCase												=  load_numpy(
						    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
						    "/unclip/karlo_v1_alpha_cat_variation_fp16.npy"     )

						__lowerCAmelCase												=  UnCLIPImageVariationPipeline.from_pretrained(
						    "kakaobrain/karlo-v1-alpha-image-variations"			,				torch_dtype=torch.floataa     )
						__lowerCAmelCase												=  pipeline.to(UpperCamelCase     )
						pipeline.set_progress_bar_config(disable=UpperCamelCase     )

						__lowerCAmelCase												=  torch.Generator(device="cpu"     ).manual_seed(0     )
						__lowerCAmelCase												=  pipeline(
						    UpperCamelCase			,				generator=UpperCamelCase			,				output_type="np"			,				)

						__lowerCAmelCase												=  output.images[0]

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

						assert_mean_pixel_difference(UpperCamelCase			,				UpperCamelCase			,				15     )




'''simple docstring'''



# flake8: noqa
# Lint as: python3

from typing import Dict, List, Optional, Type

from .. import config
from ..utils import logging
from .formatting import (
    ArrowFormatter,
    CustomFormatter,
    Formatter,
    PandasFormatter,
    PythonFormatter,
    TensorFormatter,
    format_table,
    query_table,
)
from .np_formatter import NumpyFormatter


lowerCAmelCase	:		List[str]				=   logging.get_logger(__name__)

lowerCAmelCase	:		Dict[Optional[str], Type[Formatter]]				=   {}
lowerCAmelCase	:		Dict[Optional[str], str]				=   {}
lowerCAmelCase	:		Dict[Optional[str], Exception]				=   {}



def   __lowerCAmelCase					(							lowerCamelCase   :						type				,		lowerCamelCase   :						Optional[str]				,		lowerCamelCase   :						Optional[List[str]] = None				,		):






	'''simple docstring'''




	__lowerCAmelCase												=  aliases if aliases is not None else []
	if format_type in _FORMAT_TYPES:
		logger.warning(
		    f'''Overwriting format type \'{format_type}\' ({_FORMAT_TYPES[format_type].__name__} -> {formatter_cls.__name__})'''					)
	__lowerCAmelCase												=  formatter_cls
	for alias in set(aliases + [format_type]					):
		if alias in _FORMAT_TYPES_ALIASES:
			logger.warning(
			    f'''Overwriting format type alias \'{alias}\' ({_FORMAT_TYPES_ALIASES[alias]} -> {format_type})'''					)
		__lowerCAmelCase												=  format_type



def   __lowerCAmelCase					(							lowerCamelCase   :						Exception				,		lowerCamelCase   :						Optional[str]				,		lowerCamelCase   :						Optional[List[str]] = None					):






	'''simple docstring'''




	__lowerCAmelCase												=  aliases if aliases is not None else []
	for alias in set(aliases + [format_type]					):
		__lowerCAmelCase												=  unavailable_error


# Here we define all the available formatting functions that can be used by `Dataset.set_format`
_register_formatter(PythonFormatter, None, aliases=['''python'''])
_register_formatter(ArrowFormatter, '''arrow''', aliases=['''pa''', '''pyarrow'''])
_register_formatter(NumpyFormatter, '''numpy''', aliases=['''np'''])
_register_formatter(PandasFormatter, '''pandas''', aliases=['''pd'''])
_register_formatter(CustomFormatter, '''custom''')

if config.TORCH_AVAILABLE:
	from .torch_formatter import TorchFormatter

	_register_formatter(TorchFormatter, '''torch''', aliases=['''pt''', '''pytorch'''])
else:
	lowerCAmelCase	:		Optional[int]				=   ValueError('''PyTorch needs to be installed to be able to return PyTorch tensors.''')
	_register_unavailable_formatter(_torch_error, '''torch''', aliases=['''pt''', '''pytorch'''])

if config.TF_AVAILABLE:
	from .tf_formatter import TFFormatter

	_register_formatter(TFFormatter, '''tensorflow''', aliases=['''tf'''])
else:
	lowerCAmelCase	:		str				=   ValueError('''Tensorflow needs to be installed to be able to return Tensorflow tensors.''')
	_register_unavailable_formatter(_tf_error, '''tensorflow''', aliases=['''tf'''])

if config.JAX_AVAILABLE:
	from .jax_formatter import JaxFormatter

	_register_formatter(JaxFormatter, '''jax''', aliases=[])
else:
	lowerCAmelCase	:		Any				=   ValueError('''JAX needs to be installed to be able to return JAX arrays.''')
	_register_unavailable_formatter(_jax_error, '''jax''', aliases=[])



def   __lowerCAmelCase					(							lowerCamelCase   :						Optional[str]					):






	'''simple docstring'''




	if format_type in _FORMAT_TYPES_ALIASES:
		return _FORMAT_TYPES_ALIASES[format_type]
	else:
		return format_type




def   __lowerCAmelCase					(							lowerCamelCase   :						Optional[str]				,		**lowerCamelCase   :						Tuple					):






	'''simple docstring'''




	__lowerCAmelCase												=  get_format_type_from_alias(lowerCamelCase					)
	if format_type in _FORMAT_TYPES:
		return _FORMAT_TYPES[format_type](**lowerCamelCase					)
	if format_type in _FORMAT_TYPES_ALIASES_UNAVAILABLE:
		raise _FORMAT_TYPES_ALIASES_UNAVAILABLE[format_type]
	else:
		raise ValueError(
		    f'''Return type should be None or selected in {list(type for type in _FORMAT_TYPES.keys() if type != None					)}, but got \'{format_type}\''''					)