




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       (							UpperCamelCase_      , unittest.TestCase):
       _a 					=						UnCLIPImageVariationPipeline
       _a 					=						IMAGE_VARIATION_PARAMS - {"""height""", """width""", """guidance_scale"""}
       _a 					=						IMAGE_VARIATION_BATCH_PARAMS

       _a 					=						[
           """generator""",
           """return_dict""",
           """decoder_num_inference_steps""",
           """super_res_num_inference_steps""",
       ]
       _a 					=						False






       @property
       def 					SCREAMING_SNAKE_CASE    (   self:					Optional[Any]				):
         return 32






       @property
       def 					SCREAMING_SNAKE_CASE    (   self:					Dict				):
         return 32






       @property
       def 					SCREAMING_SNAKE_CASE    (   self:					Tuple				):
         return self.time_input_dim






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






       @property
       def 					SCREAMING_SNAKE_CASE    (   self:					Any				):
         return 1_00






       @property
       def 					SCREAMING_SNAKE_CASE    (   self:					Dict				):
         lowercase    :str      	=						CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"				)
         return tokenizer






       @property
       def 					SCREAMING_SNAKE_CASE    (   self:					Dict				):
         torch.manual_seed(0				)
         lowercase    :Any      	=						CLIPTextConfig(
             bos_token_id=0       ,					eos_token_id=2       ,					hidden_size=self.text_embedder_hidden_size       ,					projection_dim=self.text_embedder_hidden_size       ,					intermediate_size=37       ,					layer_norm_eps=1e-0_5       ,					num_attention_heads=4       ,					num_hidden_layers=5       ,					pad_token_id=1       ,					vocab_size=10_00       ,					)
         return CLIPTextModelWithProjection(_a				)






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






       @property
       def 					SCREAMING_SNAKE_CASE    (   self:					int				):
         torch.manual_seed(0				)

         lowercase    :Optional[int]      	=						{
             """clip_embeddings_dim""": self.text_embedder_hidden_size,
             """time_embed_dim""": self.time_embed_dim,
             """cross_attention_dim""": self.cross_attention_dim,
         }

         lowercase    :List[Any]      	=						UnCLIPTextProjModel(**_a				)
         return model






       @property
       def 					SCREAMING_SNAKE_CASE    (   self:					Tuple				):
         torch.manual_seed(0				)

         lowercase    :Tuple      	=						{
             """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""",
         }

         lowercase    :List[str]      	=						UNetaDConditionModel(**_a				)
         return model






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

         lowercase    :List[Any]      	=						UNetaDModel(**self.dummy_super_res_kwargs				)
         return model






       @property
       def 					SCREAMING_SNAKE_CASE    (   self:					int				):
         torch.manual_seed(1				)

         lowercase    :List[Any]      	=						UNetaDModel(**self.dummy_super_res_kwargs				)
         return model






       def 					SCREAMING_SNAKE_CASE    (   self:					str				):
         lowercase    :str      	=						self.dummy_decoder
         lowercase    :Dict      	=						self.dummy_text_proj
         lowercase    :Any      	=						self.dummy_text_encoder
         lowercase    :Optional[Any]      	=						self.dummy_tokenizer
         lowercase    :Any      	=						self.dummy_super_res_first
         lowercase    :Dict      	=						self.dummy_super_res_last

         lowercase    :int      	=						UnCLIPScheduler(
             variance_type="learned_range"       ,					prediction_type="epsilon"       ,					num_train_timesteps=10_00       ,					)

         lowercase    :Dict      	=						UnCLIPScheduler(
             variance_type="fixed_small_log"       ,					prediction_type="epsilon"       ,					num_train_timesteps=10_00       ,					)

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

         lowercase    :Optional[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 					SCREAMING_SNAKE_CASE    (   self:					List[Any]       ,					_lowerCAmelCase:					int       ,					_lowerCAmelCase:					Optional[Any]=0       ,					_lowerCAmelCase:					Union[str, Any]=True				):
         lowercase    :Dict      	=						floats_tensor((1, 3, 32, 32)       ,					rng=random.Random(_a				)				).to(_a				)
         if str(_a				).startswith("mps"				):
           lowercase    :Optional[Any]      	=						torch.manual_seed(_a				)
         else:
           lowercase    :Tuple      	=						torch.Generator(device=_a				).manual_seed(_a				)

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

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






       def 					SCREAMING_SNAKE_CASE    (   self:					List[str]				):
         lowercase    :Tuple      	=						"""cpu"""

         lowercase    :str      	=						self.get_dummy_components()

         lowercase    :Optional[int]      	=						self.pipeline_class(**_a				)
         lowercase    :str      	=						pipe.to(_a				)

         pipe.set_progress_bar_config(disable=_a				)

         lowercase    :Dict      	=						self.get_dummy_inputs(_a       ,					pil_image=_a				)

         lowercase    :Optional[int]      	=						pipe(**_a				)
         lowercase    :List[Any]      	=						output.images

         lowercase    :int      	=						self.get_dummy_inputs(_a       ,					pil_image=_a				)

         lowercase    :Union[str, Any]      	=						pipe(
             **_a       ,					return_dict=_a       ,					)[0]

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

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

         lowercase    :str      	=						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 					SCREAMING_SNAKE_CASE    (   self:					str				):
         lowercase    :Optional[int]      	=						"""cpu"""

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

         lowercase    :Optional[Any]      	=						self.pipeline_class(**_a				)
         lowercase    :Optional[Any]      	=						pipe.to(_a				)

         pipe.set_progress_bar_config(disable=_a				)

         lowercase    :int      	=						self.get_dummy_inputs(_a       ,					pil_image=_a				)

         lowercase    :str      	=						pipe(**_a				)
         lowercase    :List[str]      	=						output.images

         lowercase    :Union[str, Any]      	=						self.get_dummy_inputs(_a       ,					pil_image=_a				)

         lowercase    :Any      	=						pipe(
             **_a       ,					return_dict=_a       ,					)[0]

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

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

         lowercase    :Any      	=						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 					SCREAMING_SNAKE_CASE    (   self:					Optional[Any]				):
         lowercase    :Dict      	=						"""cpu"""

         lowercase    :Dict      	=						self.get_dummy_components()

         lowercase    :str      	=						self.pipeline_class(**_a				)
         lowercase    :Dict      	=						pipe.to(_a				)

         pipe.set_progress_bar_config(disable=_a				)

         lowercase    :Union[str, Any]      	=						self.get_dummy_inputs(_a       ,					pil_image=_a				)
         lowercase    :int      	=						[
             pipeline_inputs["""image"""],
             pipeline_inputs["""image"""],
         ]

         lowercase    :int      	=						pipe(**_a				)
         lowercase    :List[Any]      	=						output.images

         lowercase    :List[str]      	=						self.get_dummy_inputs(_a       ,					pil_image=_a				)
         lowercase    :List[str]      	=						[
             tuple_pipeline_inputs["""image"""],
             tuple_pipeline_inputs["""image"""],
         ]

         lowercase    :int      	=						pipe(
             **_a       ,					return_dict=_a       ,					)[0]

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

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

         lowercase    :Union[str, Any]      	=						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 					SCREAMING_SNAKE_CASE    (   self:					str				):
         lowercase    :List[Any]      	=						torch.device("cpu"				)




         class    __lowerCAmelCase       :
                _a 					=						1







         lowercase    :str      	=						self.get_dummy_components()

         lowercase    :str      	=						self.pipeline_class(**_a				)
         lowercase    :Optional[int]      	=						pipe.to(_a				)

         pipe.set_progress_bar_config(disable=_a				)

         lowercase    :Union[str, Any]      	=						torch.Generator(device=_a				).manual_seed(0				)
         lowercase    :List[Any]      	=						pipe.decoder.dtype
         lowercase    :List[Any]      	=						1

         lowercase    :List[Any]      	=						(
             batch_size,
             pipe.decoder.config.in_channels,
             pipe.decoder.config.sample_size,
             pipe.decoder.config.sample_size,
         )
         lowercase    :Any      	=						pipe.prepare_latents(
             _a       ,					dtype=_a       ,					device=_a       ,					generator=_a       ,					latents=_a       ,					scheduler=DummyScheduler()				)

         lowercase    :Tuple      	=						(
             batch_size,
             pipe.super_res_first.config.in_channels // 2,
             pipe.super_res_first.config.sample_size,
             pipe.super_res_first.config.sample_size,
         )
         lowercase    :List[Any]      	=						pipe.prepare_latents(
             _a       ,					dtype=_a       ,					device=_a       ,					generator=_a       ,					latents=_a       ,					scheduler=DummyScheduler()				)

         lowercase    :int      	=						self.get_dummy_inputs(_a       ,					pil_image=_a				)

         lowercase    :List[Any]      	=						pipe(
             **_a       ,					decoder_latents=_a       ,					super_res_latents=_a				).images

         lowercase    :Tuple      	=						self.get_dummy_inputs(_a       ,					pil_image=_a				)
         # Don't pass image, instead pass embedding
         lowercase    :Dict      	=						pipeline_inputs.pop("image"				)
         lowercase    :Union[str, Any]      	=						pipe.image_encoder(_a				).image_embeds

         lowercase    :Optional[int]      	=						pipe(
             **_a       ,					decoder_latents=_a       ,					super_res_latents=_a       ,					image_embeddings=_a       ,					).images

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






       @skip_mps
       def 					SCREAMING_SNAKE_CASE    (   self:					Any				):
         lowercase    :List[str]      	=						torch_device == """cpu"""

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

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






       @skip_mps
       def 					SCREAMING_SNAKE_CASE    (   self:					Any				):
         lowercase    :List[str]      	=						torch_device == """cpu"""
         lowercase    :List[str]      	=						True
         lowercase    :List[str]      	=						[
             """decoder_num_inference_steps""",
             """super_res_num_inference_steps""",
         ]

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






       def 					SCREAMING_SNAKE_CASE    (   self:					List[str]				):
         lowercase    :Optional[int]      	=						[
             """decoder_num_inference_steps""",
             """super_res_num_inference_steps""",
         ]

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






       @skip_mps
       def 					SCREAMING_SNAKE_CASE    (   self:					int				):
         return super().test_dict_tuple_outputs_equivalent()






       @skip_mps
       def 					SCREAMING_SNAKE_CASE    (   self:					List[str]				):
         return super().test_save_load_local()






       @skip_mps
       def 					SCREAMING_SNAKE_CASE    (   self:					Union[str, Any]				):
         return super().test_save_load_optional_components()







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






       def 					SCREAMING_SNAKE_CASE    (   self:					List[str]				):
         super().tearDown()
         gc.collect()
         torch.cuda.empty_cache()






       def 					SCREAMING_SNAKE_CASE    (   self:					int				):
         lowercase    :Tuple      	=						load_image(
             "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/unclip/cat.png"				)
         lowercase    :str      	=						load_numpy(
             "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
             "/unclip/karlo_v1_alpha_cat_variation_fp16.npy"				)

         lowercase    :List[Any]      	=						UnCLIPImageVariationPipeline.from_pretrained(
             "kakaobrain/karlo-v1-alpha-image-variations"       ,					torch_dtype=torch.floataa				)
         lowercase    :int      	=						pipeline.to(_a				)
         pipeline.set_progress_bar_config(disable=_a				)

         lowercase    :List[Any]      	=						torch.Generator(device="cpu"				).manual_seed(0				)
         lowercase    :Any      	=						pipeline(
             _a       ,					generator=_a       ,					output_type="np"       ,					)

         lowercase    :Any      	=						output.images[0]

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

         assert_mean_pixel_difference(_a       ,					_a       ,					15				)










def       UpperCAmelCase__  (     lowerCamelCase,			lowerCamelCase   ):
  if discount_rate < 0:
    raise ValueError("Discount rate cannot be negative"   )
  if not cash_flows:
    raise ValueError("Cash flows list cannot be empty"   )
  lowercase    :Dict      	=						sum(
      cash_flow / ((1 + discount_rate) ** i) for i, cash_flow in enumerate(lowerCamelCase   )   )
  return round(lowerCamelCase,			ndigits=2   )


if __name__ == "__main__":
     import doctest

     doctest.testmod()



