





'''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_       (    _lowerCAmelCase     ,						unittest.TestCase							):
     __A								=      KandinskyVaaImgaImgPipeline
     __A								=      ["image_embeds", "negative_image_embeds", "image"]
     __A								=      [
         "image_embeds",
         "negative_image_embeds",
         "image",
     ]
     __A								=      [
         "generator",
         "height",
         "width",
         "strength",
         "guidance_scale",
         "num_inference_steps",
         "return_dict",
         "guidance_scale",
         "num_images_per_prompt",
         "output_type",
         "return_dict",
     ]
     __A								=      False





     @property
     def 							lowercase__				(			self		: Dict						):





        """simple docstring"""




        return 32





     @property
     def 							lowercase__				(			self		: Optional[Any]						):





        """simple docstring"""




        return 32





     @property
     def 							lowercase__				(			self		: Optional[Any]						):





        """simple docstring"""




        return self.time_input_dim





     @property
     def 							lowercase__				(			self		: str						):





        """simple docstring"""




        return self.time_input_dim * 4





     @property
     def 							lowercase__				(			self		: Optional[int]						):





        """simple docstring"""




        return 100





     @property
     def 							lowercase__				(			self		: Any						):





        """simple docstring"""




        torch.manual_seed(0						)

        lowercase_      :str         =			{
            "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,
        }

        lowercase_      :Any         =			UNetaDConditionModel(**lowercase						)
        return model





     @property
     def 							lowercase__				(			self		: int						):





        """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 							lowercase__				(			self		: Optional[int]						):





        """simple docstring"""




        torch.manual_seed(0						)
        lowercase_      :Optional[Any]         =			VQModel(**self.dummy_movq_kwargs						)
        return model





     def 							lowercase__				(			self		: int						):





        """simple docstring"""




        lowercase_      :List[Any]         =			self.dummy_unet
        lowercase_      :Tuple         =			self.dummy_movq

        lowercase_      :List[Any]         =			{
            "num_train_timesteps": 1_000,
            "beta_schedule": "linear",
            "beta_start": 0.0_00_85,
            "beta_end": 0.0_12,
            "clip_sample": False,
            "set_alpha_to_one": False,
            "steps_offset": 0,
            "prediction_type": "epsilon",
            "thresholding": False,
        }

        lowercase_      :Union[str, Any]         =			DDIMScheduler(**lowercase						)

        lowercase_      :Tuple         =			{
            "unet": unet,
            "scheduler": scheduler,
            "movq": movq,
        }

        return components





     def 							lowercase__				(			self		: Tuple     ,				lowercase		: Any     ,				lowercase		: Tuple=0						):





        """simple docstring"""




        lowercase_      :str         =			floats_tensor((1, self.text_embedder_hidden_size)     ,				rng=random.Random(lowercase						)						).to(lowercase						)
        lowercase_      :Tuple         =			floats_tensor((1, self.text_embedder_hidden_size)     ,				rng=random.Random(seed + 1						)						).to(
            lowercase						)
        # create init_image
        lowercase_      :Any         =			floats_tensor((1, 3, 64, 64)     ,				rng=random.Random(lowercase						)						).to(lowercase						)
        lowercase_      :Any         =			image.cpu().permute(0     ,				2     ,				3     ,				1						)[0]
        lowercase_      :List[Any]         =			Image.fromarray(np.uinta(lowercase						)						).convert("RGB"						).resize((256, 256)						)

        if str(lowercase						).startswith("mps"						):
           lowercase_      :Union[str, Any]         =			torch.manual_seed(lowercase						)
        else:
           lowercase_      :Union[str, Any]         =			torch.Generator(device=lowercase						).manual_seed(lowercase						)
        lowercase_      :Optional[int]         =			{
            "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 							lowercase__				(			self		: Dict						):





        """simple docstring"""




        lowercase_      :List[str]         =			"cpu"

        lowercase_      :int         =			self.get_dummy_components()

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

        pipe.set_progress_bar_config(disable=lowercase						)

        lowercase_      :int         =			pipe(**self.get_dummy_inputs(lowercase						)						)
        lowercase_      :List[Any]         =			output.images

        lowercase_      :Union[str, Any]         =			pipe(
            **self.get_dummy_inputs(lowercase						)     ,				return_dict=lowercase     ,				)[0]

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

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

        lowercase_      :Union[str, Any]         =			np.array(
            [0.6_19_97_78, 0.63_98_44_06, 0.46_14_57_85, 0.62_94_49_84, 0.5_62_22_15, 0.47_30_61_32, 0.47_44_14_56, 0.4_60_76_06, 0.48_71_92_63]						)
        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 							lowercase__				(			self		: Optional[Any]						):





        """simple docstring"""




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







     def 							lowercase__				(			self		: Any						):





        """simple docstring"""




        lowercase_      :Tuple         =			load_numpy(
            "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
            "/kandinskyv22/kandinskyv22_img2img_frog.npy"						)

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

        lowercase_      :Optional[Any]         =			KandinskyVaaPriorPipeline.from_pretrained(
            "kandinsky-community/kandinsky-2-2-prior"     ,				torch_dtype=torch.floataa						)
        pipe_prior.to(lowercase						)

        lowercase_      :Dict         =			KandinskyVaaImgaImgPipeline.from_pretrained(
            "kandinsky-community/kandinsky-2-2-decoder"     ,				torch_dtype=torch.floataa						)
        lowercase_      :Optional[Any]         =			pipeline.to(lowercase						)

        pipeline.set_progress_bar_config(disable=lowercase						)

        lowercase_      :List[str]         =			torch.Generator(device="cpu"						).manual_seed(0						)
        lowercase_      ,							lowercase_      :Dict         =			pipe_prior(
            lowercase     ,				generator=lowercase     ,				num_inference_steps=5     ,				negative_prompt=""     ,				).to_tuple()

        lowercase_      :List[Any]         =			pipeline(
            image=lowercase     ,				image_embeds=lowercase     ,				negative_image_embeds=lowercase     ,				generator=lowercase     ,				num_inference_steps=100     ,				height=768     ,				width=768     ,				strength=0.2     ,				output_type="np"     ,				)

        lowercase_      :Optional[Any]         =			output.images[0]

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

        assert_mean_pixel_difference(lowercase     ,				lowercase						)









'''simple docstring'''


import tempfile

import torch

from diffusers import (
    DEISMultistepScheduler,
    DPMSolverMultistepScheduler,
    DPMSolverSinglestepScheduler,
    UniPCMultistepScheduler,
)

from .test_schedulers import SchedulerCommonTest
class 				a_       (    _lowerCAmelCase							):
     __A								=      (UniPCMultistepScheduler,)
     __A								=      (("num_inference_steps", 25),)





     def 							lowercase__				(			self		: Tuple     ,				**lowercase		: Optional[Any]						):





        """simple docstring"""




        lowercase_      :Any         =			{
            "num_train_timesteps": 1_000,
            "beta_start": 0.00_01,
            "beta_end": 0.02,
            "beta_schedule": "linear",
            "solver_order": 2,
            "solver_type": "bh2",
        }

        config.update(**lowercase						)
        return config





     def 							lowercase__				(			self		: Optional[Any]     ,				lowercase		: str=0     ,				**lowercase		: Union[str, Any]						):





        """simple docstring"""




        lowercase_      :Tuple         =			dict(self.forward_default_kwargs						)
        lowercase_      :Tuple         =			kwargs.pop("num_inference_steps"     ,				lowercase						)
        lowercase_      :Union[str, Any]         =			self.dummy_sample
        lowercase_      :Union[str, Any]         =			0.1 * sample
        lowercase_      :Any         =			[residual + 0.2, residual + 0.15, residual + 0.10]

        for scheduler_class in self.scheduler_classes:
           lowercase_      :Tuple         =			self.get_scheduler_config(**lowercase						)
           lowercase_      :Tuple         =			scheduler_class(**lowercase						)
           scheduler.set_timesteps(lowercase						)
           # copy over dummy past residuals
           lowercase_      :str         =			dummy_past_residuals[: scheduler.config.solver_order]

           with tempfile.TemporaryDirectory() as tmpdirname:
              scheduler.save_config(lowercase						)
              lowercase_      :str         =			scheduler_class.from_pretrained(lowercase						)
              new_scheduler.set_timesteps(lowercase						)
              # copy over dummy past residuals
              lowercase_      :List[Any]         =			dummy_past_residuals[: new_scheduler.config.solver_order]

           lowercase_      ,							lowercase_      :Tuple         =			sample, sample
           for t in range(lowercase     ,				time_step + scheduler.config.solver_order + 1						):
              lowercase_      :Optional[int]         =			scheduler.step(lowercase     ,				lowercase     ,				lowercase     ,				**lowercase						).prev_sample
              lowercase_      :Union[str, Any]         =			new_scheduler.step(lowercase     ,				lowercase     ,				lowercase     ,				**lowercase						).prev_sample

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





     def 							lowercase__				(			self		: str     ,				lowercase		: Any=0     ,				**lowercase		: str						):





        """simple docstring"""




        lowercase_      :Tuple         =			dict(self.forward_default_kwargs						)
        lowercase_      :Dict         =			kwargs.pop("num_inference_steps"     ,				lowercase						)
        lowercase_      :str         =			self.dummy_sample
        lowercase_      :Optional[int]         =			0.1 * sample
        lowercase_      :Optional[int]         =			[residual + 0.2, residual + 0.15, residual + 0.10]

        for scheduler_class in self.scheduler_classes:
           lowercase_      :Any         =			self.get_scheduler_config()
           lowercase_      :Tuple         =			scheduler_class(**lowercase						)
           scheduler.set_timesteps(lowercase						)

           # copy over dummy past residuals (must be after setting timesteps)
           lowercase_      :str         =			dummy_past_residuals[: scheduler.config.solver_order]

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

              # copy over dummy past residual (must be after setting timesteps)
              lowercase_      :Optional[Any]         =			dummy_past_residuals[: new_scheduler.config.solver_order]

           lowercase_      :Dict         =			scheduler.step(lowercase     ,				lowercase     ,				lowercase     ,				**lowercase						).prev_sample
           lowercase_      :str         =			new_scheduler.step(lowercase     ,				lowercase     ,				lowercase     ,				**lowercase						).prev_sample

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





     def 							lowercase__				(			self		: List[str]     ,				lowercase		: Optional[int]=None     ,				**lowercase		: int						):





        """simple docstring"""




        if scheduler is None:
           lowercase_      :int         =			self.scheduler_classes[0]
           lowercase_      :Optional[Any]         =			self.get_scheduler_config(**lowercase						)
           lowercase_      :Optional[Any]         =			scheduler_class(**lowercase						)

        lowercase_      :str         =			self.scheduler_classes[0]
        lowercase_      :Optional[int]         =			self.get_scheduler_config(**lowercase						)
        lowercase_      :List[Any]         =			scheduler_class(**lowercase						)

        lowercase_      :str         =			10
        lowercase_      :List[str]         =			self.dummy_model()
        lowercase_      :Optional[int]         =			self.dummy_sample_deter
        scheduler.set_timesteps(lowercase						)

        for i, t in enumerate(scheduler.timesteps						):
           lowercase_      :int         =			model(lowercase     ,				lowercase						)
           lowercase_      :Dict         =			scheduler.step(lowercase     ,				lowercase     ,				lowercase						).prev_sample

        return sample





     def 							lowercase__				(			self		: int						):





        """simple docstring"""




        lowercase_      :str         =			dict(self.forward_default_kwargs						)

        lowercase_      :Union[str, Any]         =			kwargs.pop("num_inference_steps"     ,				lowercase						)

        for scheduler_class in self.scheduler_classes:
           lowercase_      :Optional[Any]         =			self.get_scheduler_config()
           lowercase_      :Optional[int]         =			scheduler_class(**lowercase						)

           lowercase_      :Union[str, Any]         =			self.dummy_sample
           lowercase_      :str         =			0.1 * sample

           if num_inference_steps is not None and hasattr(lowercase     ,				"set_timesteps"						):
              scheduler.set_timesteps(lowercase						)
           elif num_inference_steps is not None and not hasattr(lowercase     ,				"set_timesteps"						):
              lowercase_      :str         =			num_inference_steps

           # copy over dummy past residuals (must be done after set_timesteps)
           lowercase_      :Dict         =			[residual + 0.2, residual + 0.15, residual + 0.10]
           lowercase_      :Any         =			dummy_past_residuals[: scheduler.config.solver_order]

           lowercase_      :Optional[int]         =			scheduler.timesteps[5]
           lowercase_      :List[Any]         =			scheduler.timesteps[6]

           lowercase_      :Optional[Any]         =			scheduler.step(lowercase     ,				lowercase     ,				lowercase     ,				**lowercase						).prev_sample
           lowercase_      :List[str]         =			scheduler.step(lowercase     ,				lowercase     ,				lowercase     ,				**lowercase						).prev_sample

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





     def 							lowercase__				(			self		: List[str]						):





        """simple docstring"""




        lowercase_      :List[Any]         =			UniPCMultistepScheduler(**self.get_scheduler_config()						)
        lowercase_      :Dict         =			self.full_loop(scheduler=lowercase						)
        lowercase_      :List[Any]         =			torch.mean(torch.abs(lowercase						)						)

        assert abs(result_mean.item() - 0.24_64						) < 1e-3

        lowercase_      :List[str]         =			DPMSolverSinglestepScheduler.from_config(scheduler.config						)
        lowercase_      :Tuple         =			DEISMultistepScheduler.from_config(scheduler.config						)
        lowercase_      :Optional[int]         =			DPMSolverMultistepScheduler.from_config(scheduler.config						)
        lowercase_      :Tuple         =			UniPCMultistepScheduler.from_config(scheduler.config						)

        lowercase_      :Tuple         =			self.full_loop(scheduler=lowercase						)
        lowercase_      :Tuple         =			torch.mean(torch.abs(lowercase						)						)

        assert abs(result_mean.item() - 0.24_64						) < 1e-3





     def 							lowercase__				(			self		: Tuple						):





        """simple docstring"""




        for timesteps in [25, 50, 100, 999, 1_000]:
           self.check_over_configs(num_train_timesteps=lowercase						)





     def 							lowercase__				(			self		: int						):





        """simple docstring"""




        self.check_over_configs(thresholding=lowercase						)
        for order in [1, 2, 3]:
           for solver_type in ["bh1", "bh2"]:
              for threshold in [0.5, 1.0, 2.0]:
                 for prediction_type in ["epsilon", "sample"]:
                    self.check_over_configs(
                        thresholding=lowercase     ,				prediction_type=lowercase     ,				sample_max_value=lowercase     ,				solver_order=lowercase     ,				solver_type=lowercase     ,				)





     def 							lowercase__				(			self		: List[str]						):





        """simple docstring"""




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





     def 							lowercase__				(			self		: Optional[int]						):





        """simple docstring"""




        for solver_type in ["bh1", "bh2"]:
           for order in [1, 2, 3]:
              for prediction_type in ["epsilon", "sample"]:
                 self.check_over_configs(
                     solver_order=lowercase     ,				solver_type=lowercase     ,				prediction_type=lowercase     ,				)
                 lowercase_      :Tuple         =			self.full_loop(
                     solver_order=lowercase     ,				solver_type=lowercase     ,				prediction_type=lowercase     ,				)
                 assert not torch.isnan(lowercase						).any(), "Samples have nan numbers"





     def 							lowercase__				(			self		: List[Any]						):





        """simple docstring"""




        self.check_over_configs(lower_order_final=lowercase						)
        self.check_over_configs(lower_order_final=lowercase						)





     def 							lowercase__				(			self		: Tuple						):





        """simple docstring"""




        for num_inference_steps in [1, 2, 3, 5, 10, 50, 100, 999, 1_000]:
           self.check_over_forward(num_inference_steps=lowercase     ,				time_step=0						)





     def 							lowercase__				(			self		: Dict						):





        """simple docstring"""




        lowercase_      :List[str]         =			self.full_loop()
        lowercase_      :List[str]         =			torch.mean(torch.abs(lowercase						)						)

        assert abs(result_mean.item() - 0.24_64						) < 1e-3





     def 							lowercase__				(			self		: Any						):





        """simple docstring"""




        lowercase_      :Dict         =			self.full_loop(prediction_type="v_prediction"						)
        lowercase_      :Any         =			torch.mean(torch.abs(lowercase						)						)

        assert abs(result_mean.item() - 0.10_14						) < 1e-3





     def 							lowercase__				(			self		: Optional[Any]						):





        """simple docstring"""




        lowercase_      :Union[str, Any]         =			self.scheduler_classes[0]
        lowercase_      :str         =			self.get_scheduler_config(thresholding=lowercase     ,				dynamic_thresholding_ratio=0						)
        lowercase_      :Tuple         =			scheduler_class(**lowercase						)

        lowercase_      :Optional[int]         =			10
        lowercase_      :Optional[int]         =			self.dummy_model()
        lowercase_      :Any         =			self.dummy_sample_deter.half()
        scheduler.set_timesteps(lowercase						)

        for i, t in enumerate(scheduler.timesteps						):
           lowercase_      :Tuple         =			model(lowercase     ,				lowercase						)
           lowercase_      :Any         =			scheduler.step(lowercase     ,				lowercase     ,				lowercase						).prev_sample

        assert sample.dtype == torch.floataa







     def 							lowercase__				(			self		: List[Any]     ,				**lowercase		: Union[str, Any]						):





        """simple docstring"""




        for scheduler_class in self.scheduler_classes:
           lowercase_      :Optional[int]         =			self.get_scheduler_config(**lowercase						)
           lowercase_      :Dict         =			scheduler_class(**lowercase						)

           scheduler.set_timesteps(scheduler.config.num_train_timesteps						)
           assert len(scheduler.timesteps.unique()						) == scheduler.num_inference_steps


