
"""simple docstring"""




import gc
import tempfile
import unittest

import numpy as np
import torch

from diffusers import VersatileDiffusionPipeline
from diffusers.utils.testing_utils import load_image, nightly, require_torch_gpu, torch_device


A__   :					Any							=   False


class 			__magic_name__       (       unittest.TestCase ):
       pass


@nightly
@require_torch_gpu
class 			__magic_name__       (       unittest.TestCase ):
       def 							lowercase_							(     self       )     ->	Union[str, Any]:





         """simple docstring"""




         super().tearDown()
         gc.collect()
         torch.cuda.empty_cache()
       def 							lowercase_							(     self       )     ->	Tuple:





         """simple docstring"""




         _lowercase:			List[str]       =     VersatileDiffusionPipeline.from_pretrained('''shi-labs/versatile-diffusion''' ,				torch_dtype=torch.floataa       )
         pipe.to(A_       )
         pipe.set_progress_bar_config(disable=A_       )

         _lowercase:			Any       =     load_image(
             '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/versatile_diffusion/benz.jpg'''       )

         _lowercase:			str       =     torch.manual_seed(0       )
         _lowercase:			Any       =     pipe.dual_guided(
             prompt='''first prompt''' ,				image=A_ ,				text_to_image_strength=0.75 ,				generator=A_ ,				guidance_scale=7.5 ,				num_inference_steps=2 ,				output_type='''numpy''' ,				).images

         with tempfile.TemporaryDirectory() as tmpdirname:
           pipe.save_pretrained(A_       )
           _lowercase:			Union[str, Any]       =     VersatileDiffusionPipeline.from_pretrained(A_ ,				torch_dtype=torch.floataa       )
         pipe.to(A_       )
         pipe.set_progress_bar_config(disable=A_       )

         _lowercase:			Optional[int]       =     generator.manual_seed(0       )
         _lowercase:			Optional[int]       =     pipe.dual_guided(
             prompt='''first prompt''' ,				image=A_ ,				text_to_image_strength=0.75 ,				generator=A_ ,				guidance_scale=7.5 ,				num_inference_steps=2 ,				output_type='''numpy''' ,				).images

         assert np.abs(image - new_image       ).sum() < 1E-5, "Models don't have the same forward pass"



       def 							lowercase_							(     self       )     ->	List[Any]:





         """simple docstring"""




         _lowercase:			int       =     VersatileDiffusionPipeline.from_pretrained('''shi-labs/versatile-diffusion''' ,				torch_dtype=torch.floataa       )
         pipe.to(A_       )
         pipe.set_progress_bar_config(disable=A_       )

         _lowercase:			Tuple       =     '''cyberpunk 2077'''
         _lowercase:			Dict       =     load_image(
             '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/versatile_diffusion/benz.jpg'''       )
         _lowercase:			Union[str, Any]       =     torch.manual_seed(0       )
         _lowercase:			Union[str, Any]       =     pipe.dual_guided(
             prompt=A_ ,				image=A_ ,				text_to_image_strength=0.75 ,				generator=A_ ,				guidance_scale=7.5 ,				num_inference_steps=50 ,				output_type='''numpy''' ,				).images

         _lowercase:			List[Any]       =     image[0, 253:256, 253:256, -1]

         assert image.shape == (1, 512, 512, 3)
         _lowercase:			Dict       =     np.array([0.14_48, 0.16_19, 0.17_41, 0.10_86, 0.11_47, 0.11_28, 0.11_99, 0.11_65, 0.10_01]       )

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

         _lowercase:			List[str]       =     '''A painting of a squirrel eating a burger '''
         _lowercase:			Union[str, Any]       =     torch.manual_seed(0       )
         _lowercase:			Optional[int]       =     pipe.text_to_image(
             prompt=A_ ,				generator=A_ ,				guidance_scale=7.5 ,				num_inference_steps=50 ,				output_type='''numpy'''       ).images

         _lowercase:			int       =     image[0, 253:256, 253:256, -1]

         assert image.shape == (1, 512, 512, 3)
         _lowercase:			Dict       =     np.array([0.33_67, 0.31_69, 0.26_56, 0.38_70, 0.47_90, 0.37_96, 0.40_09, 0.48_78, 0.47_78]       )

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

         _lowercase:			Optional[int]       =     pipe.image_variation(A_ ,				generator=A_ ,				output_type='''numpy'''       ).images

         _lowercase:			List[Any]       =     image[0, 253:256, 253:256, -1]

         assert image.shape == (1, 512, 512, 3)
         _lowercase:			Optional[int]       =     np.array([0.30_76, 0.31_23, 0.32_84, 0.37_82, 0.37_70, 0.38_94, 0.42_97, 0.43_31, 0.44_56]       )

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


"""simple docstring"""




import torch

from diffusers import KDPMaDiscreteScheduler
from diffusers.utils import torch_device

from .test_schedulers import SchedulerCommonTest


class 			__magic_name__       (       SCREAMING_SNAKE_CASE__ ):
       UpperCamelCase_   =		(KDPMaDiscreteScheduler,)
       UpperCamelCase_   =		10
       def 							lowercase_							(     self ,				**A_       )     ->	List[Any]:





         """simple docstring"""




         _lowercase:			Tuple       =     {
             '''num_train_timesteps''': 1100,
             '''beta_start''': 0.00_01,
             '''beta_end''': 0.02,
             '''beta_schedule''': '''linear''',
         }

         config.update(**A_       )
         return config
       def 							lowercase_							(     self       )     ->	List[str]:





         """simple docstring"""




         for timesteps in [10, 50, 100, 1000]:
           self.check_over_configs(num_train_timesteps=A_       )
       def 							lowercase_							(     self       )     ->	Tuple:





         """simple docstring"""




         for beta_start, beta_end in zip([0.0_00_01, 0.00_01, 0.0_01] ,				[0.00_02, 0.0_02, 0.02]       ):
           self.check_over_configs(beta_start=A_ ,				beta_end=A_       )
       def 							lowercase_							(     self       )     ->	int:





         """simple docstring"""




         for schedule in ["linear", "scaled_linear"]:
           self.check_over_configs(beta_schedule=A_       )
       def 							lowercase_							(     self       )     ->	str:





         """simple docstring"""




         for prediction_type in ["epsilon", "v_prediction"]:
           self.check_over_configs(prediction_type=A_       )
       def 							lowercase_							(     self       )     ->	Tuple:





         """simple docstring"""




         _lowercase:			Optional[int]       =     self.scheduler_classes[0]
         _lowercase:			List[str]       =     self.get_scheduler_config(prediction_type='''v_prediction'''       )
         _lowercase:			List[Any]       =     scheduler_class(**A_       )

         scheduler.set_timesteps(self.num_inference_steps       )

         _lowercase:			List[str]       =     self.dummy_model()
         _lowercase:			Dict       =     self.dummy_sample_deter * scheduler.init_noise_sigma
         _lowercase:			List[Any]       =     sample.to(A_       )

         for i, t in enumerate(scheduler.timesteps       ):
           _lowercase:			Optional[int]       =     scheduler.scale_model_input(A_ ,				A_       )

           _lowercase:			Optional[Any]       =     model(A_ ,				A_       )

           _lowercase:			List[str]       =     scheduler.step(A_ ,				A_ ,				A_       )
           _lowercase:			str       =     output.prev_sample

         _lowercase:			Dict       =     torch.sum(torch.abs(A_       )       )
         _lowercase:			Optional[int]       =     torch.mean(torch.abs(A_       )       )

         if torch_device in ["cpu", "mps"]:
           assert abs(result_sum.item() - 4.69_34E-07       ) < 1E-2
           assert abs(result_mean.item() - 6.11_12E-10       ) < 1E-3
         else:
           # CUDA
           assert abs(result_sum.item() - 4.6_93_42_86_50_17_09_72E-07       ) < 1E-2
           assert abs(result_mean.item() - 0.00_02       ) < 1E-3
       def 							lowercase_							(     self       )     ->	str:





         """simple docstring"""




         if torch_device == "mps":
           return
         _lowercase:			List[str]       =     self.scheduler_classes[0]
         _lowercase:			Optional[int]       =     self.get_scheduler_config()
         _lowercase:			int       =     scheduler_class(**A_       )

         scheduler.set_timesteps(self.num_inference_steps       )

         _lowercase:			Tuple       =     self.dummy_model()
         _lowercase:			Tuple       =     self.dummy_sample_deter * scheduler.init_noise_sigma
         _lowercase:			Any       =     sample.to(A_       )

         for i, t in enumerate(scheduler.timesteps       ):
           _lowercase:			Any       =     scheduler.scale_model_input(A_ ,				A_       )

           _lowercase:			List[Any]       =     model(A_ ,				A_       )

           _lowercase:			Any       =     scheduler.step(A_ ,				A_ ,				A_       )
           _lowercase:			List[Any]       =     output.prev_sample

         _lowercase:			Optional[Any]       =     torch.sum(torch.abs(A_       )       )
         _lowercase:			Tuple       =     torch.mean(torch.abs(A_       )       )

         if torch_device in ["cpu", "mps"]:
           assert abs(result_sum.item() - 20.41_25       ) < 1E-2
           assert abs(result_mean.item() - 0.02_66       ) < 1E-3
         else:
           # CUDA
           assert abs(result_sum.item() - 20.41_25       ) < 1E-2
           assert abs(result_mean.item() - 0.02_66       ) < 1E-3



       def 							lowercase_							(     self       )     ->	Any:





         """simple docstring"""




         if torch_device == "mps":
           return
         _lowercase:			int       =     self.scheduler_classes[0]
         _lowercase:			Union[str, Any]       =     self.get_scheduler_config()
         _lowercase:			Optional[Any]       =     scheduler_class(**A_       )

         scheduler.set_timesteps(self.num_inference_steps ,				device=A_       )

         _lowercase:			List[str]       =     self.dummy_model()
         _lowercase:			Union[str, Any]       =     self.dummy_sample_deter.to(A_       ) * scheduler.init_noise_sigma

         for t in scheduler.timesteps:
           _lowercase:			Optional[Any]       =     scheduler.scale_model_input(A_ ,				A_       )

           _lowercase:			int       =     model(A_ ,				A_       )

           _lowercase:			Tuple       =     scheduler.step(A_ ,				A_ ,				A_       )
           _lowercase:			int       =     output.prev_sample

         _lowercase:			Union[str, Any]       =     torch.sum(torch.abs(A_       )       )
         _lowercase:			Dict       =     torch.mean(torch.abs(A_       )       )

         if str(A_       ).startswith('''cpu'''       ):
           # The following sum varies between 148 and 156 on mps. Why?
           assert abs(result_sum.item() - 20.41_25       ) < 1E-2
           assert abs(result_mean.item() - 0.02_66       ) < 1E-3
         else:
           # CUDA
           assert abs(result_sum.item() - 20.41_25       ) < 1E-2
           assert abs(result_mean.item() - 0.02_66       ) < 1E-3
