



"""simple docstring"""




import tempfile

import torch

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

from .test_schedulers import SchedulerCommonTest






class      UpperCamelCase    (						_lowerCamelCase     ):
	lowerCAmelCase	:   str				=	(UniPCMultistepScheduler,)
	lowerCAmelCase	:   Union[str, Any]				=	(("""num_inference_steps""", 25),)



	def 			__A							(       self			,	**UpperCAmelCase__		):
							A__							=       {
							    "num_train_timesteps": 1_000,
							    "beta_start": 0.0_001,
							    "beta_end": 0.02,
							    "beta_schedule": "linear",
							    "solver_order": 2,
							    "solver_type": "bh2",
							}

							config.update(**UpperCAmelCase__		)
							return config



	def 			__A							(       self			,	UpperCAmelCase__=0			,	**UpperCAmelCase__		):
							A__							=       dict(self.forward_default_kwargs		)
							A__							=       kwargs.pop("num_inference_steps"			,	UpperCAmelCase__		)
							A__							=       self.dummy_sample
							A__							=       0.1 * sample
							A__							=       [residual + 0.2, residual + 0.15, residual + 0.10]

							for scheduler_class in self.scheduler_classes:
													A__							=       self.get_scheduler_config(**UpperCAmelCase__		)
													A__							=       scheduler_class(**UpperCAmelCase__		)
													scheduler.set_timesteps(UpperCAmelCase__		)
													# copy over dummy past residuals
													A__							=       dummy_past_residuals[: scheduler.config.solver_order]

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

													A__						,       A__							=       sample, sample
													for t in range(UpperCAmelCase__			,	time_step + scheduler.config.solver_order + 1		):
																			A__							=       scheduler.step(UpperCAmelCase__			,	UpperCAmelCase__			,	UpperCAmelCase__			,	**UpperCAmelCase__		).prev_sample
																			A__							=       new_scheduler.step(UpperCAmelCase__			,	UpperCAmelCase__			,	UpperCAmelCase__			,	**UpperCAmelCase__		).prev_sample

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



	def 			__A							(       self			,	UpperCAmelCase__=0			,	**UpperCAmelCase__		):
							A__							=       dict(self.forward_default_kwargs		)
							A__							=       kwargs.pop("num_inference_steps"			,	UpperCAmelCase__		)
							A__							=       self.dummy_sample
							A__							=       0.1 * sample
							A__							=       [residual + 0.2, residual + 0.15, residual + 0.10]

							for scheduler_class in self.scheduler_classes:
													A__							=       self.get_scheduler_config()
													A__							=       scheduler_class(**UpperCAmelCase__		)
													scheduler.set_timesteps(UpperCAmelCase__		)

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

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

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

													A__							=       scheduler.step(UpperCAmelCase__			,	UpperCAmelCase__			,	UpperCAmelCase__			,	**UpperCAmelCase__		).prev_sample
													A__							=       new_scheduler.step(UpperCAmelCase__			,	UpperCAmelCase__			,	UpperCAmelCase__			,	**UpperCAmelCase__		).prev_sample

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



	def 			__A							(       self			,	UpperCAmelCase__=None			,	**UpperCAmelCase__		):


							if scheduler is None:
													A__							=       self.scheduler_classes[0]
													A__							=       self.get_scheduler_config(**UpperCAmelCase__		)
													A__							=       scheduler_class(**UpperCAmelCase__		)

							A__							=       self.scheduler_classes[0]
							A__							=       self.get_scheduler_config(**UpperCAmelCase__		)
							A__							=       scheduler_class(**UpperCAmelCase__		)

							A__							=       10
							A__							=       self.dummy_model()
							A__							=       self.dummy_sample_deter
							scheduler.set_timesteps(UpperCAmelCase__		)

							for i, t in enumerate(scheduler.timesteps		):
													A__							=       model(UpperCAmelCase__			,	UpperCAmelCase__		)
													A__							=       scheduler.step(UpperCAmelCase__			,	UpperCAmelCase__			,	UpperCAmelCase__		).prev_sample

							return sample



	def 			__A							(       self		):
							A__							=       dict(self.forward_default_kwargs		)

							A__							=       kwargs.pop("num_inference_steps"			,	UpperCAmelCase__		)

							for scheduler_class in self.scheduler_classes:
													A__							=       self.get_scheduler_config()
													A__							=       scheduler_class(**UpperCAmelCase__		)

													A__							=       self.dummy_sample
													A__							=       0.1 * sample

													if num_inference_steps is not None and hasattr(UpperCAmelCase__			,	"set_timesteps"		):
																			scheduler.set_timesteps(UpperCAmelCase__		)
													elif num_inference_steps is not None and not hasattr(UpperCAmelCase__			,	"set_timesteps"		):
																			A__							=       num_inference_steps

													# copy over dummy past residuals (must be done after set_timesteps)
													A__							=       [residual + 0.2, residual + 0.15, residual + 0.10]
													A__							=       dummy_past_residuals[: scheduler.config.solver_order]

													A__							=       scheduler.timesteps[5]
													A__							=       scheduler.timesteps[6]

													A__							=       scheduler.step(UpperCAmelCase__			,	UpperCAmelCase__			,	UpperCAmelCase__			,	**UpperCAmelCase__		).prev_sample
													A__							=       scheduler.step(UpperCAmelCase__			,	UpperCAmelCase__			,	UpperCAmelCase__			,	**UpperCAmelCase__		).prev_sample

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



	def 			__A							(       self		):
							A__							=       UniPCMultistepScheduler(**self.get_scheduler_config()		)
							A__							=       self.full_loop(scheduler=UpperCAmelCase__		)
							A__							=       torch.mean(torch.abs(UpperCAmelCase__		)		)

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

							A__							=       DPMSolverSinglestepScheduler.from_config(scheduler.config		)
							A__							=       DEISMultistepScheduler.from_config(scheduler.config		)
							A__							=       DPMSolverMultistepScheduler.from_config(scheduler.config		)
							A__							=       UniPCMultistepScheduler.from_config(scheduler.config		)

							A__							=       self.full_loop(scheduler=UpperCAmelCase__		)
							A__							=       torch.mean(torch.abs(UpperCAmelCase__		)		)

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



	def 			__A							(       self		):


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



	def 			__A							(       self		):
							self.check_over_configs(thresholding=UpperCAmelCase__		)
							for order in [1, 2, 3]:
													for solver_type in ["bh1", "bh2"]:
																			for threshold in [0.5, 1.0, 2.0]:
																									for prediction_type in ["epsilon", "sample"]:
																															self.check_over_configs(
																															    thresholding=UpperCAmelCase__			,	prediction_type=UpperCAmelCase__			,	sample_max_value=UpperCAmelCase__			,	solver_order=UpperCAmelCase__			,	solver_type=UpperCAmelCase__			,	)



	def 			__A							(       self		):


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



	def 			__A							(       self		):


							for solver_type in ["bh1", "bh2"]:
													for order in [1, 2, 3]:
																			for prediction_type in ["epsilon", "sample"]:
																									self.check_over_configs(
																									    solver_order=UpperCAmelCase__			,	solver_type=UpperCAmelCase__			,	prediction_type=UpperCAmelCase__			,	)
																									A__							=       self.full_loop(
																									    solver_order=UpperCAmelCase__			,	solver_type=UpperCAmelCase__			,	prediction_type=UpperCAmelCase__			,	)
																									assert not torch.isnan(UpperCAmelCase__		).any(), "Samples have nan numbers"



	def 			__A							(       self		):
							self.check_over_configs(lower_order_final=UpperCAmelCase__		)
							self.check_over_configs(lower_order_final=UpperCAmelCase__		)



	def 			__A							(       self		):


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



	def 			__A							(       self		):
							A__							=       self.full_loop()
							A__							=       torch.mean(torch.abs(UpperCAmelCase__		)		)

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



	def 			__A							(       self		):
							A__							=       self.full_loop(prediction_type="v_prediction"		)
							A__							=       torch.mean(torch.abs(UpperCAmelCase__		)		)

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



	def 			__A							(       self		):
							A__							=       self.scheduler_classes[0]
							A__							=       self.get_scheduler_config(thresholding=UpperCAmelCase__			,	dynamic_thresholding_ratio=0		)
							A__							=       scheduler_class(**UpperCAmelCase__		)

							A__							=       10
							A__							=       self.dummy_model()
							A__							=       self.dummy_sample_deter.half()
							scheduler.set_timesteps(UpperCAmelCase__		)

							for i, t in enumerate(scheduler.timesteps		):
													A__							=       model(UpperCAmelCase__			,	UpperCAmelCase__		)
													A__							=       scheduler.step(UpperCAmelCase__			,	UpperCAmelCase__			,	UpperCAmelCase__		).prev_sample

							assert sample.dtype == torch.floataa



	def 			__A							(       self			,	**UpperCAmelCase__		):


							for scheduler_class in self.scheduler_classes:
													A__							=       self.get_scheduler_config(**UpperCAmelCase__		)
													A__							=       scheduler_class(**UpperCAmelCase__		)

													scheduler.set_timesteps(scheduler.config.num_train_timesteps		)
													assert len(scheduler.timesteps.unique()		) == scheduler.num_inference_steps










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


UpperCAmelCase_							:      Dict      =  False






@skip_mps
class      UpperCamelCase    (						_UpperCAmelCase       ,    _UpperCAmelCase       ,    _UpperCAmelCase       ,    unittest.TestCase     ):
	lowerCAmelCase	:   Optional[Any]				=	StableDiffusionAttendAndExcitePipeline
	lowerCAmelCase	:   Optional[Any]				=	False
	lowerCAmelCase	:   Any				=	TEXT_TO_IMAGE_PARAMS
	lowerCAmelCase	:   Dict				=	TEXT_TO_IMAGE_BATCH_PARAMS.union({"""token_indices"""}     )
	lowerCAmelCase	:   List[str]				=	TEXT_TO_IMAGE_IMAGE_PARAMS
	lowerCAmelCase	:   List[Any]				=	TEXT_TO_IMAGE_IMAGE_PARAMS



	@classmethod
	def 			__A							(       cls		):
							super().setUpClass()
							torch.use_deterministic_algorithms(UpperCAmelCase__		)



	@classmethod
	def 			__A							(       cls		):
							super().tearDownClass()
							torch.use_deterministic_algorithms(UpperCAmelCase__		)



	def 			__A							(       self		):
							torch.manual_seed(0		)
							A__							=       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=UpperCAmelCase__			,	)
							A__							=       DDIMScheduler(
							    beta_start=0.00_085			,	beta_end=0.012			,	beta_schedule="scaled_linear"			,	clip_sample=UpperCAmelCase__			,	set_alpha_to_one=UpperCAmelCase__			,	)
							torch.manual_seed(0		)
							A__							=       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=128			,	)
							torch.manual_seed(0		)
							A__							=       CLIPTextConfig(
							    bos_token_id=0			,	eos_token_id=2			,	hidden_size=32			,	intermediate_size=37			,	layer_norm_eps=1e-0_5			,	num_attention_heads=4			,	num_hidden_layers=5			,	pad_token_id=1			,	vocab_size=1_000			,	hidden_act="gelu"			,	projection_dim=512			,	)
							A__							=       CLIPTextModel(UpperCAmelCase__		)
							A__							=       CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"		)

							A__							=       {
							    "unet": unet,
							    "scheduler": scheduler,
							    "vae": vae,
							    "text_encoder": text_encoder,
							    "tokenizer": tokenizer,
							    "safety_checker": None,
							    "feature_extractor": None,
							}

							return components



	def 			__A							(       self			,	UpperCAmelCase__			,	UpperCAmelCase__=0		):
							if str(UpperCAmelCase__		).startswith("mps"		):
													A__							=       torch.manual_seed(UpperCAmelCase__		)
							else:
													A__							=       torch.Generator(device=UpperCAmelCase__		).manual_seed(UpperCAmelCase__		)
							A__							=       A__							=       {
							    "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		):
							A__							=       "cpu"

							A__							=       self.get_dummy_components()
							A__							=       self.pipeline_class(**UpperCAmelCase__		)
							pipe.to(UpperCAmelCase__		)
							pipe.set_progress_bar_config(disable=UpperCAmelCase__		)

							A__							=       self.get_dummy_inputs(UpperCAmelCase__		)
							A__							=       pipe(**UpperCAmelCase__		).images
							A__							=       image[0, -3:, -3:, -1]

							self.assertEqual(image.shape			,	(1, 64, 64, 3)		)
							A__							=       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]		)
							A__							=       np.abs(image_slice.flatten() - expected_slice		).max()
							self.assertLessEqual(UpperCAmelCase__			,	1e-3		)



	def 			__A							(       self		):
							super().test_cpu_offload_forward_pass(expected_max_diff=5e-4		)



	def 			__A							(       self		):
							# 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		):
							self._test_inference_batch_single_identical(batch_size=2			,	expected_max_diff=7e-4		)



	def 			__A							(       self		):
							super().test_dict_tuple_outputs_equivalent(expected_max_difference=3e-3		)



	def 			__A							(       self		):
							super().test_pt_np_pil_outputs_equivalent(expected_max_diff=5e-4		)



	def 			__A							(       self		):
							super().test_save_load_local(expected_max_difference=5e-4		)



	def 			__A							(       self		):
							super().test_save_load_optional_components(expected_max_difference=4e-4		)






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



	@classmethod
	def 			__A							(       cls		):
							super().setUpClass()
							torch.use_deterministic_algorithms(UpperCAmelCase__		)



	@classmethod
	def 			__A							(       cls		):
							super().tearDownClass()
							torch.use_deterministic_algorithms(UpperCAmelCase__		)



	def 			__A							(       self		):
							super().tearDown()
							gc.collect()
							torch.cuda.empty_cache()



	def 			__A							(       self		):
							A__							=       torch.manual_seed(51		)

							A__							=       StableDiffusionAttendAndExcitePipeline.from_pretrained(
							    "CompVis/stable-diffusion-v1-4"			,	safety_checker=UpperCAmelCase__			,	torch_dtype=torch.floataa		)
							pipe.to("cuda"		)

							A__							=       "a painting of an elephant with glasses"
							A__							=       [5, 7]

							A__							=       pipe(
							    prompt=UpperCAmelCase__			,	token_indices=UpperCAmelCase__			,	guidance_scale=7.5			,	generator=UpperCAmelCase__			,	num_inference_steps=5			,	max_iter_to_alter=5			,	output_type="numpy"			,	).images[0]

							A__							=       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





