


"""simple docstring"""







import tempfile

import torch

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

from .test_schedulers import SchedulerCommonTest


class 				SCREAMING_SNAKE_CASE__    (		lowercase    ):
       """simple docstring"""


       a  :    Optional[int]       =(DPMSolverSinglestepScheduler,)
       a  :    Optional[Any]       =(("num_inference_steps", 25),)

       def 	lowercase__							(						self		,					**snake_case__						):






         """simple docstring"""





         lowerCAmelCase					:					Union[str, Any]								=							{
             "num_train_timesteps": 1_000,
             "beta_start": 0.0001,
             "beta_end": 0.02,
             "beta_schedule": "linear",
             "solver_order": 2,
             "prediction_type": "epsilon",
             "thresholding": False,
             "sample_max_value": 1.0,
             "algorithm_type": "dpmsolver++",
             "solver_type": "midpoint",
             "lambda_min_clipped": -float("inf"						),
             "variance_type": None,
         }

         config.update(**snake_case__						)
         return config

       def 	lowercase__							(						self		,					snake_case__=0		,					**snake_case__						):






         """simple docstring"""





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

         for scheduler_class in self.scheduler_classes:
           lowerCAmelCase					:					Union[str, Any]								=							self.get_scheduler_config(**snake_case__						)
           lowerCAmelCase					:					Any								=							scheduler_class(**snake_case__						)
           scheduler.set_timesteps(snake_case__						)
           # copy over dummy past residuals
           lowerCAmelCase					:					List[str]								=							dummy_past_residuals[: scheduler.config.solver_order]

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

           lowerCAmelCase						,  lowerCAmelCase					:					Tuple								=							sample, sample
           for t in range(snake_case__		,					time_step + scheduler.config.solver_order + 1						):
             lowerCAmelCase					:					List[str]								=							scheduler.step(snake_case__		,					snake_case__		,					snake_case__		,					**snake_case__						).prev_sample
             lowerCAmelCase					:					Optional[int]								=							new_scheduler.step(snake_case__		,					snake_case__		,					snake_case__		,					**snake_case__						).prev_sample

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

       def 	lowercase__							(						self						):






         """simple docstring"""





         pass

       def 	lowercase__							(						self		,					snake_case__=0		,					**snake_case__						):






         """simple docstring"""





         lowerCAmelCase					:					Any								=							dict(self.forward_default_kwargs						)
         lowerCAmelCase					:					int								=							kwargs.pop("num_inference_steps"		,					snake_case__						)
         lowerCAmelCase					:					Any								=							self.dummy_sample
         lowerCAmelCase					:					Optional[int]								=							0.1 * sample
         lowerCAmelCase					:					Tuple								=							[residual + 0.2, residual + 0.15, residual + 0.10]

         for scheduler_class in self.scheduler_classes:
           lowerCAmelCase					:					Optional[Any]								=							self.get_scheduler_config()
           lowerCAmelCase					:					Optional[int]								=							scheduler_class(**snake_case__						)
           scheduler.set_timesteps(snake_case__						)

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

           with tempfile.TemporaryDirectory() as tmpdirname:
             scheduler.save_config(snake_case__						)
             lowerCAmelCase					:					Optional[Any]								=							scheduler_class.from_pretrained(snake_case__						)
             # copy over dummy past residuals
             new_scheduler.set_timesteps(snake_case__						)

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

           lowerCAmelCase					:					Dict								=							scheduler.step(snake_case__		,					snake_case__		,					snake_case__		,					**snake_case__						).prev_sample
           lowerCAmelCase					:					Tuple								=							new_scheduler.step(snake_case__		,					snake_case__		,					snake_case__		,					**snake_case__						).prev_sample

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

       def 	lowercase__							(						self		,					snake_case__=None		,					**snake_case__						):






         """simple docstring"""





         if scheduler is None:
           lowerCAmelCase					:					Tuple								=							self.scheduler_classes[0]
           lowerCAmelCase					:					Dict								=							self.get_scheduler_config(**snake_case__						)
           lowerCAmelCase					:					Any								=							scheduler_class(**snake_case__						)

         lowerCAmelCase					:					Union[str, Any]								=							self.scheduler_classes[0]
         lowerCAmelCase					:					str								=							self.get_scheduler_config(**snake_case__						)
         lowerCAmelCase					:					Union[str, Any]								=							scheduler_class(**snake_case__						)

         lowerCAmelCase					:					List[Any]								=							10
         lowerCAmelCase					:					Any								=							self.dummy_model()
         lowerCAmelCase					:					Optional[Any]								=							self.dummy_sample_deter
         scheduler.set_timesteps(snake_case__						)

         for i, t in enumerate(scheduler.timesteps						):
           lowerCAmelCase					:					Tuple								=							model(snake_case__		,					snake_case__						)
           lowerCAmelCase					:					Optional[int]								=							scheduler.step(snake_case__		,					snake_case__		,					snake_case__						).prev_sample

         return sample

       def 	lowercase__							(						self						):






         """simple docstring"""





         lowerCAmelCase					:					List[Any]								=							DPMSolverSinglestepScheduler(**self.get_scheduler_config()						)
         lowerCAmelCase					:					str								=							50
         lowerCAmelCase					:					List[str]								=							self.dummy_model()
         lowerCAmelCase					:					List[str]								=							self.dummy_sample_deter
         scheduler.set_timesteps(snake_case__						)

         # make sure that the first t is uneven
         for i, t in enumerate(scheduler.timesteps[3:]						):
           lowerCAmelCase					:					Union[str, Any]								=							model(snake_case__		,					snake_case__						)
           lowerCAmelCase					:					Tuple								=							scheduler.step(snake_case__		,					snake_case__		,					snake_case__						).prev_sample

         lowerCAmelCase					:					List[str]								=							torch.mean(torch.abs(snake_case__						)						)

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

       def 	lowercase__							(						self						):






         """simple docstring"""





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

       def 	lowercase__							(						self						):






         """simple docstring"""





         lowerCAmelCase					:					Optional[int]								=							DPMSolverSinglestepScheduler(**self.get_scheduler_config()						)
         lowerCAmelCase					:					Union[str, Any]								=							self.full_loop(scheduler=snake_case__						)
         lowerCAmelCase					:					Tuple								=							torch.mean(torch.abs(snake_case__						)						)

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

         lowerCAmelCase					:					Union[str, Any]								=							DEISMultistepScheduler.from_config(scheduler.config						)
         lowerCAmelCase					:					str								=							DPMSolverMultistepScheduler.from_config(scheduler.config						)
         lowerCAmelCase					:					Union[str, Any]								=							UniPCMultistepScheduler.from_config(scheduler.config						)
         lowerCAmelCase					:					str								=							DPMSolverSinglestepScheduler.from_config(scheduler.config						)

         lowerCAmelCase					:					Optional[int]								=							self.full_loop(scheduler=snake_case__						)
         lowerCAmelCase					:					str								=							torch.mean(torch.abs(snake_case__						)						)

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

       def 	lowercase__							(						self						):






         """simple docstring"""





         self.check_over_configs(thresholding=snake_case__						)
         for order in [1, 2, 3]:
           for solver_type in ["midpoint", "heun"]:
             for threshold in [0.5, 1.0, 2.0]:
               for prediction_type in ["epsilon", "sample"]:
                 self.check_over_configs(
                     thresholding=snake_case__		,					prediction_type=snake_case__		,					sample_max_value=snake_case__		,					algorithm_type="dpmsolver++"		,					solver_order=snake_case__		,					solver_type=snake_case__		,					)

       def 	lowercase__							(						self						):






         """simple docstring"""





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

       def 	lowercase__							(						self						):






         """simple docstring"""





         for algorithm_type in ["dpmsolver", "dpmsolver++"]:
           for solver_type in ["midpoint", "heun"]:
             for order in [1, 2, 3]:
               for prediction_type in ["epsilon", "sample"]:
                 self.check_over_configs(
                     solver_order=snake_case__		,					solver_type=snake_case__		,					prediction_type=snake_case__		,					algorithm_type=snake_case__		,					)
                 lowerCAmelCase					:					Any								=							self.full_loop(
                     solver_order=snake_case__		,					solver_type=snake_case__		,					prediction_type=snake_case__		,					algorithm_type=snake_case__		,					)
                 assert not torch.isnan(snake_case__						).any(), "Samples have nan numbers"

       def 	lowercase__							(						self						):






         """simple docstring"""





         self.check_over_configs(lower_order_final=snake_case__						)
         self.check_over_configs(lower_order_final=snake_case__						)

       def 	lowercase__							(						self						):






         """simple docstring"""





         self.check_over_configs(lambda_min_clipped=-float("inf"						)						)
         self.check_over_configs(lambda_min_clipped=-5.1						)

       def 	lowercase__							(						self						):






         """simple docstring"""





         self.check_over_configs(variance_type=snake_case__						)
         self.check_over_configs(variance_type="learned_range"						)

       def 	lowercase__							(						self						):






         """simple docstring"""





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

       def 	lowercase__							(						self						):






         """simple docstring"""





         lowerCAmelCase					:					Tuple								=							self.full_loop()
         lowerCAmelCase					:					Union[str, Any]								=							torch.mean(torch.abs(snake_case__						)						)

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

       def 	lowercase__							(						self						):






         """simple docstring"""





         lowerCAmelCase					:					Union[str, Any]								=							self.full_loop(use_karras_sigmas=snake_case__						)
         lowerCAmelCase					:					Any								=							torch.mean(torch.abs(snake_case__						)						)

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

       def 	lowercase__							(						self						):






         """simple docstring"""





         lowerCAmelCase					:					Optional[int]								=							self.full_loop(prediction_type="v_prediction"						)
         lowerCAmelCase					:					Tuple								=							torch.mean(torch.abs(snake_case__						)						)

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

       def 	lowercase__							(						self						):






         """simple docstring"""





         lowerCAmelCase					:					Optional[int]								=							self.full_loop(prediction_type="v_prediction"		,					use_karras_sigmas=snake_case__						)
         lowerCAmelCase					:					List[str]								=							torch.mean(torch.abs(snake_case__						)						)

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







       def 	lowercase__							(						self						):






         """simple docstring"""





         lowerCAmelCase					:					str								=							self.scheduler_classes[0]
         lowerCAmelCase					:					Optional[int]								=							self.get_scheduler_config(thresholding=snake_case__		,					dynamic_thresholding_ratio=0						)
         lowerCAmelCase					:					Dict								=							scheduler_class(**snake_case__						)

         lowerCAmelCase					:					Dict								=							10
         lowerCAmelCase					:					Tuple								=							self.dummy_model()
         lowerCAmelCase					:					List[str]								=							self.dummy_sample_deter.half()
         scheduler.set_timesteps(snake_case__						)

         for i, t in enumerate(scheduler.timesteps						):
           lowerCAmelCase					:					List[str]								=							model(snake_case__		,					snake_case__						)
           lowerCAmelCase					:					Dict								=							scheduler.step(snake_case__		,					snake_case__		,					snake_case__						).prev_sample

         assert sample.dtype == torch.floataa









"""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 				SCREAMING_SNAKE_CASE__    (		lowercase       ,   unittest.TestCase    ):
       """simple docstring"""


       a  :    Optional[Any]       =KandinskyVaaImgaImgPipeline
       a  :    Optional[int]       =["image_embeds", "negative_image_embeds", "image"]
       a  :    Optional[int]       =[
           "image_embeds",
           "negative_image_embeds",
           "image",
       ]
       a  :    str       =[
           "generator",
           "height",
           "width",
           "strength",
           "guidance_scale",
           "num_inference_steps",
           "return_dict",
           "guidance_scale",
           "num_images_per_prompt",
           "output_type",
           "return_dict",
       ]
       a  :    Dict       =False

       @property
       def 	lowercase__							(						self						):






         """simple docstring"""





         return 32

       @property
       def 	lowercase__							(						self						):






         """simple docstring"""





         return 32

       @property
       def 	lowercase__							(						self						):






         """simple docstring"""





         return self.time_input_dim

       @property
       def 	lowercase__							(						self						):






         """simple docstring"""





         return self.time_input_dim * 4

       @property
       def 	lowercase__							(						self						):






         """simple docstring"""





         return 100

       @property
       def 	lowercase__							(						self						):






         """simple docstring"""





         torch.manual_seed(0						)

         lowerCAmelCase					:					List[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,
         }

         lowerCAmelCase					:					int								=							UNetaDConditionModel(**snake_case__						)
         return model

       @property
       def 	lowercase__							(						self						):






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






         """simple docstring"""





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

       def 	lowercase__							(						self						):






         """simple docstring"""





         lowerCAmelCase					:					Optional[Any]								=							self.dummy_unet
         lowerCAmelCase					:					Optional[int]								=							self.dummy_movq

         lowerCAmelCase					:					List[str]								=							{
             "num_train_timesteps": 1_000,
             "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,
         }

         lowerCAmelCase					:					Tuple								=							DDIMScheduler(**snake_case__						)

         lowerCAmelCase					:					Optional[Any]								=							{
             "unet": unet,
             "scheduler": scheduler,
             "movq": movq,
         }

         return components

       def 	lowercase__							(						self		,					snake_case__		,					snake_case__=0						):






         """simple docstring"""





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

         if str(snake_case__						).startswith("mps"						):
           lowerCAmelCase					:					Optional[int]								=							torch.manual_seed(snake_case__						)
         else:
           lowerCAmelCase					:					Optional[int]								=							torch.Generator(device=snake_case__						).manual_seed(snake_case__						)
         lowerCAmelCase					:					List[str]								=							{
             "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						):






         """simple docstring"""





         lowerCAmelCase					:					Tuple								=							"cpu"

         lowerCAmelCase					:					Dict								=							self.get_dummy_components()

         lowerCAmelCase					:					Union[str, Any]								=							self.pipeline_class(**snake_case__						)
         lowerCAmelCase					:					Optional[int]								=							pipe.to(snake_case__						)

         pipe.set_progress_bar_config(disable=snake_case__						)

         lowerCAmelCase					:					int								=							pipe(**self.get_dummy_inputs(snake_case__						)						)
         lowerCAmelCase					:					Union[str, Any]								=							output.images

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

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

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

         lowerCAmelCase					:					int								=							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 				SCREAMING_SNAKE_CASE__    (		unittest.TestCase    ):
       """simple docstring"""


       def 	lowercase__							(						self						):






         """simple docstring"""





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







       def 	lowercase__							(						self						):






         """simple docstring"""





         lowerCAmelCase					:					List[Any]								=							load_numpy(
             "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
             "/kandinskyv22/kandinskyv22_img2img_frog.npy"						)

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

         lowerCAmelCase					:					int								=							KandinskyVaaPriorPipeline.from_pretrained(
             "kandinsky-community/kandinsky-2-2-prior"		,					torch_dtype=torch.floataa						)
         pipe_prior.to(snake_case__						)

         lowerCAmelCase					:					List[Any]								=							KandinskyVaaImgaImgPipeline.from_pretrained(
             "kandinsky-community/kandinsky-2-2-decoder"		,					torch_dtype=torch.floataa						)
         lowerCAmelCase					:					Tuple								=							pipeline.to(snake_case__						)

         pipeline.set_progress_bar_config(disable=snake_case__						)

         lowerCAmelCase					:					str								=							torch.Generator(device="cpu"						).manual_seed(0						)
         lowerCAmelCase						,  lowerCAmelCase					:					Optional[Any]								=							pipe_prior(
             snake_case__		,					generator=snake_case__		,					num_inference_steps=5		,					negative_prompt=""		,					).to_tuple()

         lowerCAmelCase					:					Tuple								=							pipeline(
             image=snake_case__		,					image_embeds=snake_case__		,					negative_image_embeds=snake_case__		,					generator=snake_case__		,					num_inference_steps=100		,					height=768		,					width=768		,					strength=0.2		,					output_type="np"		,					)

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

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

         assert_mean_pixel_difference(snake_case__		,					snake_case__						)





