

'''simple docstring'''







import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image

from diffusers import (
    DDIMScheduler,
    KandinskyVaaImgaImgPipeline,
    KandinskyVaaPriorPipeline,
    UNetaDConditionModel,
    VQModel,
)
from diffusers.utils import floats_tensor, load_image, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()


class 					A__      (      _snake_case						,						unittest.TestCase			):
    lowercase						=		KandinskyVaaImgaImgPipeline
    lowercase						=		["image_embeds", "negative_image_embeds", "image"]
    lowercase						=		[
        "image_embeds",
        "negative_image_embeds",
        "image",
    ]
    lowercase						=		[
        "generator",
        "height",
        "width",
        "strength",
        "guidance_scale",
        "num_inference_steps",
        "return_dict",
        "guidance_scale",
        "num_images_per_prompt",
        "output_type",
        "return_dict",
    ]
    lowercase						=		False



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





           '''simple docstring'''
           return 32



    @property
    def 					snake_case_			(					self       )						->      str:





           '''simple docstring'''
           return 32



    @property
    def 					snake_case_			(					self       )						->      Tuple:





           '''simple docstring'''
           return self.time_input_dim



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





           '''simple docstring'''
           return self.time_input_dim * 4



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





           '''simple docstring'''
           return 100



    @property
    def 					snake_case_			(					self       )						->      List[Any]:





           '''simple docstring'''
           torch.manual_seed(0       )

           A_						=      {
               """in_channels""": 4,
               # Out channels is double in channels because predicts mean and variance
               """out_channels""": 8,
               """addition_embed_type""": """image""",
               """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,
               """encoder_hid_dim""": self.text_embedder_hidden_size,
               """encoder_hid_dim_type""": """image_proj""",
               """cross_attention_dim""": self.cross_attention_dim,
               """attention_head_dim""": 4,
               """resnet_time_scale_shift""": """scale_shift""",
               """class_embed_type""": None,
           }

           A_						=      UNetaDConditionModel(**UpperCamelCase__       )
           return model



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





           '''simple docstring'''
           return {
               "block_out_channels": [32, 64],
               "down_block_types": ["DownEncoderBlock2D", "AttnDownEncoderBlock2D"],
               "in_channels": 3,
               "latent_channels": 4,
               "layers_per_block": 1,
               "norm_num_groups": 8,
               "norm_type": "spatial",
               "num_vq_embeddings": 12,
               "out_channels": 3,
               "up_block_types": [
                   "AttnUpDecoderBlock2D",
                   "UpDecoderBlock2D",
               ],
               "vq_embed_dim": 4,
           }



    @property
    def 					snake_case_			(					self       )						->      Any:





           '''simple docstring'''
           torch.manual_seed(0       )
           A_						=      VQModel(**self.dummy_movq_kwargs       )
           return model



    def 					snake_case_			(					self       )						->      Any:





           '''simple docstring'''
           A_						=      self.dummy_unet
           A_						=      self.dummy_movq

           A_						=      {
               """num_train_timesteps""": 1000,
               """beta_schedule""": """linear""",
               """beta_start""": 0.00085,
               """beta_end""": 0.012,
               """clip_sample""": False,
               """set_alpha_to_one""": False,
               """steps_offset""": 0,
               """prediction_type""": """epsilon""",
               """thresholding""": False,
           }

           A_						=      DDIMScheduler(**UpperCamelCase__       )

           A_						=      {
               """unet""": unet,
               """scheduler""": scheduler,
               """movq""": movq,
           }

           return components



    def 					snake_case_			(					self		,				UpperCamelCase__		,				UpperCamelCase__=0       )						->      List[str]:





           '''simple docstring'''
           A_						=      floats_tensor((1, self.text_embedder_hidden_size)		,				rng=random.Random(UpperCamelCase__       )       ).to(UpperCamelCase__       )
           A_						=      floats_tensor((1, self.text_embedder_hidden_size)		,				rng=random.Random(seed + 1       )       ).to(
               UpperCamelCase__       )
           # create init_image
           A_						=      floats_tensor((1, 3, 64, 64)		,				rng=random.Random(UpperCamelCase__       )       ).to(UpperCamelCase__       )
           A_						=      image.cpu().permute(0		,				2		,				3		,				1       )[0]
           A_						=      Image.fromarray(np.uinta(UpperCamelCase__       )       ).convert("""RGB"""       ).resize((256, 256)       )

           if str(UpperCamelCase__       ).startswith("""mps"""       ):
                  A_						=      torch.manual_seed(UpperCamelCase__       )
           else:
                  A_						=      torch.Generator(device=UpperCamelCase__       ).manual_seed(UpperCamelCase__       )
           A_						=      {
               """image""": init_image,
               """image_embeds""": image_embeds,
               """negative_image_embeds""": negative_image_embeds,
               """generator""": generator,
               """height""": 64,
               """width""": 64,
               """num_inference_steps""": 10,
               """guidance_scale""": 7.0,
               """strength""": 0.2,
               """output_type""": """np""",
           }
           return inputs







    def 					snake_case_			(					self       )						->      Optional[int]:





           '''simple docstring'''
           A_						=      """cpu"""

           A_						=      self.get_dummy_components()

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

           pipe.set_progress_bar_config(disable=UpperCamelCase__       )

           A_						=      pipe(**self.get_dummy_inputs(UpperCamelCase__       )       )
           A_						=      output.images

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

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

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

           A_						=      np.array(
               [0.6199778, 0.63984406, 0.46145785, 0.62944984, 0.5622215, 0.47306132, 0.47441456, 0.4607606, 0.48719263]       )
           assert (
               np.abs(image_slice.flatten() - expected_slice       ).max() < 1e-2
           ), f''' expected_slice {expected_slice}, but got {image_slice.flatten()}'''
           assert (
               np.abs(image_from_tuple_slice.flatten() - expected_slice       ).max() < 1e-2
           ), f''' expected_slice {expected_slice}, but got {image_from_tuple_slice.flatten()}'''


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



    def 					snake_case_			(					self       )						->      Tuple:





           '''simple docstring'''
           # clean up the VRAM after each test
           super().tearDown()
           gc.collect()
           torch.cuda.empty_cache()







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





           '''simple docstring'''
           A_						=      load_numpy(
               """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
               """/kandinskyv22/kandinskyv22_img2img_frog.npy"""       )

           A_						=      load_image(
               """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main""" """/kandinsky/cat.png"""       )
           A_						=      """A red cartoon frog, 4k"""

           A_						=      KandinskyVaaPriorPipeline.from_pretrained(
               """kandinsky-community/kandinsky-2-2-prior"""		,				torch_dtype=torch.floataa       )
           pipe_prior.to(UpperCamelCase__       )

           A_						=      KandinskyVaaImgaImgPipeline.from_pretrained(
               """kandinsky-community/kandinsky-2-2-decoder"""		,				torch_dtype=torch.floataa       )
           A_						=      pipeline.to(UpperCamelCase__       )

           pipeline.set_progress_bar_config(disable=UpperCamelCase__       )

           A_						=      torch.Generator(device="""cpu"""       ).manual_seed(0       )
           A_					,			A_						=      pipe_prior(
               UpperCamelCase__		,				generator=UpperCamelCase__		,				num_inference_steps=5		,				negative_prompt=""""""		,				).to_tuple()

           A_						=      pipeline(
               image=UpperCamelCase__		,				image_embeds=UpperCamelCase__		,				negative_image_embeds=UpperCamelCase__		,				generator=UpperCamelCase__		,				num_inference_steps=100		,				height=768		,				width=768		,				strength=0.2		,				output_type="""np"""		,				)

           A_						=      output.images[0]

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

           assert_mean_pixel_difference(UpperCamelCase__		,				UpperCamelCase__       )



'''simple docstring'''







import tempfile

import torch

from diffusers import IPNDMScheduler

from .test_schedulers import SchedulerCommonTest


class 					A__      (      _snake_case			):
    lowercase						=		(IPNDMScheduler,)
    lowercase						=		(("num_inference_steps", 50),)



    def 					snake_case_			(					self		,				**UpperCamelCase__       )						->      Optional[Any]:





           '''simple docstring'''
           A_						=      {"""num_train_timesteps""": 1000}
           config.update(**UpperCamelCase__       )
           return config



    def 					snake_case_			(					self		,				UpperCamelCase__=0		,				**UpperCamelCase__       )						->      str:





           '''simple docstring'''
           A_						=      dict(self.forward_default_kwargs       )
           A_						=      kwargs.pop("""num_inference_steps"""		,				UpperCamelCase__       )
           A_						=      self.dummy_sample
           A_						=      0.1 * sample
           A_						=      [residual + 0.2, residual + 0.15, residual + 0.1, residual + 0.05]

           for scheduler_class in self.scheduler_classes:
                  A_						=      self.get_scheduler_config(**UpperCamelCase__       )
                  A_						=      scheduler_class(**UpperCamelCase__       )
                  scheduler.set_timesteps(UpperCamelCase__       )
                  # copy over dummy past residuals
                  A_						=      dummy_past_residuals[:]

                  if time_step is None:
                         A_						=      scheduler.timesteps[len(scheduler.timesteps       ) // 2]

                  with tempfile.TemporaryDirectory() as tmpdirname:
                         scheduler.save_config(UpperCamelCase__       )
                         A_						=      scheduler_class.from_pretrained(UpperCamelCase__       )
                         new_scheduler.set_timesteps(UpperCamelCase__       )
                         # copy over dummy past residuals
                         A_						=      dummy_past_residuals[:]

                  A_						=      scheduler.step(UpperCamelCase__		,				UpperCamelCase__		,				UpperCamelCase__		,				**UpperCamelCase__       ).prev_sample
                  A_						=      new_scheduler.step(UpperCamelCase__		,				UpperCamelCase__		,				UpperCamelCase__		,				**UpperCamelCase__       ).prev_sample

                  assert torch.sum(torch.abs(output - new_output       )       ) < 1e-5, "Scheduler outputs are not identical"

                  A_						=      scheduler.step(UpperCamelCase__		,				UpperCamelCase__		,				UpperCamelCase__		,				**UpperCamelCase__       ).prev_sample
                  A_						=      new_scheduler.step(UpperCamelCase__		,				UpperCamelCase__		,				UpperCamelCase__		,				**UpperCamelCase__       ).prev_sample

                  assert torch.sum(torch.abs(output - new_output       )       ) < 1e-5, "Scheduler outputs are not identical"



    def 					snake_case_			(					self       )						->      Optional[int]:





           '''simple docstring'''
           pass



    def 					snake_case_			(					self		,				UpperCamelCase__=0		,				**UpperCamelCase__       )						->      str:





           '''simple docstring'''
           A_						=      dict(self.forward_default_kwargs       )
           A_						=      kwargs.pop("""num_inference_steps"""		,				UpperCamelCase__       )
           A_						=      self.dummy_sample
           A_						=      0.1 * sample
           A_						=      [residual + 0.2, residual + 0.15, residual + 0.1, residual + 0.05]

           for scheduler_class in self.scheduler_classes:
                  A_						=      self.get_scheduler_config()
                  A_						=      scheduler_class(**UpperCamelCase__       )
                  scheduler.set_timesteps(UpperCamelCase__       )

                  # copy over dummy past residuals (must be after setting timesteps)
                  A_						=      dummy_past_residuals[:]

                  if time_step is None:
                         A_						=      scheduler.timesteps[len(scheduler.timesteps       ) // 2]

                  with tempfile.TemporaryDirectory() as tmpdirname:
                         scheduler.save_config(UpperCamelCase__       )
                         A_						=      scheduler_class.from_pretrained(UpperCamelCase__       )
                         # copy over dummy past residuals
                         new_scheduler.set_timesteps(UpperCamelCase__       )

                         # copy over dummy past residual (must be after setting timesteps)
                         A_						=      dummy_past_residuals[:]

                  A_						=      scheduler.step(UpperCamelCase__		,				UpperCamelCase__		,				UpperCamelCase__		,				**UpperCamelCase__       ).prev_sample
                  A_						=      new_scheduler.step(UpperCamelCase__		,				UpperCamelCase__		,				UpperCamelCase__		,				**UpperCamelCase__       ).prev_sample

                  assert torch.sum(torch.abs(output - new_output       )       ) < 1e-5, "Scheduler outputs are not identical"

                  A_						=      scheduler.step(UpperCamelCase__		,				UpperCamelCase__		,				UpperCamelCase__		,				**UpperCamelCase__       ).prev_sample
                  A_						=      new_scheduler.step(UpperCamelCase__		,				UpperCamelCase__		,				UpperCamelCase__		,				**UpperCamelCase__       ).prev_sample

                  assert torch.sum(torch.abs(output - new_output       )       ) < 1e-5, "Scheduler outputs are not identical"



    def 					snake_case_			(					self		,				**UpperCamelCase__       )						->      Tuple:





           '''simple docstring'''
           A_						=      self.scheduler_classes[0]
           A_						=      self.get_scheduler_config(**UpperCamelCase__       )
           A_						=      scheduler_class(**UpperCamelCase__       )

           A_						=      10
           A_						=      self.dummy_model()
           A_						=      self.dummy_sample_deter
           scheduler.set_timesteps(UpperCamelCase__       )

           for i, t in enumerate(scheduler.timesteps       ):
                  A_						=      model(UpperCamelCase__		,				UpperCamelCase__       )
                  A_						=      scheduler.step(UpperCamelCase__		,				UpperCamelCase__		,				UpperCamelCase__       ).prev_sample

           for i, t in enumerate(scheduler.timesteps       ):
                  A_						=      model(UpperCamelCase__		,				UpperCamelCase__       )
                  A_						=      scheduler.step(UpperCamelCase__		,				UpperCamelCase__		,				UpperCamelCase__       ).prev_sample

           return sample



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





           '''simple docstring'''
           A_						=      dict(self.forward_default_kwargs       )

           A_						=      kwargs.pop("""num_inference_steps"""		,				UpperCamelCase__       )

           for scheduler_class in self.scheduler_classes:
                  A_						=      self.get_scheduler_config()
                  A_						=      scheduler_class(**UpperCamelCase__       )

                  A_						=      self.dummy_sample
                  A_						=      0.1 * sample

                  if num_inference_steps is not None and hasattr(UpperCamelCase__		,				"""set_timesteps"""       ):
                         scheduler.set_timesteps(UpperCamelCase__       )
                  elif num_inference_steps is not None and not hasattr(UpperCamelCase__		,				"""set_timesteps"""       ):
                         A_						=      num_inference_steps

                  # copy over dummy past residuals (must be done after set_timesteps)
                  A_						=      [residual + 0.2, residual + 0.15, residual + 0.1, residual + 0.05]
                  A_						=      dummy_past_residuals[:]

                  A_						=      scheduler.timesteps[5]
                  A_						=      scheduler.timesteps[6]

                  A_						=      scheduler.step(UpperCamelCase__		,				UpperCamelCase__		,				UpperCamelCase__		,				**UpperCamelCase__       ).prev_sample
                  A_						=      scheduler.step(UpperCamelCase__		,				UpperCamelCase__		,				UpperCamelCase__		,				**UpperCamelCase__       ).prev_sample

                  self.assertEqual(output_a.shape		,				sample.shape       )
                  self.assertEqual(output_a.shape		,				output_a.shape       )

                  A_						=      scheduler.step(UpperCamelCase__		,				UpperCamelCase__		,				UpperCamelCase__		,				**UpperCamelCase__       ).prev_sample
                  A_						=      scheduler.step(UpperCamelCase__		,				UpperCamelCase__		,				UpperCamelCase__		,				**UpperCamelCase__       ).prev_sample

                  self.assertEqual(output_a.shape		,				sample.shape       )
                  self.assertEqual(output_a.shape		,				output_a.shape       )



    def 					snake_case_			(					self       )						->      Any:





           '''simple docstring'''
           for timesteps in [100, 1000]:
                  self.check_over_configs(num_train_timesteps=UpperCamelCase__		,				time_step=UpperCamelCase__       )



    def 					snake_case_			(					self       )						->      Any:





           '''simple docstring'''
           for t, num_inference_steps in zip([1, 5, 10]		,				[10, 50, 100]       ):
                  self.check_over_forward(num_inference_steps=UpperCamelCase__		,				time_step=UpperCamelCase__       )







    def 					snake_case_			(					self       )						->      Dict:





           '''simple docstring'''
           A_						=      self.full_loop()
           A_						=      torch.mean(torch.abs(UpperCamelCase__       )       )

           assert abs(result_mean.item() - 2540529       ) < 10
