


import torch

from diffusers import UnCLIPScheduler

from .test_schedulers import SchedulerCommonTest







class      lowerCAmelCase__       (				a):




     '''simple docstring'''
     __SCREAMING_SNAKE_CASE  			= (UnCLIPScheduler,)


     def  _lowerCamelCase (						self ,						**__lowerCamelCase)			->				Any:
            _A      :       Dict				    = {
                "num_train_timesteps": 1_0_0_0,
                "variance_type": "fixed_small_log",
                "clip_sample": True,
                "clip_sample_range": 1.0,
                "prediction_type": "epsilon",
            }

            config.update(**__lowerCamelCase)
            return config


     def  _lowerCamelCase (						self)			->				Dict:
            for timesteps in [1, 5, 1_0_0, 1_0_0_0]:
                   self.check_over_configs(num_train_timesteps=__lowerCamelCase)


     def  _lowerCamelCase (						self)			->				str:
            for variance in ["fixed_small_log", "learned_range"]:
                   self.check_over_configs(variance_type=__lowerCamelCase)


     def  _lowerCamelCase (						self)			->				Any:
            for clip_sample in [True, False]:
                   self.check_over_configs(clip_sample=__lowerCamelCase)


     def  _lowerCamelCase (						self)			->				int:
            for clip_sample_range in [1, 5, 1_0, 2_0]:
                   self.check_over_configs(clip_sample_range=__lowerCamelCase)


     def  _lowerCamelCase (						self)			->				List[Any]:
            for prediction_type in ["epsilon", "sample"]:
                   self.check_over_configs(prediction_type=__lowerCamelCase)


     def  _lowerCamelCase (						self)			->				Dict:
            for time_step in [0, 5_0_0, 9_9_9]:
                   for prev_timestep in [None, 5, 1_0_0, 2_5_0, 5_0_0, 7_5_0]:
                          if prev_timestep is not None and prev_timestep >= time_step:
                                 continue

                          self.check_over_forward(time_step=__lowerCamelCase ,						prev_timestep=__lowerCamelCase)


     def  _lowerCamelCase (						self)			->				Optional[int]:
            _A      :       List[Any]				    = self.scheduler_classes[0]
            _A      :       Dict				    = self.get_scheduler_config(variance_type="fixed_small_log")
            _A      :       Optional[Any]				    = scheduler_class(**__lowerCamelCase)

            assert torch.sum(torch.abs(scheduler._get_variance(0) - 1.00_00e-10)) < 1e-5
            assert torch.sum(torch.abs(scheduler._get_variance(4_8_7) - 0.0_5_4_9_6_2_5)) < 1e-5
            assert torch.sum(torch.abs(scheduler._get_variance(9_9_9) - 0.9_9_9_4_9_8_7)) < 1e-5


     def  _lowerCamelCase (						self)			->				str:
            _A      :       List[Any]				    = self.scheduler_classes[0]
            _A      :       Union[str, Any]				    = self.get_scheduler_config(variance_type="learned_range")
            _A      :       List[str]				    = scheduler_class(**__lowerCamelCase)

            _A      :       Any				    = 0.5

            assert scheduler._get_variance(1 ,						predicted_variance=__lowerCamelCase) - -1_0.1_7_1_2_7_9_0 < 1e-5
            assert scheduler._get_variance(4_8_7 ,						predicted_variance=__lowerCamelCase) - -5.7_9_9_8_0_5_2 < 1e-5
            assert scheduler._get_variance(9_9_9 ,						predicted_variance=__lowerCamelCase) - -0.0_0_1_0_0_1_1 < 1e-5


     def  _lowerCamelCase (						self)			->				Optional[int]:
            _A      :       List[str]				    = self.scheduler_classes[0]
            _A      :       Dict				    = self.get_scheduler_config()
            _A      :       List[str]				    = scheduler_class(**__lowerCamelCase)

            _A      :       Optional[int]				    = scheduler.timesteps

            _A      :       Union[str, Any]				    = self.dummy_model()
            _A      :       List[Any]				    = self.dummy_sample_deter
            _A      :       Any				    = torch.manual_seed(0)

            for i, t in enumerate(__lowerCamelCase):
                   # 1. predict noise residual
                   _A      :       Optional[Any]				    = model(__lowerCamelCase ,						__lowerCamelCase)

                   # 2. predict previous mean of sample x_t-1
                   _A      :       Optional[int]				    = scheduler.step(__lowerCamelCase ,						__lowerCamelCase ,						__lowerCamelCase ,						generator=__lowerCamelCase).prev_sample

                   _A      :       List[Any]				    = pred_prev_sample

            _A      :       str				    = torch.sum(torch.abs(__lowerCamelCase))
            _A      :       Optional[Any]				    = torch.mean(torch.abs(__lowerCamelCase))

            assert abs(result_sum.item() - 2_5_2.2_6_8_2_4_9_5) < 1e-2
            assert abs(result_mean.item() - 0.3_2_8_4_7_4_3) < 1e-3


     def  _lowerCamelCase (						self)			->				Any:
            _A      :       Optional[Any]				    = self.scheduler_classes[0]
            _A      :       Union[str, Any]				    = self.get_scheduler_config()
            _A      :       Dict				    = scheduler_class(**__lowerCamelCase)

            scheduler.set_timesteps(2_5)

            _A      :       Dict				    = scheduler.timesteps

            _A      :       List[Any]				    = self.dummy_model()
            _A      :       Optional[int]				    = self.dummy_sample_deter
            _A      :       str				    = torch.manual_seed(0)

            for i, t in enumerate(__lowerCamelCase):
                   # 1. predict noise residual
                   _A      :       Optional[int]				    = model(__lowerCamelCase ,						__lowerCamelCase)

                   if i + 1 == timesteps.shape[0]:
                          _A      :       List[str]				    = None
                   else:
                          _A      :       int				    = timesteps[i + 1]

                   # 2. predict previous mean of sample x_t-1
                   _A      :       Any				    = scheduler.step(
                       __lowerCamelCase ,						__lowerCamelCase ,						__lowerCamelCase ,						prev_timestep=__lowerCamelCase ,						generator=__lowerCamelCase).prev_sample

                   _A      :       Dict				    = pred_prev_sample

            _A      :       Dict				    = torch.sum(torch.abs(__lowerCamelCase))
            _A      :       Any				    = torch.mean(torch.abs(__lowerCamelCase))

            assert abs(result_sum.item() - 2_5_8.2_0_4_4_9_8_3) < 1e-2
            assert abs(result_mean.item() - 0.3_3_6_2_0_3_8) < 1e-3


     def  _lowerCamelCase (						self)			->				str:
            pass


     def  _lowerCamelCase (						self)			->				int:
            pass




import gc
import unittest

import numpy as np
import torch
from torch.backends.cuda import sdp_kernel

from diffusers import (
    CMStochasticIterativeScheduler,
    ConsistencyModelPipeline,
    UNetaDModel,
)
from diffusers.utils import randn_tensor, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_a, require_torch_gpu

from ..pipeline_params import UNCONDITIONAL_IMAGE_GENERATION_BATCH_PARAMS, UNCONDITIONAL_IMAGE_GENERATION_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()






class _lowerCamelCase(						_a, unittest.TestCase   ):
       lowercase_ :	Union[str, Any]     =			ConsistencyModelPipeline
       lowercase_ :	Tuple     =			UNCONDITIONAL_IMAGE_GENERATION_PARAMS
       lowercase_ :	List[str]     =			UNCONDITIONAL_IMAGE_GENERATION_BATCH_PARAMS

       # Override required_optional_params to remove num_images_per_prompt
       lowercase_ :	List[str]     =			frozenset(
           [
               """num_inference_steps""",
               """generator""",
               """latents""",
               """output_type""",
               """return_dict""",
               """callback""",
               """callback_steps""",
           ]   )
       @property
       def 			UpperCamelCase     (					self)       ->   Tuple:




           """simple docstring"""

           _lowercase       :							Tuple      		=		UNetaDModel.from_pretrained(
               'diffusers/consistency-models-test',      subfolder='test_unet',      )
           return unet
       @property
       def 			UpperCamelCase     (					self)       ->   List[Any]:




           """simple docstring"""

           _lowercase       :							Tuple      		=		UNetaDModel.from_pretrained(
               'diffusers/consistency-models-test',      subfolder='test_unet_class_cond',      )
           return unet
       def 			UpperCamelCase     (					self,      lowerCamelCase=False)       ->   Dict:




           """simple docstring"""

           if class_cond:
               _lowercase       :							Union[str, Any]      		=		self.dummy_cond_unet
           else:
               _lowercase       :							Union[str, Any]      		=		self.dummy_uncond_unet

           # Default to CM multistep sampler
           _lowercase       :							List[str]      		=		CMStochasticIterativeScheduler(
               num_train_timesteps=40,      sigma_min=0.0_0_2,      sigma_max=8_0.0,      )

           _lowercase       :							Optional[Any]      		=		{
               'unet': unet,
               'scheduler': scheduler,
           }

           return components
       def 			UpperCamelCase     (					self,      lowerCamelCase,      lowerCamelCase=0)       ->   Tuple:




           """simple docstring"""

           if str(lowerCamelCase).startswith('mps'):
               _lowercase       :							str      		=		torch.manual_seed(lowerCamelCase)
           else:
               _lowercase       :							int      		=		torch.Generator(device=lowerCamelCase).manual_seed(lowerCamelCase)

           _lowercase       :							Tuple      		=		{
               'batch_size': 1,
               'num_inference_steps': None,
               'timesteps': [22, 0],
               'generator': generator,
               'output_type': 'np',
           }

           return inputs
       def 			UpperCamelCase     (					self)       ->   Any:




           """simple docstring"""

           _lowercase       :							Optional[int]      		=		'cpu'  # ensure determinism for the device-dependent torch.Generator
           _lowercase       :							Optional[int]      		=		self.get_dummy_components()
           _lowercase       :							str      		=		ConsistencyModelPipeline(**lowerCamelCase)
           _lowercase       :							Dict      		=		pipe.to(lowerCamelCase)
           pipe.set_progress_bar_config(disable=lowerCamelCase)

           _lowercase       :							Tuple      		=		self.get_dummy_inputs(lowerCamelCase)
           _lowercase       :							Optional[int]      		=		pipe(**lowerCamelCase).images
           assert image.shape == (1, 32, 32, 3)

           _lowercase       :							int      		=		image[0, -3:, -3:, -1]
           _lowercase       :							Dict      		=		np.array([0.3_5_7_2, 0.6_2_7_3, 0.4_0_3_1, 0.3_9_6_1, 0.4_3_2_1, 0.5_7_3_0, 0.5_2_6_6, 0.4_7_8_0, 0.5_0_0_4])

           assert np.abs(image_slice.flatten() - expected_slice).max() < 1E-3
       def 			UpperCamelCase     (					self)       ->   Any:




           """simple docstring"""

           _lowercase       :							Union[str, Any]      		=		'cpu'  # ensure determinism for the device-dependent torch.Generator
           _lowercase       :							Dict      		=		self.get_dummy_components(class_cond=lowerCamelCase)
           _lowercase       :							Any      		=		ConsistencyModelPipeline(**lowerCamelCase)
           _lowercase       :							str      		=		pipe.to(lowerCamelCase)
           pipe.set_progress_bar_config(disable=lowerCamelCase)

           _lowercase       :							Any      		=		self.get_dummy_inputs(lowerCamelCase)
           _lowercase       :							Any      		=		0
           _lowercase       :							List[str]      		=		pipe(**lowerCamelCase).images
           assert image.shape == (1, 32, 32, 3)

           _lowercase       :							Any      		=		image[0, -3:, -3:, -1]
           _lowercase       :							Union[str, Any]      		=		np.array([0.3_5_7_2, 0.6_2_7_3, 0.4_0_3_1, 0.3_9_6_1, 0.4_3_2_1, 0.5_7_3_0, 0.5_2_6_6, 0.4_7_8_0, 0.5_0_0_4])

           assert np.abs(image_slice.flatten() - expected_slice).max() < 1E-3
       def 			UpperCamelCase     (					self)       ->   Tuple:




           """simple docstring"""

           _lowercase       :							Optional[int]      		=		'cpu'  # ensure determinism for the device-dependent torch.Generator
           _lowercase       :							Any      		=		self.get_dummy_components()
           _lowercase       :							Optional[Any]      		=		ConsistencyModelPipeline(**lowerCamelCase)
           _lowercase       :							List[str]      		=		pipe.to(lowerCamelCase)
           pipe.set_progress_bar_config(disable=lowerCamelCase)

           _lowercase       :							Optional[Any]      		=		self.get_dummy_inputs(lowerCamelCase)
           _lowercase       :							Union[str, Any]      		=		1
           _lowercase       :							Tuple      		=		None
           _lowercase       :							Tuple      		=		pipe(**lowerCamelCase).images
           assert image.shape == (1, 32, 32, 3)

           _lowercase       :							str      		=		image[0, -3:, -3:, -1]
           _lowercase       :							List[str]      		=		np.array([0.5_0_0_4, 0.5_0_0_4, 0.4_9_9_4, 0.5_0_0_8, 0.4_9_7_6, 0.5_0_1_8, 0.4_9_9_0, 0.4_9_8_2, 0.4_9_8_7])

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




       def 			UpperCamelCase     (					self)       ->   str:




           """simple docstring"""

           _lowercase       :							Union[str, Any]      		=		'cpu'  # ensure determinism for the device-dependent torch.Generator
           _lowercase       :							Dict      		=		self.get_dummy_components(class_cond=lowerCamelCase)
           _lowercase       :							Dict      		=		ConsistencyModelPipeline(**lowerCamelCase)
           _lowercase       :							Optional[Any]      		=		pipe.to(lowerCamelCase)
           pipe.set_progress_bar_config(disable=lowerCamelCase)

           _lowercase       :							Tuple      		=		self.get_dummy_inputs(lowerCamelCase)
           _lowercase       :							Tuple      		=		1
           _lowercase       :							int      		=		None
           _lowercase       :							Tuple      		=		0
           _lowercase       :							Dict      		=		pipe(**lowerCamelCase).images
           assert image.shape == (1, 32, 32, 3)

           _lowercase       :							List[str]      		=		image[0, -3:, -3:, -1]
           _lowercase       :							Any      		=		np.array([0.5_0_0_4, 0.5_0_0_4, 0.4_9_9_4, 0.5_0_0_8, 0.4_9_7_6, 0.5_0_1_8, 0.4_9_9_0, 0.4_9_8_2, 0.4_9_8_7])

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






@slow
@require_torch_gpu
class _lowerCamelCase(						unittest.TestCase   ):
       def 			UpperCamelCase     (					self)       ->   Union[str, Any]:




           """simple docstring"""

           super().tearDown()
           gc.collect()
           torch.cuda.empty_cache()
       def 			UpperCamelCase     (					self,      lowerCamelCase=0,      lowerCamelCase=False,      lowerCamelCase="cpu",      lowerCamelCase=torch.floataa,      lowerCamelCase=(1, 3, 64, 64))       ->   Optional[Any]:




           """simple docstring"""

           _lowercase       :							List[Any]      		=		torch.manual_seed(lowerCamelCase)

           _lowercase       :							str      		=		{
               'num_inference_steps': None,
               'timesteps': [22, 0],
               'class_labels': 0,
               'generator': generator,
               'output_type': 'np',
           }

           if get_fixed_latents:
               _lowercase       :							Optional[Any]      		=		self.get_fixed_latents(seed=lowerCamelCase,      device=lowerCamelCase,      dtype=lowerCamelCase,      shape=lowerCamelCase)
               _lowercase       :							Tuple      		=		latents

           return inputs
       def 			UpperCamelCase     (					self,      lowerCamelCase=0,      lowerCamelCase="cpu",      lowerCamelCase=torch.floataa,      lowerCamelCase=(1, 3, 64, 64))       ->   Any:




           """simple docstring"""

           if type(lowerCamelCase) == str:
               _lowercase       :							Union[str, Any]      		=		torch.device(lowerCamelCase)
           _lowercase       :							int      		=		torch.Generator(device=lowerCamelCase).manual_seed(lowerCamelCase)
           _lowercase       :							List[str]      		=		randn_tensor(lowerCamelCase,      generator=lowerCamelCase,      device=lowerCamelCase,      dtype=lowerCamelCase)
           return latents
       def 			UpperCamelCase     (					self)       ->   str:




           """simple docstring"""

           _lowercase       :							Tuple      		=		UNetaDModel.from_pretrained('diffusers/consistency_models',      subfolder='diffusers_cd_imagenet64_l2')
           _lowercase       :							Optional[int]      		=		CMStochasticIterativeScheduler(
               num_train_timesteps=40,      sigma_min=0.0_0_2,      sigma_max=8_0.0,      )
           _lowercase       :							Any      		=		ConsistencyModelPipeline(unet=lowerCamelCase,      scheduler=lowerCamelCase)
           pipe.to(torch_device=lowerCamelCase)
           pipe.set_progress_bar_config(disable=lowerCamelCase)

           _lowercase       :							str      		=		self.get_inputs()
           _lowercase       :							Optional[int]      		=		pipe(**lowerCamelCase).images
           assert image.shape == (1, 64, 64, 3)

           _lowercase       :							str      		=		image[0, -3:, -3:, -1]

           _lowercase       :							Optional[Any]      		=		np.array([0.0_8_8_8, 0.0_8_8_1, 0.0_6_6_6, 0.0_4_7_9, 0.0_2_9_2, 0.0_1_9_5, 0.0_2_0_1, 0.0_1_6_3, 0.0_2_5_4])

           assert np.abs(image_slice.flatten() - expected_slice).max() < 2E-2
       def 			UpperCamelCase     (					self)       ->   str:




           """simple docstring"""

           _lowercase       :							List[str]      		=		UNetaDModel.from_pretrained('diffusers/consistency_models',      subfolder='diffusers_cd_imagenet64_l2')
           _lowercase       :							List[str]      		=		CMStochasticIterativeScheduler(
               num_train_timesteps=40,      sigma_min=0.0_0_2,      sigma_max=8_0.0,      )
           _lowercase       :							Union[str, Any]      		=		ConsistencyModelPipeline(unet=lowerCamelCase,      scheduler=lowerCamelCase)
           pipe.to(torch_device=lowerCamelCase)
           pipe.set_progress_bar_config(disable=lowerCamelCase)

           _lowercase       :							List[Any]      		=		self.get_inputs()
           _lowercase       :							int      		=		1
           _lowercase       :							Optional[Any]      		=		None
           _lowercase       :							str      		=		pipe(**lowerCamelCase).images
           assert image.shape == (1, 64, 64, 3)

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

           _lowercase       :							List[str]      		=		np.array([0.0_3_4_0, 0.0_1_5_2, 0.0_0_6_3, 0.0_2_6_7, 0.0_2_2_1, 0.0_1_0_7, 0.0_4_1_6, 0.0_1_8_6, 0.0_2_1_7])

           assert np.abs(image_slice.flatten() - expected_slice).max() < 2E-2
       @require_torch_a
       def 			UpperCamelCase     (					self)       ->   Union[str, Any]:




           """simple docstring"""

           _lowercase       :							str      		=		UNetaDModel.from_pretrained('diffusers/consistency_models',      subfolder='diffusers_cd_imagenet64_l2')
           _lowercase       :							Optional[int]      		=		CMStochasticIterativeScheduler(
               num_train_timesteps=40,      sigma_min=0.0_0_2,      sigma_max=8_0.0,      )
           _lowercase       :							Optional[int]      		=		ConsistencyModelPipeline(unet=lowerCamelCase,      scheduler=lowerCamelCase)
           pipe.to(torch_device=lowerCamelCase,      torch_dtype=torch.floataa)
           pipe.set_progress_bar_config(disable=lowerCamelCase)

           _lowercase       :							Any      		=		self.get_inputs(get_fixed_latents=lowerCamelCase,      device=lowerCamelCase)
           # Ensure usage of flash attention in torch 2.0
           with sdp_kernel(enable_flash=lowerCamelCase,      enable_math=lowerCamelCase,      enable_mem_efficient=lowerCamelCase):
               _lowercase       :							Dict      		=		pipe(**lowerCamelCase).images
           assert image.shape == (1, 64, 64, 3)

           _lowercase       :							Any      		=		image[0, -3:, -3:, -1]

           _lowercase       :							Union[str, Any]      		=		np.array([0.1_8_7_5, 0.1_4_2_8, 0.1_2_8_9, 0.2_1_5_1, 0.2_0_9_2, 0.1_4_7_7, 0.1_8_7_7, 0.1_6_4_1, 0.1_3_5_3])

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




       @require_torch_a
       def 			UpperCamelCase     (					self)       ->   Union[str, Any]:




           """simple docstring"""

           _lowercase       :							Dict      		=		UNetaDModel.from_pretrained('diffusers/consistency_models',      subfolder='diffusers_cd_imagenet64_l2')
           _lowercase       :							Optional[Any]      		=		CMStochasticIterativeScheduler(
               num_train_timesteps=40,      sigma_min=0.0_0_2,      sigma_max=8_0.0,      )
           _lowercase       :							int      		=		ConsistencyModelPipeline(unet=lowerCamelCase,      scheduler=lowerCamelCase)
           pipe.to(torch_device=lowerCamelCase,      torch_dtype=torch.floataa)
           pipe.set_progress_bar_config(disable=lowerCamelCase)

           _lowercase       :							List[Any]      		=		self.get_inputs(get_fixed_latents=lowerCamelCase,      device=lowerCamelCase)
           _lowercase       :							int      		=		1
           _lowercase       :							str      		=		None
           # Ensure usage of flash attention in torch 2.0
           with sdp_kernel(enable_flash=lowerCamelCase,      enable_math=lowerCamelCase,      enable_mem_efficient=lowerCamelCase):
               _lowercase       :							Union[str, Any]      		=		pipe(**lowerCamelCase).images
           assert image.shape == (1, 64, 64, 3)

           _lowercase       :							Any      		=		image[0, -3:, -3:, -1]

           _lowercase       :							int      		=		np.array([0.1_6_6_3, 0.1_9_4_8, 0.2_2_7_5, 0.1_6_8_0, 0.1_2_0_4, 0.1_2_4_5, 0.1_8_5_8, 0.1_3_3_8, 0.2_0_9_5])

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


