






'''simple docstring'''







import tempfile

import torch

from diffusers import IPNDMScheduler

from .test_schedulers import SchedulerCommonTest





class 						SCREAMING_SNAKE_CASE       (       snake_case__							):
			snake_case__	=					(IPNDMScheduler,)
			snake_case__	=					(('num_inference_steps', 50),)






			def       SCREAMING_SNAKE_CASE   (			self   :      Optional[int]	,    **__SCREAMING_SNAKE_CASE   :      Dict       )      ->      Any:
							a_		:	Any									=   {'num_train_timesteps': 1000}
							config.update(**__SCREAMING_SNAKE_CASE       )
							return config






			def       SCREAMING_SNAKE_CASE   (			self   :      Optional[int]	,    __SCREAMING_SNAKE_CASE   :      Tuple=0	,    **__SCREAMING_SNAKE_CASE   :      int       )      ->      Union[str, Any]:
							a_		:	Tuple									=   dict(self.forward_default_kwargs       )
							a_		:	str									=   kwargs.pop('''num_inference_steps'''	,    __SCREAMING_SNAKE_CASE       )
							a_		:	int									=   self.dummy_sample
							a_		:	List[Any]									=   0.1 * sample
							a_		:	Dict									=   [residual + 0.2, residual + 0.15, residual + 0.1, residual + 0.05]

							for scheduler_class in self.scheduler_classes:
											a_		:	int									=   self.get_scheduler_config(**__SCREAMING_SNAKE_CASE       )
											a_		:	Any									=   scheduler_class(**__SCREAMING_SNAKE_CASE       )
											scheduler.set_timesteps(__SCREAMING_SNAKE_CASE       )
											# copy over dummy past residuals
											a_		:	Optional[Any]									=   dummy_past_residuals[:]

											if time_step is None:
															a_		:	Tuple									=   scheduler.timesteps[len(scheduler.timesteps       ) // 2]

											with tempfile.TemporaryDirectory() as tmpdirname:
															scheduler.save_config(__SCREAMING_SNAKE_CASE       )
															a_		:	Union[str, Any]									=   scheduler_class.from_pretrained(__SCREAMING_SNAKE_CASE       )
															new_scheduler.set_timesteps(__SCREAMING_SNAKE_CASE       )
															# copy over dummy past residuals
															a_		:	int									=   dummy_past_residuals[:]

											a_		:	Dict									=   scheduler.step(__SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE	,    **__SCREAMING_SNAKE_CASE       ).prev_sample
											a_		:	str									=   new_scheduler.step(__SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE	,    **__SCREAMING_SNAKE_CASE       ).prev_sample

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

											a_		:	List[Any]									=   scheduler.step(__SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE	,    **__SCREAMING_SNAKE_CASE       ).prev_sample
											a_		:	List[str]									=   new_scheduler.step(__SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE	,    **__SCREAMING_SNAKE_CASE       ).prev_sample

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






			def       SCREAMING_SNAKE_CASE   (			self   :      str       )      ->      int:
							pass






			def       SCREAMING_SNAKE_CASE   (			self   :      List[str]	,    __SCREAMING_SNAKE_CASE   :      List[Any]=0	,    **__SCREAMING_SNAKE_CASE   :      Union[str, Any]       )      ->      Union[str, Any]:
							a_		:	Any									=   dict(self.forward_default_kwargs       )
							a_		:	str									=   kwargs.pop('''num_inference_steps'''	,    __SCREAMING_SNAKE_CASE       )
							a_		:	Optional[int]									=   self.dummy_sample
							a_		:	Tuple									=   0.1 * sample
							a_		:	List[Any]									=   [residual + 0.2, residual + 0.15, residual + 0.1, residual + 0.05]

							for scheduler_class in self.scheduler_classes:
											a_		:	List[str]									=   self.get_scheduler_config()
											a_		:	int									=   scheduler_class(**__SCREAMING_SNAKE_CASE       )
											scheduler.set_timesteps(__SCREAMING_SNAKE_CASE       )

											# copy over dummy past residuals (must be after setting timesteps)
											a_		:	Optional[int]									=   dummy_past_residuals[:]

											if time_step is None:
															a_		:	List[str]									=   scheduler.timesteps[len(scheduler.timesteps       ) // 2]

											with tempfile.TemporaryDirectory() as tmpdirname:
															scheduler.save_config(__SCREAMING_SNAKE_CASE       )
															a_		:	Tuple									=   scheduler_class.from_pretrained(__SCREAMING_SNAKE_CASE       )
															# copy over dummy past residuals
															new_scheduler.set_timesteps(__SCREAMING_SNAKE_CASE       )

															# copy over dummy past residual (must be after setting timesteps)
															a_		:	Any									=   dummy_past_residuals[:]

											a_		:	Any									=   scheduler.step(__SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE	,    **__SCREAMING_SNAKE_CASE       ).prev_sample
											a_		:	Any									=   new_scheduler.step(__SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE	,    **__SCREAMING_SNAKE_CASE       ).prev_sample

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

											a_		:	List[str]									=   scheduler.step(__SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE	,    **__SCREAMING_SNAKE_CASE       ).prev_sample
											a_		:	Dict									=   new_scheduler.step(__SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE	,    **__SCREAMING_SNAKE_CASE       ).prev_sample

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






			def       SCREAMING_SNAKE_CASE   (			self   :      Union[str, Any]	,    **__SCREAMING_SNAKE_CASE   :      Union[str, Any]       )      ->      List[str]:
							a_		:	List[Any]									=   self.scheduler_classes[0]
							a_		:	str									=   self.get_scheduler_config(**__SCREAMING_SNAKE_CASE       )
							a_		:	int									=   scheduler_class(**__SCREAMING_SNAKE_CASE       )

							a_		:	Tuple									=   10
							a_		:	List[Any]									=   self.dummy_model()
							a_		:	Optional[int]									=   self.dummy_sample_deter
							scheduler.set_timesteps(__SCREAMING_SNAKE_CASE       )

							for i, t in enumerate(scheduler.timesteps       ):
											a_		:	List[str]									=   model(__SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE       )
											a_		:	Any									=   scheduler.step(__SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE       ).prev_sample

							for i, t in enumerate(scheduler.timesteps       ):
											a_		:	int									=   model(__SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE       )
											a_		:	Optional[Any]									=   scheduler.step(__SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE       ).prev_sample

							return sample






			def       SCREAMING_SNAKE_CASE   (			self   :      List[Any]       )      ->      Any:
							a_		:	str									=   dict(self.forward_default_kwargs       )

							a_		:	Any									=   kwargs.pop('''num_inference_steps'''	,    __SCREAMING_SNAKE_CASE       )

							for scheduler_class in self.scheduler_classes:
											a_		:	Tuple									=   self.get_scheduler_config()
											a_		:	Union[str, Any]									=   scheduler_class(**__SCREAMING_SNAKE_CASE       )

											a_		:	Dict									=   self.dummy_sample
											a_		:	str									=   0.1 * sample

											if num_inference_steps is not None and hasattr(__SCREAMING_SNAKE_CASE	,    '''set_timesteps'''       ):
															scheduler.set_timesteps(__SCREAMING_SNAKE_CASE       )
											elif num_inference_steps is not None and not hasattr(__SCREAMING_SNAKE_CASE	,    '''set_timesteps'''       ):
															a_		:	str									=   num_inference_steps

											# copy over dummy past residuals (must be done after set_timesteps)
											a_		:	Union[str, Any]									=   [residual + 0.2, residual + 0.15, residual + 0.1, residual + 0.05]
											a_		:	int									=   dummy_past_residuals[:]

											a_		:	Union[str, Any]									=   scheduler.timesteps[5]
											a_		:	Tuple									=   scheduler.timesteps[6]

											a_		:	Dict									=   scheduler.step(__SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE	,    **__SCREAMING_SNAKE_CASE       ).prev_sample
											a_		:	Tuple									=   scheduler.step(__SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE	,    **__SCREAMING_SNAKE_CASE       ).prev_sample

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

											a_		:	str									=   scheduler.step(__SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE	,    **__SCREAMING_SNAKE_CASE       ).prev_sample
											a_		:	Any									=   scheduler.step(__SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE	,    **__SCREAMING_SNAKE_CASE       ).prev_sample

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






			def       SCREAMING_SNAKE_CASE   (			self   :      int       )      ->      Optional[Any]:
							for timesteps in [100, 1000]:
											self.check_over_configs(num_train_timesteps=__SCREAMING_SNAKE_CASE	,    time_step=__SCREAMING_SNAKE_CASE       )






			def       SCREAMING_SNAKE_CASE   (			self   :      Tuple       )      ->      List[str]:
							for t, num_inference_steps in zip([1, 5, 10]	,    [10, 50, 100]       ):
											self.check_over_forward(num_inference_steps=__SCREAMING_SNAKE_CASE	,    time_step=__SCREAMING_SNAKE_CASE       )







			def       SCREAMING_SNAKE_CASE   (			self   :      Tuple       )      ->      List[str]:
							a_		:	List[str]									=   self.full_loop()
							a_		:	Tuple									=   torch.mean(torch.abs(__SCREAMING_SNAKE_CASE       )       )

							assert abs(result_mean.item() - 254_0529       ) < 10











import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    StableDiffusionAttendAndExcitePipeline,
    UNetaDConditionModel,
)
from diffusers.utils import load_numpy, skip_mps, slow
from diffusers.utils.testing_utils import require_torch_gpu

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


_lowercase		:  Any			       =False






@skip_mps
class 						UpperCamelCase_  (  snake_case__  ,     snake_case__  ,     snake_case__  ,     unittest.TestCase							):
					_a      :						Optional[int]						    =					StableDiffusionAttendAndExcitePipeline
					_a      :						Union[str, Any]						    =					False
					_a      :						Dict						    =					TEXT_TO_IMAGE_PARAMS
					_a      :						List[str]						    =					TEXT_TO_IMAGE_BATCH_PARAMS.union({'token_indices'}							)
					_a      :						Optional[int]						    =					TEXT_TO_IMAGE_IMAGE_PARAMS
					_a      :						List[Any]						    =					TEXT_TO_IMAGE_IMAGE_PARAMS



					@classmethod
					def   __a     (		cls     :     Tuple						):
							super().setUpClass()
							torch.use_deterministic_algorithms(lowerCamelCase						)



					@classmethod
					def   __a     (		cls     :     Tuple						):
							super().tearDownClass()
							torch.use_deterministic_algorithms(lowerCamelCase						)



					def   __a     (		self     :     Dict						):
							torch.manual_seed(0						)
							lowerCamelCase_	:     int 						=      UNetaDConditionModel(
							    block_out_channels=(32, 64)       ,      layers_per_block=1       ,      sample_size=32       ,      in_channels=4       ,      out_channels=4       ,      down_block_types=('DownBlock2D', 'CrossAttnDownBlock2D')       ,      up_block_types=('CrossAttnUpBlock2D', 'UpBlock2D')       ,      cross_attention_dim=32       ,      attention_head_dim=(2, 4)       ,      use_linear_projection=lowerCamelCase       ,      )
							lowerCamelCase_	:     Dict 						=      DDIMScheduler(
							    beta_start=0.00_085       ,      beta_end=0.012       ,      beta_schedule='scaled_linear'       ,      clip_sample=lowerCamelCase       ,      set_alpha_to_one=lowerCamelCase       ,      )
							torch.manual_seed(0						)
							lowerCamelCase_	:     int 						=      AutoencoderKL(
							    block_out_channels=[32, 64]       ,      in_channels=3       ,      out_channels=3       ,      down_block_types=['DownEncoderBlock2D', 'DownEncoderBlock2D']       ,      up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D']       ,      latent_channels=4       ,      sample_size=1_28       ,      )
							torch.manual_seed(0						)
							lowerCamelCase_	:     List[Any] 						=      CLIPTextConfig(
							    bos_token_id=0       ,      eos_token_id=2       ,      hidden_size=32       ,      intermediate_size=37       ,      layer_norm_eps=1E-05       ,      num_attention_heads=4       ,      num_hidden_layers=5       ,      pad_token_id=1       ,      vocab_size=10_00       ,      hidden_act='gelu'       ,      projection_dim=5_12       ,      )
							lowerCamelCase_	:     Any 						=      CLIPTextModel(lowerCamelCase						)
							lowerCamelCase_	:     int 						=      CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'						)

							lowerCamelCase_	:     Optional[Any] 						=      {
							    'unet': unet,
							    'scheduler': scheduler,
							    'vae': vae,
							    'text_encoder': text_encoder,
							    'tokenizer': tokenizer,
							    'safety_checker': None,
							    'feature_extractor': None,
							}

							return components



					def   __a     (		self     :     List[str]       ,      lowerCamelCase     :     Tuple       ,      lowerCamelCase     :     List[Any]=0						):
							if str(lowerCamelCase						).startswith('mps'						):
									lowerCamelCase_	:     Union[str, Any] 						=      torch.manual_seed(lowerCamelCase						)
							else:
									lowerCamelCase_	:     Optional[Any] 						=      torch.Generator(device=lowerCamelCase						).manual_seed(lowerCamelCase						)
							lowerCamelCase_	:     Dict 						=      {
							    'prompt': 'a cat and a frog',
							    'token_indices': [2, 5],
							    'generator': generator,
							    'num_inference_steps': 1,
							    'guidance_scale': 6.0,
							    'output_type': 'numpy',
							    'max_iter_to_alter': 2,
							    'thresholds': {0: 0.7},
							}
							return inputs



					def   __a     (		self     :     Union[str, Any]						):
							lowerCamelCase_	:     List[Any] 						=      'cpu'

							lowerCamelCase_	:     List[Any] 						=      self.get_dummy_components()
							lowerCamelCase_	:     List[str] 						=      self.pipeline_class(**lowerCamelCase						)
							pipe.to(lowerCamelCase						)
							pipe.set_progress_bar_config(disable=lowerCamelCase						)

							lowerCamelCase_	:     List[Any] 						=      self.get_dummy_inputs(lowerCamelCase						)
							lowerCamelCase_	:     List[str] 						=      pipe(**lowerCamelCase						).images
							lowerCamelCase_	:     Optional[Any] 						=      image[0, -3:, -3:, -1]

							self.assertEqual(image.shape       ,      (1, 64, 64, 3)						)
							lowerCamelCase_	:     Dict 						=      np.array(
							    [0.63_905_364, 0.62_897_307, 0.48_599_017, 0.5_133_624, 0.5_550_048, 0.45_769_516, 0.50_326_973, 0.5_023_139, 0.45_384_496]						)
							lowerCamelCase_	:     Any 						=      np.abs(image_slice.flatten() - expected_slice						).max()
							self.assertLessEqual(lowerCamelCase       ,      1E-3						)



					def   __a     (		self     :     Tuple						):
							super().test_cpu_offload_forward_pass(expected_max_diff=5E-4						)



					def   __a     (		self     :     Dict						):
							# NOTE: Larger batch sizes cause this test to timeout, only test on smaller batches
							self._test_inference_batch_consistent(batch_sizes=[1, 2]						)



					def   __a     (		self     :     List[str]						):
							self._test_inference_batch_single_identical(batch_size=2       ,      expected_max_diff=7E-4						)



					def   __a     (		self     :     str						):
							super().test_dict_tuple_outputs_equivalent(expected_max_difference=3E-3						)



					def   __a     (		self     :     Optional[Any]						):
							super().test_pt_np_pil_outputs_equivalent(expected_max_diff=5E-4						)



					def   __a     (		self     :     Tuple						):
							super().test_save_load_local(expected_max_difference=5E-4						)





					def   __a     (		self     :     List[Any]						):
							super().test_save_load_optional_components(expected_max_difference=4E-4						)






@require_torch_gpu
@slow
class 						UpperCamelCase_  (  unittest.TestCase							):



					@classmethod
					def   __a     (		cls     :     Union[str, Any]						):
							super().setUpClass()
							torch.use_deterministic_algorithms(lowerCamelCase						)



					@classmethod
					def   __a     (		cls     :     List[Any]						):
							super().tearDownClass()
							torch.use_deterministic_algorithms(lowerCamelCase						)



					def   __a     (		self     :     List[str]						):
							super().tearDown()
							gc.collect()
							torch.cuda.empty_cache()





					def   __a     (		self     :     str						):
							lowerCamelCase_	:     Optional[Any] 						=      torch.manual_seed(51						)

							lowerCamelCase_	:     Any 						=      StableDiffusionAttendAndExcitePipeline.from_pretrained(
							    'CompVis/stable-diffusion-v1-4'       ,      safety_checker=lowerCamelCase       ,      torch_dtype=torch.floataa						)
							pipe.to('cuda'						)

							lowerCamelCase_	:     List[Any] 						=      'a painting of an elephant with glasses'
							lowerCamelCase_	:     Tuple 						=      [5, 7]

							lowerCamelCase_	:     List[str] 						=      pipe(
							    prompt=lowerCamelCase       ,      token_indices=lowerCamelCase       ,      guidance_scale=7.5       ,      generator=lowerCamelCase       ,      num_inference_steps=5       ,      max_iter_to_alter=5       ,      output_type='numpy'       ,      ).images[0]

							lowerCamelCase_	:     Optional[Any] 						=      load_numpy(
							    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/attend-and-excite/elephant_glasses.npy'						)
							assert np.abs((expected_image - image).max()						) < 5E-1
