

'''simple docstring'''

import random
import unittest

import numpy as np
import torch
from PIL import Image
from torch import nn
from transformers import (
    CLIPImageProcessor,
    CLIPTextConfig,
    CLIPTextModelWithProjection,
    CLIPTokenizer,
    CLIPVisionConfig,
    CLIPVisionModelWithProjection,
)

from diffusers import KandinskyVaaPriorEmbaEmbPipeline, PriorTransformer, UnCLIPScheduler
from diffusers.utils import floats_tensor, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, skip_mps

from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()


class 					lowerCAmelCase       ( A				,     unittest.TestCase				):
       lowerCAmelCase_				=	KandinskyVaaPriorEmbaEmbPipeline
       lowerCAmelCase_				=	["prompt", "image"]
       lowerCAmelCase_				=	["prompt", "image"]
       lowerCAmelCase_				=	[
           "num_images_per_prompt",
           "strength",
           "generator",
           "num_inference_steps",
           "latents",
           "negative_prompt",
           "guidance_scale",
           "output_type",
           "return_dict",
       ]
       lowerCAmelCase_				=	False







       @property
       def      snake_case				(							self     :	List[str] ):







            """simple docstring"""







            return 32







       @property
       def      snake_case				(							self     :	Union[str, Any] ):







            """simple docstring"""







            return 32







       @property
       def      snake_case				(							self     :	List[Any] ):







            """simple docstring"""







            return self.time_input_dim







       @property
       def      snake_case				(							self     :	Union[str, Any] ):







            """simple docstring"""







            return self.time_input_dim * 4







       @property
       def      snake_case				(							self     :	Union[str, Any] ):







            """simple docstring"""







            return 100







       @property
       def      snake_case				(							self     :	str ):







            """simple docstring"""







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







       @property
       def      snake_case				(							self     :	List[Any] ):







            """simple docstring"""







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







       @property
       def      snake_case				(							self     :	int ):







            """simple docstring"""







            torch.manual_seed(0 )

            __lowercase     						={
                'num_attention_heads': 2,
                'attention_head_dim': 12,
                'embedding_dim': self.text_embedder_hidden_size,
                'num_layers': 1,
            }

            __lowercase     						=PriorTransformer(**__lowercase )
            # clip_std and clip_mean is initialized to be 0 so PriorTransformer.post_process_latents will always return 0 - set clip_std to be 1 so it won't return 0
            __lowercase     						=nn.Parameter(torch.ones(model.clip_std.shape ) )
            return model







       @property
       def      snake_case				(							self     :	Dict ):







            """simple docstring"""







            torch.manual_seed(0 )
            __lowercase     						=CLIPVisionConfig(
                hidden_size=self.text_embedder_hidden_size  , image_size=224  , projection_dim=self.text_embedder_hidden_size  , intermediate_size=37  , num_attention_heads=4  , num_channels=3  , num_hidden_layers=5  , patch_size=14  , )

            __lowercase     						=CLIPVisionModelWithProjection(__lowercase )
            return model







       @property
       def      snake_case				(							self     :	Tuple ):







            """simple docstring"""







            __lowercase     						=CLIPImageProcessor(
                crop_size=224  , do_center_crop=__lowercase  , do_normalize=__lowercase  , do_resize=__lowercase  , image_mean=[0.4_8_1_4_5_4_6_6, 0.4_5_7_8_2_7_5, 0.4_0_8_2_1_0_7_3]  , image_std=[0.2_6_8_6_2_9_5_4, 0.2_6_1_3_0_2_5_8, 0.2_7_5_7_7_7_1_1]  , resample=3  , size=224  , )

            return image_processor







       def      snake_case				(							self     :	Optional[Any] ):







            """simple docstring"""







            __lowercase     						=self.dummy_prior
            __lowercase     						=self.dummy_image_encoder
            __lowercase     						=self.dummy_text_encoder
            __lowercase     						=self.dummy_tokenizer
            __lowercase     						=self.dummy_image_processor

            __lowercase     						=UnCLIPScheduler(
                variance_type='fixed_small_log'  , prediction_type='sample'  , num_train_timesteps=1000  , clip_sample=__lowercase  , clip_sample_range=1_0.0  , )

            __lowercase     						={
                'prior': prior,
                'image_encoder': image_encoder,
                'text_encoder': text_encoder,
                'tokenizer': tokenizer,
                'scheduler': scheduler,
                'image_processor': image_processor,
            }

            return components







       def      snake_case				(							self     :	str  , __lowercase     :	Tuple  , __lowercase     :	List[str]=0 ):







            """simple docstring"""







            if str(__lowercase ).startswith('mps' ):
                 __lowercase     						=torch.manual_seed(__lowercase )
            else:
                 __lowercase     						=torch.Generator(device=__lowercase ).manual_seed(__lowercase )

            __lowercase     						=floats_tensor((1, 3, 64, 64)  , rng=random.Random(__lowercase ) ).to(__lowercase )
            __lowercase     						=image.cpu().permute(0  , 2  , 3  , 1 )[0]
            __lowercase     						=Image.fromarray(np.uinta(__lowercase ) ).convert('RGB' ).resize((256, 256) )

            __lowercase     						={
                'prompt': 'horse',
                'image': init_image,
                'strength': 0.5,
                'generator': generator,
                'guidance_scale': 4.0,
                'num_inference_steps': 2,
                'output_type': 'np',
            }
            return inputs







       def      snake_case				(							self     :	Optional[int] ):







            """simple docstring"""







            __lowercase     						='cpu'

            __lowercase     						=self.get_dummy_components()

            __lowercase     						=self.pipeline_class(**__lowercase )
            __lowercase     						=pipe.to(__lowercase )

            pipe.set_progress_bar_config(disable=__lowercase )

            __lowercase     						=pipe(**self.get_dummy_inputs(__lowercase ) )
            __lowercase     						=output.image_embeds

            __lowercase     						=pipe(
                **self.get_dummy_inputs(__lowercase )  , return_dict=__lowercase  , )[0]

            __lowercase     						=image[0, -10:]
            __lowercase     						=image_from_tuple[0, -10:]

            assert image.shape == (1, 32)

            __lowercase     						=np.array(
                [
                    0.1_0_7_1_2_8_4,
                    1.3_3_3_0_2_7_1,
                    0.6_1_2_6_0_2_2_3,
                    -0.6_6_9_1_0_6_5,
                    -0.3_8_4_6_8_5_2,
                    -1.0_3_0_3_6_6_1,
                    0.2_2_7_1_6_1_1_1,
                    0.0_3_3_4_8_9_0_1,
                    0.3_0_0_4_0_6_7_5,
                    -0.2_4_8_0_5_0_2_9,
                ] )

            assert np.abs(image_slice.flatten() - expected_slice ).max() < 1E-2
            assert np.abs(image_from_tuple_slice.flatten() - expected_slice ).max() < 1E-2







       @skip_mps
       def      snake_case				(							self     :	List[str] ):







            """simple docstring"""







            __lowercase     						=torch_device == 'cpu'
            __lowercase     						=True
            __lowercase     						=False

            self._test_inference_batch_single_identical(
                test_max_difference=__lowercase  , relax_max_difference=__lowercase  , test_mean_pixel_difference=__lowercase  , )







       @skip_mps
       def      snake_case				(							self     :	Union[str, Any] ):







            """simple docstring"""







            __lowercase     						=torch_device == 'cpu'
            __lowercase     						=False

            self._test_attention_slicing_forward_pass(
                test_max_difference=__lowercase  , test_mean_pixel_difference=__lowercase  , )





'''simple docstring'''

import torch

from diffusers import DDIMScheduler

from .test_schedulers import SchedulerCommonTest


class 					lowerCAmelCase       ( A				):
       lowerCAmelCase_				=	(DDIMScheduler,)
       lowerCAmelCase_				=	(("eta", 0.0), ("num_inference_steps", 5_0))







       def      snake_case				(							self     :	List[Any]  , **__lowercase     :	Optional[Any] ):







            """simple docstring"""







            __lowercase     						={
                'num_train_timesteps': 1000,
                'beta_start': 0.0_0_0_1,
                'beta_end': 0.0_2,
                'beta_schedule': 'linear',
                'clip_sample': True,
            }

            config.update(**__lowercase )
            return config







       def      snake_case				(							self     :	Optional[Any]  , **__lowercase     :	str ):







            """simple docstring"""







            __lowercase     						=self.scheduler_classes[0]
            __lowercase     						=self.get_scheduler_config(**__lowercase )
            __lowercase     						=scheduler_class(**__lowercase )

            __lowercase						,       __lowercase     						=10, 0.0

            __lowercase     						=self.dummy_model()
            __lowercase     						=self.dummy_sample_deter

            scheduler.set_timesteps(__lowercase )

            for t in scheduler.timesteps:
                 __lowercase     						=model(__lowercase  , __lowercase )
                 __lowercase     						=scheduler.step(__lowercase  , __lowercase  , __lowercase  , __lowercase ).prev_sample

            return sample







       def      snake_case				(							self     :	List[Any] ):







            """simple docstring"""







            for timesteps in [100, 500, 1000]:
                 self.check_over_configs(num_train_timesteps=__lowercase )







       def      snake_case				(							self     :	List[Any] ):







            """simple docstring"""







            for steps_offset in [0, 1]:
                 self.check_over_configs(steps_offset=__lowercase )

            __lowercase     						=self.scheduler_classes[0]
            __lowercase     						=self.get_scheduler_config(steps_offset=1 )
            __lowercase     						=scheduler_class(**__lowercase )
            scheduler.set_timesteps(5 )
            assert torch.equal(scheduler.timesteps  , torch.LongTensor([801, 601, 401, 201, 1] ) )







       def      snake_case				(							self     :	List[str] ):







            """simple docstring"""







            for beta_start, beta_end in zip([0.0_0_0_1, 0.0_0_1, 0.0_1, 0.1]  , [0.0_0_2, 0.0_2, 0.2, 2] ):
                 self.check_over_configs(beta_start=__lowercase  , beta_end=__lowercase )







       def      snake_case				(							self     :	str ):







            """simple docstring"""







            for schedule in ["linear", "squaredcos_cap_v2"]:
                 self.check_over_configs(beta_schedule=__lowercase )







       def      snake_case				(							self     :	str ):







            """simple docstring"""







            for prediction_type in ["epsilon", "v_prediction"]:
                 self.check_over_configs(prediction_type=__lowercase )







       def      snake_case				(							self     :	List[str] ):







            """simple docstring"""







            for clip_sample in [True, False]:
                 self.check_over_configs(clip_sample=__lowercase )







       def      snake_case				(							self     :	Dict ):







            """simple docstring"""







            for timestep_spacing in ["trailing", "leading"]:
                 self.check_over_configs(timestep_spacing=__lowercase )







       def      snake_case				(							self     :	Tuple ):







            """simple docstring"""







            for rescale_betas_zero_snr in [True, False]:
                 self.check_over_configs(rescale_betas_zero_snr=__lowercase )







       def      snake_case				(							self     :	int ):







            """simple docstring"""







            self.check_over_configs(thresholding=__lowercase )
            for threshold in [0.5, 1.0, 2.0]:
                 for prediction_type in ["epsilon", "v_prediction"]:
                      self.check_over_configs(
                          thresholding=__lowercase  , prediction_type=__lowercase  , sample_max_value=__lowercase  , )







       def      snake_case				(							self     :	Dict ):







            """simple docstring"""







            for t in [1, 10, 49]:
                 self.check_over_forward(time_step=__lowercase )







       def      snake_case				(							self     :	List[str] ):







            """simple docstring"""







            for t, num_inference_steps in zip([1, 10, 50]  , [10, 50, 500] ):
                 self.check_over_forward(time_step=__lowercase  , num_inference_steps=__lowercase )







       def      snake_case				(							self     :	Tuple ):







            """simple docstring"""







            for t, eta in zip([1, 10, 49]  , [0.0, 0.5, 1.0] ):
                 self.check_over_forward(time_step=__lowercase  , eta=__lowercase )







       def      snake_case				(							self     :	Optional[int] ):







            """simple docstring"""







            __lowercase     						=self.scheduler_classes[0]
            __lowercase     						=self.get_scheduler_config()
            __lowercase     						=scheduler_class(**__lowercase )

            assert torch.sum(torch.abs(scheduler._get_variance(0  , 0 ) - 0.0 ) ) < 1E-5
            assert torch.sum(torch.abs(scheduler._get_variance(420  , 400 ) - 0.1_4_7_7_1 ) ) < 1E-5
            assert torch.sum(torch.abs(scheduler._get_variance(980  , 960 ) - 0.3_2_4_6_0 ) ) < 1E-5
            assert torch.sum(torch.abs(scheduler._get_variance(0  , 0 ) - 0.0 ) ) < 1E-5
            assert torch.sum(torch.abs(scheduler._get_variance(487  , 486 ) - 0.0_0_9_7_9 ) ) < 1E-5
            assert torch.sum(torch.abs(scheduler._get_variance(999  , 998 ) - 0.0_2 ) ) < 1E-5







       def      snake_case				(							self     :	Tuple ):







            """simple docstring"""







            __lowercase     						=self.full_loop()

            __lowercase     						=torch.sum(torch.abs(__lowercase ) )
            __lowercase     						=torch.mean(torch.abs(__lowercase ) )

            assert abs(result_sum.item() - 1_7_2.0_0_6_7 ) < 1E-2
            assert abs(result_mean.item() - 0.2_2_3_9_6_7 ) < 1E-3







       def      snake_case				(							self     :	List[str] ):







            """simple docstring"""







            __lowercase     						=self.full_loop(prediction_type='v_prediction' )

            __lowercase     						=torch.sum(torch.abs(__lowercase ) )
            __lowercase     						=torch.mean(torch.abs(__lowercase ) )

            assert abs(result_sum.item() - 5_2.5_3_0_2 ) < 1E-2
            assert abs(result_mean.item() - 0.0_6_8_4 ) < 1E-3







       def      snake_case				(							self     :	Optional[Any] ):







            """simple docstring"""







            __lowercase     						=self.full_loop(set_alpha_to_one=__lowercase  , beta_start=0.0_1 )
            __lowercase     						=torch.sum(torch.abs(__lowercase ) )
            __lowercase     						=torch.mean(torch.abs(__lowercase ) )

            assert abs(result_sum.item() - 1_4_9.8_2_9_5 ) < 1E-2
            assert abs(result_mean.item() - 0.1_9_5_1 ) < 1E-3







       def      snake_case				(							self     :	Dict ):







            """simple docstring"""







            __lowercase     						=self.full_loop(set_alpha_to_one=__lowercase  , beta_start=0.0_1 )
            __lowercase     						=torch.sum(torch.abs(__lowercase ) )
            __lowercase     						=torch.mean(torch.abs(__lowercase ) )

            assert abs(result_sum.item() - 1_4_9.0_7_8_4 ) < 1E-2
            assert abs(result_mean.item() - 0.1_9_4_1 ) < 1E-3


