






'''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    _SCREAMING_SNAKE_CASE							(		SCREAMING_SNAKE_CASE       ,unittest.TestCase	):


			'''simple docstring'''





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

			__a							:							Union[str, Any]					     =					[
			    "generator",
			    "return_dict",
			    "decoder_num_inference_steps",
			    "super_res_num_inference_steps",
			]
			__a							:							str					     =					False






			@property
			def 				A      (   self							:						Tuple			)							->    Optional[int]:

									'''simple docstring'''







									return 3_2






			@property
			def 				A      (   self							:						str			)							->    Optional[Any]:

									'''simple docstring'''







									return 3_2






			@property
			def 				A      (   self							:						Optional[Any]			)							->    Optional[int]:

									'''simple docstring'''







									return self.time_input_dim






			@property
			def 				A      (   self							:						str			)							->    Any:

									'''simple docstring'''







									return self.time_input_dim * 4






			@property
			def 				A      (   self							:						Tuple			)							->    Dict:

									'''simple docstring'''







									return 1_0_0






			@property
			def 				A      (   self							:						Tuple			)							->    Dict:

									'''simple docstring'''







									UpperCamelCase__         = CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""			)
									return tokenizer






			@property
			def 				A      (   self							:						Optional[int]			)							->    Dict:

									'''simple docstring'''







									torch.manual_seed(0			)
									UpperCamelCase__         = 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=3_7 ,      layer_norm_eps=1e-05 ,      num_attention_heads=4 ,      num_hidden_layers=5 ,      pad_token_id=1 ,      vocab_size=1_0_0_0 ,      )
									return CLIPTextModelWithProjection(lowercase			)






			@property
			def 				A      (   self							:						List[str]			)							->    Dict:

									'''simple docstring'''







									torch.manual_seed(0			)
									UpperCamelCase__         = 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=3_2 ,      intermediate_size=3_7 ,      patch_size=1 ,      )
									return CLIPVisionModelWithProjection(lowercase			)






			@property
			def 				A      (   self							:						List[str]			)							->    int:

									'''simple docstring'''







									torch.manual_seed(0			)

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

									UpperCamelCase__         = UnCLIPTextProjModel(**lowercase			)
									return model






			@property
			def 				A      (   self							:						Tuple			)							->    int:

									'''simple docstring'''







									torch.manual_seed(0			)

									UpperCamelCase__         = {
									    """sample_size""": 3_2,
									    # 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__         = UNetaDConditionModel(**lowercase			)
									return model






			@property
			def 				A      (   self							:						Any			)							->    Union[str, Any]:

									'''simple docstring'''







									return {
									    "sample_size": 6_4,
									    "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							:						Union[str, Any]			)							->    Dict:

									'''simple docstring'''







									torch.manual_seed(0			)

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






			@property
			def 				A      (   self							:						Union[str, Any]			)							->    int:

									'''simple docstring'''







									torch.manual_seed(1			)

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






			def 				A      (   self							:						Optional[int]			)							->    str:

									'''simple docstring'''







									UpperCamelCase__         = self.dummy_decoder
									UpperCamelCase__         = self.dummy_text_proj
									UpperCamelCase__         = self.dummy_text_encoder
									UpperCamelCase__         = self.dummy_tokenizer
									UpperCamelCase__         = self.dummy_super_res_first
									UpperCamelCase__         = self.dummy_super_res_last

									UpperCamelCase__         = UnCLIPScheduler(
									    variance_type="""learned_range""" ,      prediction_type="""epsilon""" ,      num_train_timesteps=1_0_0_0 ,      )

									UpperCamelCase__         = UnCLIPScheduler(
									    variance_type="""fixed_small_log""" ,      prediction_type="""epsilon""" ,      num_train_timesteps=1_0_0_0 ,      )

									UpperCamelCase__         = CLIPImageProcessor(crop_size=3_2 ,      size=3_2			)

									UpperCamelCase__         = 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							:						List[str] ,      lowercase							:						List[str] ,      lowercase							:						str=0 ,      lowercase							:						Tuple=True			)							->    Union[str, Any]:

									'''simple docstring'''







									UpperCamelCase__         = floats_tensor((1, 3, 3_2, 3_2) ,      rng=random.Random(lowercase			)			).to(lowercase			)
									if str(lowercase			).startswith("""mps"""			):
															UpperCamelCase__         = torch.manual_seed(lowercase			)
									else:
															UpperCamelCase__         = torch.Generator(device=lowercase			).manual_seed(lowercase			)

									if pil_image:
															UpperCamelCase__         = input_image * 0.5 + 0.5
															UpperCamelCase__         = input_image.clamp(0 ,      1			)
															UpperCamelCase__         = input_image.cpu().permute(0 ,      2 ,      3 ,      1			).float().numpy()
															UpperCamelCase__         = DiffusionPipeline.numpy_to_pil(lowercase			)[0]

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






			def 				A      (   self							:						str			)							->    List[str]:

									'''simple docstring'''







									UpperCamelCase__         = """cpu"""

									UpperCamelCase__         = self.get_dummy_components()

									UpperCamelCase__         = self.pipeline_class(**lowercase			)
									UpperCamelCase__         = pipe.to(lowercase			)

									pipe.set_progress_bar_config(disable=lowercase			)

									UpperCamelCase__         = self.get_dummy_inputs(lowercase ,      pil_image=lowercase			)

									UpperCamelCase__         = pipe(**lowercase			)
									UpperCamelCase__         = output.images

									UpperCamelCase__         = self.get_dummy_inputs(lowercase ,      pil_image=lowercase			)

									UpperCamelCase__         = pipe(
									    **lowercase ,      return_dict=lowercase ,      )[0]

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

									assert image.shape == (1, 6_4, 6_4, 3)

									UpperCamelCase__         = 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							:						Dict			)							->    Optional[int]:

									'''simple docstring'''







									UpperCamelCase__         = """cpu"""

									UpperCamelCase__         = self.get_dummy_components()

									UpperCamelCase__         = self.pipeline_class(**lowercase			)
									UpperCamelCase__         = pipe.to(lowercase			)

									pipe.set_progress_bar_config(disable=lowercase			)

									UpperCamelCase__         = self.get_dummy_inputs(lowercase ,      pil_image=lowercase			)

									UpperCamelCase__         = pipe(**lowercase			)
									UpperCamelCase__         = output.images

									UpperCamelCase__         = self.get_dummy_inputs(lowercase ,      pil_image=lowercase			)

									UpperCamelCase__         = pipe(
									    **lowercase ,      return_dict=lowercase ,      )[0]

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

									assert image.shape == (1, 6_4, 6_4, 3)

									UpperCamelCase__         = 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							:						Tuple			)							->    Optional[Any]:

									'''simple docstring'''







									UpperCamelCase__         = """cpu"""

									UpperCamelCase__         = self.get_dummy_components()

									UpperCamelCase__         = self.pipeline_class(**lowercase			)
									UpperCamelCase__         = pipe.to(lowercase			)

									pipe.set_progress_bar_config(disable=lowercase			)

									UpperCamelCase__         = self.get_dummy_inputs(lowercase ,      pil_image=lowercase			)
									UpperCamelCase__         = [
									    pipeline_inputs["""image"""],
									    pipeline_inputs["""image"""],
									]

									UpperCamelCase__         = pipe(**lowercase			)
									UpperCamelCase__         = output.images

									UpperCamelCase__         = self.get_dummy_inputs(lowercase ,      pil_image=lowercase			)
									UpperCamelCase__         = [
									    tuple_pipeline_inputs["""image"""],
									    tuple_pipeline_inputs["""image"""],
									]

									UpperCamelCase__         = pipe(
									    **lowercase ,      return_dict=lowercase ,      )[0]

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

									assert image.shape == (2, 6_4, 6_4, 3)

									UpperCamelCase__         = 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							:						Optional[int]			)							->    Tuple:

									'''simple docstring'''







									UpperCamelCase__         = torch.device("""cpu"""			)
									class    _SCREAMING_SNAKE_CASE							:


												'''simple docstring'''





												__a							:							List[Any]					     =					1
									UpperCamelCase__         = self.get_dummy_components()

									UpperCamelCase__         = self.pipeline_class(**lowercase			)
									UpperCamelCase__         = pipe.to(lowercase			)

									pipe.set_progress_bar_config(disable=lowercase			)

									UpperCamelCase__         = torch.Generator(device=lowercase			).manual_seed(0			)
									UpperCamelCase__         = pipe.decoder.dtype
									UpperCamelCase__         = 1

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

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

									UpperCamelCase__         = self.get_dummy_inputs(lowercase ,      pil_image=lowercase			)

									UpperCamelCase__         = pipe(
									    **lowercase ,      decoder_latents=lowercase ,      super_res_latents=lowercase			).images

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

									UpperCamelCase__         = pipe(
									    **lowercase ,      decoder_latents=lowercase ,      super_res_latents=lowercase ,      image_embeddings=lowercase ,      ).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							:						int			)							->    str:

									'''simple docstring'''







									UpperCamelCase__         = torch_device == """cpu"""

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

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






			@skip_mps
			def 				A      (   self							:						int			)							->    List[Any]:

									'''simple docstring'''







									UpperCamelCase__         = torch_device == """cpu"""
									UpperCamelCase__         = True
									UpperCamelCase__         = [
									    """decoder_num_inference_steps""",
									    """super_res_num_inference_steps""",
									]

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






			def 				A      (   self							:						Any			)							->    Optional[Any]:

									'''simple docstring'''







									UpperCamelCase__         = [
									    """decoder_num_inference_steps""",
									    """super_res_num_inference_steps""",
									]

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






			@skip_mps
			def 				A      (   self							:						Any			)							->    Any:

									'''simple docstring'''







									return super().test_dict_tuple_outputs_equivalent()






			@skip_mps
			def 				A      (   self							:						int			)							->    List[str]:

									'''simple docstring'''







									return super().test_save_load_local()







			@skip_mps
			def 				A      (   self							:						int			)							->    List[Any]:

									'''simple docstring'''







									return super().test_save_load_optional_components()
@slow
@require_torch_gpu
class    _SCREAMING_SNAKE_CASE							(		unittest.TestCase	):


			'''simple docstring'''





			def 				A      (   self							:						Union[str, Any]			)							->    Optional[Any]:

									'''simple docstring'''







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







			def 				A      (   self							:						Union[str, Any]			)							->    Union[str, Any]:

									'''simple docstring'''







									UpperCamelCase__         = load_image(
									    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/unclip/cat.png"""			)
									UpperCamelCase__         = load_numpy(
									    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
									    """/unclip/karlo_v1_alpha_cat_variation_fp16.npy"""			)

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

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

									UpperCamelCase__         = output.images[0]

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

									assert_mean_pixel_difference(lowercase ,      lowercase ,      1_5			)








'''simple docstring'''


import enum
import os
from hashlib import shaaaa
from typing import Optional

from .. import config
from .logging import get_logger


lowerCamelCase_				: Any    	=							get_logger(__name__)
class    _SCREAMING_SNAKE_CASE							(		enum.Enum	):


			'''simple docstring'''





			__a							:							Tuple					     =					"all_checks"
			__a							:							List[Any]					     =					"basic_checks"
			__a							:							Dict					     =					"no_checks"
class    _SCREAMING_SNAKE_CASE							(		SCREAMING_SNAKE_CASE	):


			'''simple docstring'''





class    _SCREAMING_SNAKE_CASE							(		SCREAMING_SNAKE_CASE	):


			'''simple docstring'''





class    _SCREAMING_SNAKE_CASE							(		SCREAMING_SNAKE_CASE	):


			'''simple docstring'''





class    _SCREAMING_SNAKE_CASE							(		SCREAMING_SNAKE_CASE	):


			'''simple docstring'''












def 	__magic_name__( _A						,      _A						,      _A=None					):
						'''simple docstring'''





						if expected_checksums is None:
												logger.info("""Unable to verify checksums."""					)
												return
						if len(set(_A					) - set(_A					)					) > 0:
												raise ExpectedMoreDownloadedFiles(str(set(_A					) - set(_A					)					)					)
						if len(set(_A					) - set(_A					)					) > 0:
												raise UnexpectedDownloadedFile(str(set(_A					) - set(_A					)					)					)
						UpperCamelCase__         = [url for url in expected_checksums if expected_checksums[url] != recorded_checksums[url]]
						UpperCamelCase__         = """ for """ + verification_name if verification_name is not None else """"""
						if len(_A					) > 0:
												raise NonMatchingChecksumError(
												    f"Checksums didn't match{for_verification_name}:\n"
												    f"{bad_urls}\n"
												    """Set `verification_mode='no_checks'` to skip checksums verification and ignore this error"""					)
						logger.info("""All the checksums matched successfully""" + for_verification_name					)
class    _SCREAMING_SNAKE_CASE							(		SCREAMING_SNAKE_CASE	):


			'''simple docstring'''





class    _SCREAMING_SNAKE_CASE							(		SCREAMING_SNAKE_CASE	):


			'''simple docstring'''





class    _SCREAMING_SNAKE_CASE							(		SCREAMING_SNAKE_CASE	):


			'''simple docstring'''





class    _SCREAMING_SNAKE_CASE							(		SCREAMING_SNAKE_CASE	):


			'''simple docstring'''












def 	__magic_name__( _A						,      _A					):
						'''simple docstring'''





						if expected_splits is None:
												logger.info("""Unable to verify splits sizes."""					)
												return
						if len(set(_A					) - set(_A					)					) > 0:
												raise ExpectedMoreSplits(str(set(_A					) - set(_A					)					)					)
						if len(set(_A					) - set(_A					)					) > 0:
												raise UnexpectedSplits(str(set(_A					) - set(_A					)					)					)
						UpperCamelCase__         = [
						    {"""expected""": expected_splits[name], """recorded""": recorded_splits[name]}
						    for name in expected_splits
						    if expected_splits[name].num_examples != recorded_splits[name].num_examples
						]
						if len(_A					) > 0:
												raise NonMatchingSplitsSizesError(str(_A					)					)
						logger.info("""All the splits matched successfully."""					)







def 	__magic_name__( _A						,      _A = True					):
						'''simple docstring'''





						if record_checksum:
												UpperCamelCase__         = shaaaa()
												with open(_A						,      """rb"""					) as f:
																		for chunk in iter(lambda: f.read(1 << 20					)						,      B""""""					):
																								m.update(_A					)
																		UpperCamelCase__         = m.hexdigest()
						else:
												UpperCamelCase__         = None
						return {"num_bytes": os.path.getsize(_A					), "checksum": checksum}







def 	__magic_name__( _A					):
						'''simple docstring'''





						if dataset_size and config.IN_MEMORY_MAX_SIZE:
												return dataset_size < config.IN_MEMORY_MAX_SIZE
						else:
												return False
