




'''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      lowerCAmelCase (			__lowerCAmelCase					, unittest.TestCase):
       __lowercase	:   Optional[Any]										=      UnCLIPImageVariationPipeline
       __lowercase	:   Union[str, Any]										=      IMAGE_VARIATION_PARAMS - {'''height''', '''width''', '''guidance_scale'''}
       __lowercase	:   Any										=      IMAGE_VARIATION_BATCH_PARAMS

       __lowercase	:   Dict										=      [
           '''generator''',
           '''return_dict''',
           '''decoder_num_inference_steps''',
           '''super_res_num_inference_steps''',
       ]
       __lowercase	:   int										=      False


       @property
       def    lowerCAmelCase   ( self	)   ->	str:



            '''simple docstring'''




            return 32


       @property
       def    lowerCAmelCase   ( self	)   ->	str:



            '''simple docstring'''




            return 32


       @property
       def    lowerCAmelCase   ( self	)   ->	Any:



            '''simple docstring'''




            return self.time_input_dim


       @property
       def    lowerCAmelCase   ( self	)   ->	str:



            '''simple docstring'''




            return self.time_input_dim * 4


       @property
       def    lowerCAmelCase   ( self	)   ->	Tuple:



            '''simple docstring'''




            return 100


       @property
       def    lowerCAmelCase   ( self	)   ->	str:



            '''simple docstring'''




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


       @property
       def    lowerCAmelCase   ( self	)   ->	Any:



            '''simple docstring'''




            torch.manual_seed(0	)
            __snake_case            =	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(__SCREAMING_SNAKE_CASE	)


       @property
       def    lowerCAmelCase   ( self	)   ->	Dict:



            '''simple docstring'''




            torch.manual_seed(0	)
            __snake_case            =	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(__SCREAMING_SNAKE_CASE	)


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



            '''simple docstring'''




            torch.manual_seed(0	)

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

            __snake_case            =	UnCLIPTextProjModel(**__SCREAMING_SNAKE_CASE	)
            return model


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



            '''simple docstring'''




            torch.manual_seed(0	)

            __snake_case            =	{
                '''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''',
            }

            __snake_case            =	UNetaDConditionModel(**__SCREAMING_SNAKE_CASE	)
            return model


       @property
       def    lowerCAmelCase   ( self	)   ->	int:



            '''simple docstring'''




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



            '''simple docstring'''




            torch.manual_seed(0	)

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


       @property
       def    lowerCAmelCase   ( self	)   ->	int:



            '''simple docstring'''




            torch.manual_seed(1	)

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


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



            '''simple docstring'''




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

            __snake_case            =	UnCLIPScheduler(
                variance_type='''learned_range'''     ,	prediction_type='''epsilon'''     ,	num_train_timesteps=1000     ,	)

            __snake_case            =	UnCLIPScheduler(
                variance_type='''fixed_small_log'''     ,	prediction_type='''epsilon'''     ,	num_train_timesteps=1000     ,	)

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

            __snake_case            =	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     ,	__SCREAMING_SNAKE_CASE     ,	__SCREAMING_SNAKE_CASE=0     ,	__SCREAMING_SNAKE_CASE=True	)   ->	Union[str, Any]:



            '''simple docstring'''




            __snake_case            =	floats_tensor((1, 3, 32, 32)     ,	rng=random.Random(__SCREAMING_SNAKE_CASE	)	).to(__SCREAMING_SNAKE_CASE	)
            if str(__SCREAMING_SNAKE_CASE	).startswith('''mps'''	):
                 __snake_case            =	torch.manual_seed(__SCREAMING_SNAKE_CASE	)
            else:
                 __snake_case            =	torch.Generator(device=__SCREAMING_SNAKE_CASE	).manual_seed(__SCREAMING_SNAKE_CASE	)

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

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


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



            '''simple docstring'''




            __snake_case            =	'''cpu'''

            __snake_case            =	self.get_dummy_components()

            __snake_case            =	self.pipeline_class(**__SCREAMING_SNAKE_CASE	)
            __snake_case            =	pipe.to(__SCREAMING_SNAKE_CASE	)

            pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE	)

            __snake_case            =	self.get_dummy_inputs(__SCREAMING_SNAKE_CASE     ,	pil_image=__SCREAMING_SNAKE_CASE	)

            __snake_case            =	pipe(**__SCREAMING_SNAKE_CASE	)
            __snake_case            =	output.images

            __snake_case            =	self.get_dummy_inputs(__SCREAMING_SNAKE_CASE     ,	pil_image=__SCREAMING_SNAKE_CASE	)

            __snake_case            =	pipe(
                **__SCREAMING_SNAKE_CASE     ,	return_dict=__SCREAMING_SNAKE_CASE     ,	)[0]

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

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

            __snake_case            =	np.array(
                [
                    0.9_997,
                    0.0_002,
                    0.9_997,
                    0.9_997,
                    0.9_969,
                    0.0_023,
                    0.9_997,
                    0.9_969,
                    0.9_970,
                ]	)

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



            '''simple docstring'''




            __snake_case            =	'''cpu'''

            __snake_case            =	self.get_dummy_components()

            __snake_case            =	self.pipeline_class(**__SCREAMING_SNAKE_CASE	)
            __snake_case            =	pipe.to(__SCREAMING_SNAKE_CASE	)

            pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE	)

            __snake_case            =	self.get_dummy_inputs(__SCREAMING_SNAKE_CASE     ,	pil_image=__SCREAMING_SNAKE_CASE	)

            __snake_case            =	pipe(**__SCREAMING_SNAKE_CASE	)
            __snake_case            =	output.images

            __snake_case            =	self.get_dummy_inputs(__SCREAMING_SNAKE_CASE     ,	pil_image=__SCREAMING_SNAKE_CASE	)

            __snake_case            =	pipe(
                **__SCREAMING_SNAKE_CASE     ,	return_dict=__SCREAMING_SNAKE_CASE     ,	)[0]

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

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

            __snake_case            =	np.array([0.9_997, 0.0_003, 0.9_997, 0.9_997, 0.9_970, 0.0_024, 0.9_997, 0.9_971, 0.9_971]	)

            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	)   ->	Tuple:



            '''simple docstring'''




            __snake_case            =	'''cpu'''

            __snake_case            =	self.get_dummy_components()

            __snake_case            =	self.pipeline_class(**__SCREAMING_SNAKE_CASE	)
            __snake_case            =	pipe.to(__SCREAMING_SNAKE_CASE	)

            pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE	)

            __snake_case            =	self.get_dummy_inputs(__SCREAMING_SNAKE_CASE     ,	pil_image=__SCREAMING_SNAKE_CASE	)
            __snake_case            =	[
                pipeline_inputs['''image'''],
                pipeline_inputs['''image'''],
            ]

            __snake_case            =	pipe(**__SCREAMING_SNAKE_CASE	)
            __snake_case            =	output.images

            __snake_case            =	self.get_dummy_inputs(__SCREAMING_SNAKE_CASE     ,	pil_image=__SCREAMING_SNAKE_CASE	)
            __snake_case            =	[
                tuple_pipeline_inputs['''image'''],
                tuple_pipeline_inputs['''image'''],
            ]

            __snake_case            =	pipe(
                **__SCREAMING_SNAKE_CASE     ,	return_dict=__SCREAMING_SNAKE_CASE     ,	)[0]

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

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

            __snake_case            =	np.array(
                [
                    0.9_997,
                    0.9_989,
                    0.0_008,
                    0.0_021,
                    0.9_960,
                    0.0_018,
                    0.0_014,
                    0.0_002,
                    0.9_933,
                ]	)

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



            '''simple docstring'''




            __snake_case            =	torch.device('''cpu'''	)





            class      lowerCAmelCase :
                   __lowercase	:   List[Any]										=      1





            __snake_case            =	self.get_dummy_components()

            __snake_case            =	self.pipeline_class(**__SCREAMING_SNAKE_CASE	)
            __snake_case            =	pipe.to(__SCREAMING_SNAKE_CASE	)

            pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE	)

            __snake_case            =	torch.Generator(device=__SCREAMING_SNAKE_CASE	).manual_seed(0	)
            __snake_case            =	pipe.decoder.dtype
            __snake_case            =	1

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

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

            __snake_case            =	self.get_dummy_inputs(__SCREAMING_SNAKE_CASE     ,	pil_image=__SCREAMING_SNAKE_CASE	)

            __snake_case            =	pipe(
                **__SCREAMING_SNAKE_CASE     ,	decoder_latents=__SCREAMING_SNAKE_CASE     ,	super_res_latents=__SCREAMING_SNAKE_CASE	).images

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

            __snake_case            =	pipe(
                **__SCREAMING_SNAKE_CASE     ,	decoder_latents=__SCREAMING_SNAKE_CASE     ,	super_res_latents=__SCREAMING_SNAKE_CASE     ,	image_embeddings=__SCREAMING_SNAKE_CASE     ,	).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	)   ->	Optional[int]:



            '''simple docstring'''




            __snake_case            =	torch_device == '''cpu'''

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

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


       @skip_mps
       def    lowerCAmelCase   ( self	)   ->	Dict:



            '''simple docstring'''




            __snake_case            =	torch_device == '''cpu'''
            __snake_case            =	True
            __snake_case            =	[
                '''decoder_num_inference_steps''',
                '''super_res_num_inference_steps''',
            ]

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


       def    lowerCAmelCase   ( self	)   ->	Tuple:



            '''simple docstring'''




            __snake_case            =	[
                '''decoder_num_inference_steps''',
                '''super_res_num_inference_steps''',
            ]

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


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



            '''simple docstring'''




            return super().test_dict_tuple_outputs_equivalent()


       @skip_mps
       def    lowerCAmelCase   ( self	)   ->	Union[str, Any]:



            '''simple docstring'''




            return super().test_save_load_local()




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



            '''simple docstring'''




            return super().test_save_load_optional_components()





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


       def    lowerCAmelCase   ( self	)   ->	List[Any]:



            '''simple docstring'''




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




       def    lowerCAmelCase   ( self	)   ->	Tuple:



            '''simple docstring'''




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

            __snake_case            =	UnCLIPImageVariationPipeline.from_pretrained(
                '''kakaobrain/karlo-v1-alpha-image-variations'''     ,	torch_dtype=torch.floataa	)
            __snake_case            =	pipeline.to(__SCREAMING_SNAKE_CASE	)
            pipeline.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE	)

            __snake_case            =	torch.Generator(device='''cpu'''	).manual_seed(0	)
            __snake_case            =	pipeline(
                __SCREAMING_SNAKE_CASE     ,	generator=__SCREAMING_SNAKE_CASE     ,	output_type='''np'''     ,	)

            __snake_case            =	output.images[0]

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

            assert_mean_pixel_difference(__SCREAMING_SNAKE_CASE     ,	__SCREAMING_SNAKE_CASE     ,	15	)












'''simple docstring'''



import inspect
import warnings
from typing import Any, Dict, Optional, Union

from packaging import version







def       _UpperCamelCase			(*_lowerCamelCase				:  str ,						_lowerCamelCase				:  Optional[Union[Dict, Any]] = None ,						_lowerCamelCase				:  List[Any]=True ,						_lowerCamelCase				:  str=2						)-> str:




     '''simple docstring'''





     from .. import __version__

     __snake_case            =	take_from
     __snake_case            =	()
     if not isinstance(args[0] ,						_lowerCamelCase						):
          __snake_case            =	(args,)

     for attribute, version_name, message in args:
          if version.parse(version.parse(_lowerCamelCase						).base_version						) >= version.parse(_lowerCamelCase						):
               raise ValueError(
                   f'''The deprecation tuple {(attribute, version_name, message)} should be removed since diffusers\''''
                   f''' version {__version__} is >= {version_name}'''						)

          __snake_case            =	None
          if isinstance(_lowerCamelCase ,						_lowerCamelCase						) and attribute in deprecated_kwargs:
               values += (deprecated_kwargs.pop(_lowerCamelCase						),)
               __snake_case            =	f'''The `{attribute}` argument is deprecated and will be removed in version {version_name}.'''
          elif hasattr(_lowerCamelCase ,						_lowerCamelCase						):
               values += (getattr(_lowerCamelCase ,						_lowerCamelCase						),)
               __snake_case            =	f'''The `{attribute}` attribute is deprecated and will be removed in version {version_name}.'''
          elif deprecated_kwargs is None:
               __snake_case            =	f'''`{attribute}` is deprecated and will be removed in version {version_name}.'''

          if warning is not None:
               __snake_case            =	warning + ''' ''' if standard_warn else ''''''
               warnings.warn(warning + message ,						_lowerCamelCase ,						stacklevel=_lowerCamelCase						)

     if isinstance(_lowerCamelCase ,						_lowerCamelCase						) and len(_lowerCamelCase						) > 0:
          __snake_case            =	inspect.getouterframes(inspect.currentframe()						)[1]
          __snake_case            =	call_frame.filename
          __snake_case            =	call_frame.lineno
          __snake_case            =	call_frame.function
          __snake_case       ,     __snake_case            =	next(iter(deprecated_kwargs.items()						)						)
          raise TypeError(f'''{function} in {filename} line {line_number-1} got an unexpected keyword argument `{key}`'''						)

     if len(_lowerCamelCase						) == 0:
          return
     elif len(_lowerCamelCase						) == 1:
          return values[0]
     return values






