



import gc
import math
import unittest

import torch

from diffusers import UNetaDModel
from diffusers.utils import floats_tensor, logging, slow, torch_all_close, torch_device
from diffusers.utils.testing_utils import enable_full_determinism

from .test_modeling_common import ModelTesterMixin, UNetTesterMixin


__UpperCAmelCase			  =			logging.get_logger(__name__)

enable_full_determinism()



class 							lowerCAmelCase_						(    a__			,       a__			,       unittest.TestCase				):
       UpperCAmelCase__							:   int  	=    UNetaDModel
       UpperCAmelCase__							:   List[Any]  	=    "sample"
       @property
       def    snake_case_			(     self			)	->						Optional[Any]:
          UpperCamelCase       :    List[Any]		      =      4
          UpperCamelCase       :    Any		      =      3
          UpperCamelCase       :    List[Any]		      =      (32, 32)

          UpperCamelCase       :    Union[str, Any]		      =      floats_tensor((batch_size, num_channels) + sizes			).to(SCREAMING_SNAKE_CASE_			)
          UpperCamelCase       :    Any		      =      torch.tensor([10]			).to(SCREAMING_SNAKE_CASE_			)

          return {"sample": noise, "timestep": time_step}
       @property
       def    snake_case_			(     self			)	->						List[str]:
          return (3, 32, 32)
       @property
       def    snake_case_			(     self			)	->						str:
          return (3, 32, 32)






       def    snake_case_			(     self			)	->						Any:
          UpperCamelCase       :    Dict		      =      {
              'block_out_channels': (32, 64),
              'down_block_types': ('DownBlock2D', 'AttnDownBlock2D'),
              'up_block_types': ('AttnUpBlock2D', 'UpBlock2D'),
              'attention_head_dim': 3,
              'out_channels': 3,
              'in_channels': 3,
              'layers_per_block': 2,
              'sample_size': 32,
          }
          UpperCamelCase       :    Tuple		      =      self.dummy_input
          return init_dict, inputs_dict






class 							lowerCAmelCase_						(    a__			,       a__			,       unittest.TestCase				):
       UpperCAmelCase__							:   Optional[int]  	=    UNetaDModel
       UpperCAmelCase__							:   Optional[int]  	=    "sample"
       @property
       def    snake_case_			(     self			)	->						Any:
          UpperCamelCase       :    List[str]		      =      4
          UpperCamelCase       :    Tuple		      =      4
          UpperCamelCase       :    Tuple		      =      (32, 32)

          UpperCamelCase       :    Optional[int]		      =      floats_tensor((batch_size, num_channels) + sizes			).to(SCREAMING_SNAKE_CASE_			)
          UpperCamelCase       :    List[Any]		      =      torch.tensor([10]			).to(SCREAMING_SNAKE_CASE_			)

          return {"sample": noise, "timestep": time_step}
       @property
       def    snake_case_			(     self			)	->						Optional[int]:
          return (4, 32, 32)
       @property
       def    snake_case_			(     self			)	->						Union[str, Any]:
          return (4, 32, 32)
       def    snake_case_			(     self			)	->						str:
          UpperCamelCase       :    Union[str, Any]		      =      {
              'sample_size': 32,
              'in_channels': 4,
              'out_channels': 4,
              'layers_per_block': 2,
              'block_out_channels': (32, 64),
              'attention_head_dim': 32,
              'down_block_types': ('DownBlock2D', 'DownBlock2D'),
              'up_block_types': ('UpBlock2D', 'UpBlock2D'),
          }
          UpperCamelCase       :    Optional[Any]		      =      self.dummy_input
          return init_dict, inputs_dict
       def    snake_case_			(     self			)	->						Optional[Any]:
          UpperCamelCase      ,	UpperCamelCase       :    Any		      =      UNetaDModel.from_pretrained('fusing/unet-ldm-dummy-update',     output_loading_info=SCREAMING_SNAKE_CASE_			)

          self.assertIsNotNone(SCREAMING_SNAKE_CASE_			)
          self.assertEqual(len(loading_info['missing_keys']			),     0			)

          model.to(SCREAMING_SNAKE_CASE_			)
          UpperCamelCase       :    Optional[int]		      =      model(**self.dummy_input			).sample

          assert image is not None, "Make sure output is not None"
       @unittest.skipIf(torch_device != 'cuda',     'This test is supposed to run on GPU'			)
       def    snake_case_			(     self			)	->						Any:
          UpperCamelCase      ,	UpperCamelCase       :    int		      =      UNetaDModel.from_pretrained('fusing/unet-ldm-dummy-update',     output_loading_info=SCREAMING_SNAKE_CASE_			)
          model.to(SCREAMING_SNAKE_CASE_			)
          UpperCamelCase       :    List[str]		      =      model(**self.dummy_input			).sample

          assert image is not None, "Make sure output is not None"
       @unittest.skipIf(torch_device != 'cuda',     'This test is supposed to run on GPU'			)
       def    snake_case_			(     self			)	->						str:
          # by defautl model loading will use accelerate as `low_cpu_mem_usage=True`
          UpperCamelCase      ,	UpperCamelCase       :    Optional[int]		      =      UNetaDModel.from_pretrained('fusing/unet-ldm-dummy-update',     output_loading_info=SCREAMING_SNAKE_CASE_			)
          model_accelerate.to(SCREAMING_SNAKE_CASE_			)
          model_accelerate.eval()

          UpperCamelCase       :    str		      =      torch.randn(
              1,     model_accelerate.config.in_channels,     model_accelerate.config.sample_size,     model_accelerate.config.sample_size,     generator=torch.manual_seed(0			),     )
          UpperCamelCase       :    Union[str, Any]		      =      noise.to(SCREAMING_SNAKE_CASE_			)
          UpperCamelCase       :    Optional[Any]		      =      torch.tensor([10] * noise.shape[0]			).to(SCREAMING_SNAKE_CASE_			)

          UpperCamelCase       :    List[str]		      =      model_accelerate(SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_			)['sample']

          # two models don't need to stay in the device at the same time
          del model_accelerate
          torch.cuda.empty_cache()
          gc.collect()

          UpperCamelCase      ,	UpperCamelCase       :    Optional[Any]		      =      UNetaDModel.from_pretrained(
              'fusing/unet-ldm-dummy-update',     output_loading_info=SCREAMING_SNAKE_CASE_,     low_cpu_mem_usage=SCREAMING_SNAKE_CASE_			)
          model_normal_load.to(SCREAMING_SNAKE_CASE_			)
          model_normal_load.eval()
          UpperCamelCase       :    Tuple		      =      model_normal_load(SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_			)['sample']

          assert torch_all_close(SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_,     rtol=1e-3			)






       def    snake_case_			(     self			)	->						List[str]:
          UpperCamelCase       :    Union[str, Any]		      =      UNetaDModel.from_pretrained('fusing/unet-ldm-dummy-update'			)
          model.eval()
          model.to(SCREAMING_SNAKE_CASE_			)

          UpperCamelCase       :    Dict		      =      torch.randn(
              1,     model.config.in_channels,     model.config.sample_size,     model.config.sample_size,     generator=torch.manual_seed(0			),     )
          UpperCamelCase       :    Optional[Any]		      =      noise.to(SCREAMING_SNAKE_CASE_			)
          UpperCamelCase       :    List[Any]		      =      torch.tensor([10] * noise.shape[0]			).to(SCREAMING_SNAKE_CASE_			)

          with torch.no_grad():
             UpperCamelCase       :    List[str]		      =      model(SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_			).sample

          UpperCamelCase       :    int		      =      output[0, -1, -3:, -3:].flatten().cpu()
          # fmt: off
          UpperCamelCase       :    Tuple		      =      torch.tensor([-13.32_58, -20.11_00, -15.98_73, -17.66_17, -23.05_96, -17.94_19, -13.36_75, -16.18_89, -12.38_00]			)
          # fmt: on

          self.assertTrue(torch_all_close(SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_,     rtol=1e-3			)			)






class 							lowerCAmelCase_						(    a__			,       a__			,       unittest.TestCase				):
       UpperCAmelCase__							:   str  	=    UNetaDModel
       UpperCAmelCase__							:   Optional[Any]  	=    "sample"
       @property
       def    snake_case_			(     self,     SCREAMING_SNAKE_CASE_=(32, 32)			)	->						Any:
          UpperCamelCase       :    List[Any]		      =      4
          UpperCamelCase       :    str		      =      3

          UpperCamelCase       :    Tuple		      =      floats_tensor((batch_size, num_channels) + sizes			).to(SCREAMING_SNAKE_CASE_			)
          UpperCamelCase       :    int		      =      torch.tensor(batch_size * [10]			).to(dtype=torch.intaa,     device=SCREAMING_SNAKE_CASE_			)

          return {"sample": noise, "timestep": time_step}
       @property
       def    snake_case_			(     self			)	->						str:
          return (3, 32, 32)
       @property
       def    snake_case_			(     self			)	->						Optional[Any]:
          return (3, 32, 32)
       def    snake_case_			(     self			)	->						Tuple:
          UpperCamelCase       :    List[str]		      =      {
              'block_out_channels': [32, 64, 64, 64],
              'in_channels': 3,
              'layers_per_block': 1,
              'out_channels': 3,
              'time_embedding_type': 'fourier',
              'norm_eps': 1e-6,
              'mid_block_scale_factor': math.sqrt(2.0			),
              'norm_num_groups': None,
              'down_block_types': [
                  'SkipDownBlock2D',
                  'AttnSkipDownBlock2D',
                  'SkipDownBlock2D',
                  'SkipDownBlock2D',
              ],
              'up_block_types': [
                  'SkipUpBlock2D',
                  'SkipUpBlock2D',
                  'AttnSkipUpBlock2D',
                  'SkipUpBlock2D',
              ],
          }
          UpperCamelCase       :    Union[str, Any]		      =      self.dummy_input
          return init_dict, inputs_dict
       @slow
       def    snake_case_			(     self			)	->						Any:
          UpperCamelCase      ,	UpperCamelCase       :    Tuple		      =      UNetaDModel.from_pretrained('google/ncsnpp-celebahq-256',     output_loading_info=SCREAMING_SNAKE_CASE_			)
          self.assertIsNotNone(SCREAMING_SNAKE_CASE_			)
          self.assertEqual(len(loading_info['missing_keys']			),     0			)

          model.to(SCREAMING_SNAKE_CASE_			)
          UpperCamelCase       :    int		      =      self.dummy_input
          UpperCamelCase       :    Optional[Any]		      =      floats_tensor((4, 3) + (256, 256)			).to(SCREAMING_SNAKE_CASE_			)
          UpperCamelCase       :    Optional[Any]		      =      noise
          UpperCamelCase       :    Union[str, Any]		      =      model(**SCREAMING_SNAKE_CASE_			)

          assert image is not None, "Make sure output is not None"
       @slow
       def    snake_case_			(     self			)	->						List[str]:
          UpperCamelCase       :    List[Any]		      =      UNetaDModel.from_pretrained('google/ncsnpp-celebahq-256'			)
          model.to(SCREAMING_SNAKE_CASE_			)

          UpperCamelCase       :    Union[str, Any]		      =      4
          UpperCamelCase       :    Any		      =      3
          UpperCamelCase       :    Union[str, Any]		      =      (256, 256)

          UpperCamelCase       :    Optional[int]		      =      torch.ones((batch_size, num_channels) + sizes			).to(SCREAMING_SNAKE_CASE_			)
          UpperCamelCase       :    Union[str, Any]		      =      torch.tensor(batch_size * [1e-4]			).to(SCREAMING_SNAKE_CASE_			)

          with torch.no_grad():
             UpperCamelCase       :    Optional[int]		      =      model(SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_			).sample

          UpperCamelCase       :    str		      =      output[0, -3:, -3:, -1].flatten().cpu()
          # fmt: off
          UpperCamelCase       :    Tuple		      =      torch.tensor([-48_42.86_91, -64_99.66_31, -38_00.19_53, -79_78.26_86, -1_09_80.71_29, -2_00_28.85_35, 81_48.28_22, 23_42.29_05, 5_67.76_08]			)
          # fmt: on

          self.assertTrue(torch_all_close(SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_,     rtol=1e-2			)			)
       def    snake_case_			(     self			)	->						Optional[int]:
          UpperCamelCase       :    Tuple		      =      UNetaDModel.from_pretrained('fusing/ncsnpp-ffhq-ve-dummy-update'			)
          model.to(SCREAMING_SNAKE_CASE_			)

          UpperCamelCase       :    str		      =      4
          UpperCamelCase       :    Dict		      =      3
          UpperCamelCase       :    str		      =      (32, 32)

          UpperCamelCase       :    int		      =      torch.ones((batch_size, num_channels) + sizes			).to(SCREAMING_SNAKE_CASE_			)
          UpperCamelCase       :    int		      =      torch.tensor(batch_size * [1e-4]			).to(SCREAMING_SNAKE_CASE_			)

          with torch.no_grad():
             UpperCamelCase       :    Any		      =      model(SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_			).sample

          UpperCamelCase       :    Union[str, Any]		      =      output[0, -3:, -3:, -1].flatten().cpu()
          # fmt: off
          UpperCamelCase       :    Tuple		      =      torch.tensor([-0.03_25, -0.09_00, -0.08_69, -0.03_32, -0.07_25, -0.02_70, -0.01_01, 0.02_27, 0.02_56]			)
          # fmt: on

          self.assertTrue(torch_all_close(SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_,     rtol=1e-2			)			)






       def    snake_case_			(     self			)	->						Optional[int]:
          # not required for this model
          pass





import torch

from diffusers import KDPMaDiscreteScheduler
from diffusers.utils import torch_device

from .test_schedulers import SchedulerCommonTest







class 							lowerCAmelCase__   ( __lowercase						):
						UpperCamelCase_				:			int						=		(KDPMaDiscreteScheduler,)
						UpperCamelCase_				:			Optional[int]						=		10

						def 			A_					(      self							,				**a					)       ->   int:


								'''simple docstring'''

								_UpperCamelCase         =						{
								    """num_train_timesteps""": 11_00,
								    """beta_start""": 0.0001,
								    """beta_end""": 0.02,
								    """beta_schedule""": """linear""",
								}

								config.update(**a					)
								return config

						def 			A_					(      self					)       ->   List[str]:


								'''simple docstring'''

								for timesteps in [10, 50, 1_00, 10_00]:
										self.check_over_configs(num_train_timesteps=a					)

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


								'''simple docstring'''

								for beta_start, beta_end in zip([0.0_0001, 0.0001, 0.001]							,				[0.0002, 0.002, 0.02]					):
										self.check_over_configs(beta_start=a							,				beta_end=a					)

						def 			A_					(      self					)       ->   Any:


								'''simple docstring'''

								for schedule in ["linear", "scaled_linear"]:
										self.check_over_configs(beta_schedule=a					)

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


								'''simple docstring'''

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

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


								'''simple docstring'''

								_UpperCamelCase         =						self.scheduler_classes[0]
								_UpperCamelCase         =						self.get_scheduler_config(prediction_type="""v_prediction"""					)
								_UpperCamelCase         =						scheduler_class(**a					)

								scheduler.set_timesteps(self.num_inference_steps					)

								_UpperCamelCase         =						self.dummy_model()
								_UpperCamelCase         =						self.dummy_sample_deter * scheduler.init_noise_sigma
								_UpperCamelCase         =						sample.to(a					)

								for i, t in enumerate(scheduler.timesteps					):
										_UpperCamelCase         =						scheduler.scale_model_input(a							,				a					)

										_UpperCamelCase         =						model(a							,				a					)

										_UpperCamelCase         =						scheduler.step(a							,				a							,				a					)
										_UpperCamelCase         =						output.prev_sample

								_UpperCamelCase         =						torch.sum(torch.abs(a					)					)
								_UpperCamelCase         =						torch.mean(torch.abs(a					)					)

								if torch_device in ["cpu", "mps"]:
										assert abs(result_sum.item() - 4.6934e-07					) < 1e-2
										assert abs(result_mean.item() - 6.1112e-10					) < 1e-3
								else:
										# CUDA
										assert abs(result_sum.item() - 4.693_4286_5017_0972e-07					) < 1e-2
										assert abs(result_mean.item() - 0.0002					) < 1e-3

						def 			A_					(      self					)       ->   str:


								'''simple docstring'''

								if torch_device == "mps":
										return
								_UpperCamelCase         =						self.scheduler_classes[0]
								_UpperCamelCase         =						self.get_scheduler_config()
								_UpperCamelCase         =						scheduler_class(**a					)

								scheduler.set_timesteps(self.num_inference_steps					)

								_UpperCamelCase         =						self.dummy_model()
								_UpperCamelCase         =						self.dummy_sample_deter * scheduler.init_noise_sigma
								_UpperCamelCase         =						sample.to(a					)

								for i, t in enumerate(scheduler.timesteps					):
										_UpperCamelCase         =						scheduler.scale_model_input(a							,				a					)

										_UpperCamelCase         =						model(a							,				a					)

										_UpperCamelCase         =						scheduler.step(a							,				a							,				a					)
										_UpperCamelCase         =						output.prev_sample

								_UpperCamelCase         =						torch.sum(torch.abs(a					)					)
								_UpperCamelCase         =						torch.mean(torch.abs(a					)					)

								if torch_device in ["cpu", "mps"]:
										assert abs(result_sum.item() - 20.4125					) < 1e-2
										assert abs(result_mean.item() - 0.0266					) < 1e-3
								else:
										# CUDA
										assert abs(result_sum.item() - 20.4125					) < 1e-2
										assert abs(result_mean.item() - 0.0266					) < 1e-3







						def 			A_					(      self					)       ->   str:


								'''simple docstring'''

								if torch_device == "mps":
										return
								_UpperCamelCase         =						self.scheduler_classes[0]
								_UpperCamelCase         =						self.get_scheduler_config()
								_UpperCamelCase         =						scheduler_class(**a					)

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

								_UpperCamelCase         =						self.dummy_model()
								_UpperCamelCase         =						self.dummy_sample_deter.to(a					) * scheduler.init_noise_sigma

								for t in scheduler.timesteps:
										_UpperCamelCase         =						scheduler.scale_model_input(a							,				a					)

										_UpperCamelCase         =						model(a							,				a					)

										_UpperCamelCase         =						scheduler.step(a							,				a							,				a					)
										_UpperCamelCase         =						output.prev_sample

								_UpperCamelCase         =						torch.sum(torch.abs(a					)					)
								_UpperCamelCase         =						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.4125					) < 1e-2
										assert abs(result_mean.item() - 0.0266					) < 1e-3
								else:
										# CUDA
										assert abs(result_sum.item() - 20.4125					) < 1e-2
										assert abs(result_mean.item() - 0.0266					) < 1e-3



