import tempfile

import torch

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

from .test_schedulers import SchedulerCommonTest







class       __snake_case (  lowerCamelCase__ ):
 __lowerCamelCase     : Tuple 	=   (DPMSolverMultistepScheduler,)
 __lowerCamelCase     : List[str] 	=   (("""num_inference_steps""", 25),)
 def 						UpperCAmelCase__							(	self		, **snake_case__    )						->				str:
       '''simple docstring'''





       UpperCAmelCase     :			List[Any]     	={
           '''num_train_timesteps''': 1000,
           '''beta_start''': 0.0001,
           '''beta_end''': 0.02,
           '''beta_schedule''': '''linear''',
           '''solver_order''': 2,
           '''prediction_type''': '''epsilon''',
           '''thresholding''': False,
           '''sample_max_value''': 1.0,
           '''algorithm_type''': '''dpmsolver++''',
           '''solver_type''': '''midpoint''',
           '''lower_order_final''': False,
           '''lambda_min_clipped''': -float('''inf'''    ),
           '''variance_type''': None,
       }

       config.update(**snake_case__    )
       return config
 def 						UpperCAmelCase__							(	self		, snake_case__=0		, **snake_case__    )						->				str:
       '''simple docstring'''





       UpperCAmelCase     :			Optional[Any]     	=dict(self.forward_default_kwargs    )
       UpperCAmelCase     :			int     	=kwargs.pop('''num_inference_steps'''		, snake_case__    )
       UpperCAmelCase     :			Optional[Any]     	=self.dummy_sample
       UpperCAmelCase     :			int     	=0.1 * sample
       UpperCAmelCase     :			Union[str, Any]     	=[residual + 0.2, residual + 0.15, residual + 0.10]

       for scheduler_class in self.scheduler_classes:
             UpperCAmelCase     :			Dict     	=self.get_scheduler_config(**snake_case__    )
             UpperCAmelCase     :			List[Any]     	=scheduler_class(**snake_case__    )
             scheduler.set_timesteps(snake_case__    )
             # copy over dummy past residuals
             UpperCAmelCase     :			Any     	=dummy_past_residuals[: scheduler.config.solver_order]

             with tempfile.TemporaryDirectory() as tmpdirname:
                   scheduler.save_config(snake_case__    )
                   UpperCAmelCase     :			Optional[Any]     	=scheduler_class.from_pretrained(snake_case__    )
                   new_scheduler.set_timesteps(snake_case__    )
                   # copy over dummy past residuals
                   UpperCAmelCase     :			str     	=dummy_past_residuals[: new_scheduler.config.solver_order]

             UpperCAmelCase	,    UpperCAmelCase     :			Union[str, Any]     	=sample, sample
             for t in range(snake_case__		, time_step + scheduler.config.solver_order + 1    ):
                   UpperCAmelCase     :			Dict     	=scheduler.step(snake_case__		, snake_case__		, snake_case__		, **snake_case__    ).prev_sample
                   UpperCAmelCase     :			Optional[int]     	=new_scheduler.step(snake_case__		, snake_case__		, snake_case__		, **snake_case__    ).prev_sample

                   assert torch.sum(torch.abs(output - new_output    )    ) < 1e-5, "Scheduler outputs are not identical"
 def 						UpperCAmelCase__							(	self    )						->				str:
       '''simple docstring'''





       pass
 def 						UpperCAmelCase__							(	self		, snake_case__=0		, **snake_case__    )						->				Optional[int]:
       '''simple docstring'''





       UpperCAmelCase     :			Any     	=dict(self.forward_default_kwargs    )
       UpperCAmelCase     :			Dict     	=kwargs.pop('''num_inference_steps'''		, snake_case__    )
       UpperCAmelCase     :			Any     	=self.dummy_sample
       UpperCAmelCase     :			Dict     	=0.1 * sample
       UpperCAmelCase     :			List[Any]     	=[residual + 0.2, residual + 0.15, residual + 0.10]

       for scheduler_class in self.scheduler_classes:
             UpperCAmelCase     :			Any     	=self.get_scheduler_config()
             UpperCAmelCase     :			List[str]     	=scheduler_class(**snake_case__    )
             scheduler.set_timesteps(snake_case__    )

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

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

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

             UpperCAmelCase     :			Any     	=scheduler.step(snake_case__		, snake_case__		, snake_case__		, **snake_case__    ).prev_sample
             UpperCAmelCase     :			List[Any]     	=new_scheduler.step(snake_case__		, snake_case__		, snake_case__		, **snake_case__    ).prev_sample

             assert torch.sum(torch.abs(output - new_output    )    ) < 1e-5, "Scheduler outputs are not identical"
 def 						UpperCAmelCase__							(	self		, snake_case__=None		, **snake_case__    )						->				Dict:
       '''simple docstring'''





       if scheduler is None:
             UpperCAmelCase     :			Dict     	=self.scheduler_classes[0]
             UpperCAmelCase     :			str     	=self.get_scheduler_config(**snake_case__    )
             UpperCAmelCase     :			Union[str, Any]     	=scheduler_class(**snake_case__    )

       UpperCAmelCase     :			int     	=10
       UpperCAmelCase     :			Any     	=self.dummy_model()
       UpperCAmelCase     :			List[Any]     	=self.dummy_sample_deter
       scheduler.set_timesteps(snake_case__    )

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

       return sample
 def 						UpperCAmelCase__							(	self    )						->				Dict:
       '''simple docstring'''





       UpperCAmelCase     :			Any     	=dict(self.forward_default_kwargs    )

       UpperCAmelCase     :			List[str]     	=kwargs.pop('''num_inference_steps'''		, snake_case__    )

       for scheduler_class in self.scheduler_classes:
             UpperCAmelCase     :			Any     	=self.get_scheduler_config()
             UpperCAmelCase     :			Tuple     	=scheduler_class(**snake_case__    )

             UpperCAmelCase     :			Union[str, Any]     	=self.dummy_sample
             UpperCAmelCase     :			Optional[int]     	=0.1 * sample

             if num_inference_steps is not None and hasattr(snake_case__		, '''set_timesteps'''    ):
                   scheduler.set_timesteps(snake_case__    )
             elif num_inference_steps is not None and not hasattr(snake_case__		, '''set_timesteps'''    ):
                   UpperCAmelCase     :			Any     	=num_inference_steps

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

             UpperCAmelCase     :			str     	=scheduler.timesteps[5]
             UpperCAmelCase     :			Optional[Any]     	=scheduler.timesteps[6]

             UpperCAmelCase     :			str     	=scheduler.step(snake_case__		, snake_case__		, snake_case__		, **snake_case__    ).prev_sample
             UpperCAmelCase     :			Optional[int]     	=scheduler.step(snake_case__		, snake_case__		, snake_case__		, **snake_case__    ).prev_sample

             self.assertEqual(output_a.shape		, sample.shape    )
             self.assertEqual(output_a.shape		, output_a.shape    )
 def 						UpperCAmelCase__							(	self    )						->				Optional[int]:
       '''simple docstring'''





       for timesteps in [25, 50, 100, 999, 1000]:
             self.check_over_configs(num_train_timesteps=snake_case__    )
 def 						UpperCAmelCase__							(	self    )						->				int:
       '''simple docstring'''





       self.check_over_configs(thresholding=snake_case__    )
       for order in [1, 2, 3]:
             for solver_type in ["midpoint", "heun"]:
                   for threshold in [0.5, 1.0, 2.0]:
                         for prediction_type in ["epsilon", "sample"]:
                               self.check_over_configs(
                                   thresholding=snake_case__		, prediction_type=snake_case__		, sample_max_value=snake_case__		, algorithm_type='''dpmsolver++'''		, solver_order=snake_case__		, solver_type=snake_case__		, )
 def 						UpperCAmelCase__							(	self    )						->				List[Any]:
       '''simple docstring'''





       for prediction_type in ["epsilon", "v_prediction"]:
             self.check_over_configs(prediction_type=snake_case__    )
 def 						UpperCAmelCase__							(	self    )						->				Union[str, Any]:
       '''simple docstring'''





       for algorithm_type in ["dpmsolver", "dpmsolver++", "sde-dpmsolver", "sde-dpmsolver++"]:
             for solver_type in ["midpoint", "heun"]:
                   for order in [1, 2, 3]:
                         for prediction_type in ["epsilon", "sample"]:
                               if algorithm_type in ["sde-dpmsolver", "sde-dpmsolver++"]:
                                     if order == 3:
                                           continue
                               else:
                                     self.check_over_configs(
                                         solver_order=snake_case__		, solver_type=snake_case__		, prediction_type=snake_case__		, algorithm_type=snake_case__		, )
                               UpperCAmelCase     :			Tuple     	=self.full_loop(
                                   solver_order=snake_case__		, solver_type=snake_case__		, prediction_type=snake_case__		, algorithm_type=snake_case__		, )
                               assert not torch.isnan(snake_case__    ).any(), "Samples have nan numbers"
 def 						UpperCAmelCase__							(	self    )						->				Optional[Any]:
       '''simple docstring'''





       self.check_over_configs(lower_order_final=snake_case__    )
       self.check_over_configs(lower_order_final=snake_case__    )
 def 						UpperCAmelCase__							(	self    )						->				int:
       '''simple docstring'''





       self.check_over_configs(lambda_min_clipped=-float('''inf'''    )    )
       self.check_over_configs(lambda_min_clipped=-5.1    )
 def 						UpperCAmelCase__							(	self    )						->				Optional[Any]:
       '''simple docstring'''





       self.check_over_configs(variance_type=snake_case__    )
       self.check_over_configs(variance_type='''learned_range'''    )
 def 						UpperCAmelCase__							(	self    )						->				int:
       '''simple docstring'''





       for num_inference_steps in [1, 2, 3, 5, 10, 50, 100, 999, 1000]:
             self.check_over_forward(num_inference_steps=snake_case__		, time_step=0    )
 def 						UpperCAmelCase__							(	self    )						->				List[str]:
       '''simple docstring'''





       UpperCAmelCase     :			Union[str, Any]     	=self.full_loop()
       UpperCAmelCase     :			Optional[int]     	=torch.mean(torch.abs(snake_case__    )    )

       assert abs(result_mean.item() - 0.3301    ) < 1e-3
 def 						UpperCAmelCase__							(	self    )						->				Optional[Any]:
       '''simple docstring'''





       UpperCAmelCase     :			str     	=self.full_loop(thresholding=snake_case__		, dynamic_thresholding_ratio=0.87		, sample_max_value=0.5    )
       UpperCAmelCase     :			Dict     	=torch.mean(torch.abs(snake_case__    )    )

       assert abs(result_mean.item() - 1.1364    ) < 1e-3
 def 						UpperCAmelCase__							(	self    )						->				Optional[Any]:
       '''simple docstring'''





       UpperCAmelCase     :			Optional[int]     	=self.full_loop(prediction_type='''v_prediction'''    )
       UpperCAmelCase     :			List[str]     	=torch.mean(torch.abs(snake_case__    )    )

       assert abs(result_mean.item() - 0.2251    ) < 1e-3
 def 						UpperCAmelCase__							(	self    )						->				Dict:
       '''simple docstring'''





       UpperCAmelCase     :			Dict     	=self.full_loop(prediction_type='''v_prediction'''		, use_karras_sigmas=snake_case__    )
       UpperCAmelCase     :			List[str]     	=torch.mean(torch.abs(snake_case__    )    )

       assert abs(result_mean.item() - 0.2096    ) < 1e-3
 def 						UpperCAmelCase__							(	self    )						->				Optional[int]:
       '''simple docstring'''





       UpperCAmelCase     :			Any     	=DPMSolverMultistepScheduler(**self.get_scheduler_config()    )
       UpperCAmelCase     :			Union[str, Any]     	=self.full_loop(scheduler=snake_case__    )
       UpperCAmelCase     :			List[Any]     	=torch.mean(torch.abs(snake_case__    )    )

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

       UpperCAmelCase     :			Optional[int]     	=DPMSolverSinglestepScheduler.from_config(scheduler.config    )
       UpperCAmelCase     :			str     	=UniPCMultistepScheduler.from_config(scheduler.config    )
       UpperCAmelCase     :			Any     	=DEISMultistepScheduler.from_config(scheduler.config    )
       UpperCAmelCase     :			List[str]     	=DPMSolverMultistepScheduler.from_config(scheduler.config    )

       UpperCAmelCase     :			int     	=self.full_loop(scheduler=snake_case__    )
       UpperCAmelCase     :			List[str]     	=torch.mean(torch.abs(snake_case__    )    )

       assert abs(result_mean.item() - 0.3301    ) < 1e-3
 def 						UpperCAmelCase__							(	self    )						->				str:
       '''simple docstring'''





       UpperCAmelCase     :			Union[str, Any]     	=self.scheduler_classes[0]
       UpperCAmelCase     :			Dict     	=self.get_scheduler_config(thresholding=snake_case__		, dynamic_thresholding_ratio=0    )
       UpperCAmelCase     :			Tuple     	=scheduler_class(**snake_case__    )

       UpperCAmelCase     :			List[Any]     	=10
       UpperCAmelCase     :			str     	=self.dummy_model()
       UpperCAmelCase     :			Union[str, Any]     	=self.dummy_sample_deter.half()
       scheduler.set_timesteps(snake_case__    )

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

       assert sample.dtype == torch.floataa






 def 						UpperCAmelCase__							(	self		, **snake_case__    )						->				List[Any]:
       '''simple docstring'''





       for scheduler_class in self.scheduler_classes:
             UpperCAmelCase     :			Dict     	=self.get_scheduler_config(**snake_case__    )
             UpperCAmelCase     :			Tuple     	=scheduler_class(**snake_case__    )

             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, CLIPTextModelWithProjection, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DiffusionPipeline,
    EulerDiscreteScheduler,
    StableDiffusionXLPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, 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 PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()







class       __snake_case (  lowerCamelCase__       , lowerCamelCase__       , unittest.TestCase ):
 __lowerCamelCase     : int 	=   StableDiffusionXLPipeline
 __lowerCamelCase     : Any 	=   TEXT_TO_IMAGE_PARAMS
 __lowerCamelCase     : List[str] 	=   TEXT_TO_IMAGE_BATCH_PARAMS
 __lowerCamelCase     : Optional[int] 	=   TEXT_TO_IMAGE_IMAGE_PARAMS
 __lowerCamelCase     : Union[str, Any] 	=   TEXT_TO_IMAGE_IMAGE_PARAMS
 def 						UpperCAmelCase__							(	self    )						->				Optional[Any]:
       '''simple docstring'''





       torch.manual_seed(0    )
       UpperCAmelCase     :			Union[str, Any]     	=UNetaDConditionModel(
           block_out_channels=(32, 64)		, layers_per_block=2		, sample_size=32		, in_channels=4		, out_channels=4		, down_block_types=('''DownBlock2D''', '''CrossAttnDownBlock2D''')		, up_block_types=('''CrossAttnUpBlock2D''', '''UpBlock2D''')		, attention_head_dim=(2, 4)		, use_linear_projection=snake_case__		, addition_embed_type='''text_time'''		, addition_time_embed_dim=8		, transformer_layers_per_block=(1, 2)		, projection_class_embeddings_input_dim=80		, cross_attention_dim=64		, )
       UpperCAmelCase     :			Optional[int]     	=EulerDiscreteScheduler(
           beta_start=0.0_0085		, beta_end=0.012		, steps_offset=1		, beta_schedule='''scaled_linear'''		, timestep_spacing='''leading'''		, )
       torch.manual_seed(0    )
       UpperCAmelCase     :			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=128		, )
       torch.manual_seed(0    )
       UpperCAmelCase     :			Optional[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=1000		, hidden_act='''gelu'''		, projection_dim=32		, )
       UpperCAmelCase     :			List[str]     	=CLIPTextModel(snake_case__    )
       UpperCAmelCase     :			int     	=CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''		, local_files_only=snake_case__    )

       UpperCAmelCase     :			Tuple     	=CLIPTextModelWithProjection(snake_case__    )
       UpperCAmelCase     :			int     	=CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''		, local_files_only=snake_case__    )

       UpperCAmelCase     :			Optional[int]     	={
           '''unet''': unet,
           '''scheduler''': scheduler,
           '''vae''': vae,
           '''text_encoder''': text_encoder,
           '''tokenizer''': tokenizer,
           '''text_encoder_2''': text_encoder_a,
           '''tokenizer_2''': tokenizer_a,
           # "safety_checker": None,
           # "feature_extractor": None,
       }
       return components
 def 						UpperCAmelCase__							(	self		, snake_case__		, snake_case__=0    )						->				Optional[Any]:
       '''simple docstring'''





       if str(snake_case__    ).startswith('''mps'''    ):
             UpperCAmelCase     :			Union[str, Any]     	=torch.manual_seed(snake_case__    )
       else:
             UpperCAmelCase     :			Optional[int]     	=torch.Generator(device=snake_case__    ).manual_seed(snake_case__    )
       UpperCAmelCase     :			Any     	={
           '''prompt''': '''A painting of a squirrel eating a burger''',
           '''generator''': generator,
           '''num_inference_steps''': 2,
           '''guidance_scale''': 5.0,
           '''output_type''': '''numpy''',
       }
       return inputs
 def 						UpperCAmelCase__							(	self    )						->				List[Any]:
       '''simple docstring'''





       UpperCAmelCase     :			Optional[Any]     	='''cpu'''  # ensure determinism for the device-dependent torch.Generator
       UpperCAmelCase     :			List[Any]     	=self.get_dummy_components()
       UpperCAmelCase     :			Tuple     	=StableDiffusionXLPipeline(**snake_case__    )
       UpperCAmelCase     :			Optional[Any]     	=sd_pipe.to(snake_case__    )
       sd_pipe.set_progress_bar_config(disable=snake_case__    )

       UpperCAmelCase     :			Optional[Any]     	=self.get_dummy_inputs(snake_case__    )
       UpperCAmelCase     :			Optional[int]     	=sd_pipe(**snake_case__    ).images
       UpperCAmelCase     :			str     	=image[0, -3:, -3:, -1]

       assert image.shape == (1, 64, 64, 3)
       UpperCAmelCase     :			Dict     	=np.array([0.5873, 0.6128, 0.4797, 0.5122, 0.5674, 0.4639, 0.5227, 0.5149, 0.4747]    )

       assert np.abs(image_slice.flatten() - expected_slice    ).max() < 1e-2
 def 						UpperCAmelCase__							(	self    )						->				List[str]:
       '''simple docstring'''





       UpperCAmelCase     :			Optional[Any]     	=self.get_dummy_components()
       UpperCAmelCase     :			List[Any]     	=StableDiffusionXLPipeline(**snake_case__    )
       UpperCAmelCase     :			List[str]     	=sd_pipe.to(snake_case__    )
       UpperCAmelCase     :			Optional[int]     	=sd_pipe.to(snake_case__    )
       sd_pipe.set_progress_bar_config(disable=snake_case__    )

       # forward without prompt embeds
       UpperCAmelCase     :			int     	=self.get_dummy_inputs(snake_case__    )
       UpperCAmelCase     :			int     	=3 * ['''this is a negative prompt''']
       UpperCAmelCase     :			Union[str, Any]     	=negative_prompt
       UpperCAmelCase     :			Union[str, Any]     	=3 * [inputs['''prompt''']]

       UpperCAmelCase     :			int     	=sd_pipe(**snake_case__    )
       UpperCAmelCase     :			str     	=output.images[0, -3:, -3:, -1]

       # forward with prompt embeds
       UpperCAmelCase     :			Any     	=self.get_dummy_inputs(snake_case__    )
       UpperCAmelCase     :			List[str]     	=3 * ['''this is a negative prompt''']
       UpperCAmelCase     :			int     	=3 * [inputs.pop('''prompt'''    )]

       (
           (
           UpperCAmelCase
       )	,    (
           UpperCAmelCase
       )	,    (
           UpperCAmelCase
       )	,    (
           UpperCAmelCase
       )	,    
       )     :			Union[str, Any]     	=sd_pipe.encode_prompt(snake_case__		, negative_prompt=snake_case__    )

       UpperCAmelCase     :			Union[str, Any]     	=sd_pipe(
           **snake_case__		, prompt_embeds=snake_case__		, negative_prompt_embeds=snake_case__		, pooled_prompt_embeds=snake_case__		, negative_pooled_prompt_embeds=snake_case__		, )
       UpperCAmelCase     :			Union[str, Any]     	=output.images[0, -3:, -3:, -1]

       # make sure that it's equal
       assert np.abs(image_slice_a.flatten() - image_slice_a.flatten()    ).max() < 1e-4
 def 						UpperCAmelCase__							(	self    )						->				List[Any]:
       '''simple docstring'''





       super().test_attention_slicing_forward_pass(expected_max_diff=3e-3    )






 def 						UpperCAmelCase__							(	self    )						->				Any:
       '''simple docstring'''





       super().test_inference_batch_single_identical(expected_max_diff=3e-3    )







@slow
@require_torch_gpu
class       __snake_case (  unittest.TestCase ):
 def 						UpperCAmelCase__							(	self    )						->				List[str]:
       '''simple docstring'''





       super().tearDown()
       gc.collect()
       torch.cuda.empty_cache()
 def 						UpperCAmelCase__							(	self		, snake_case__		, snake_case__="cpu"		, snake_case__=torch.floataa		, snake_case__=0    )						->				Any:
       '''simple docstring'''





       UpperCAmelCase     :			str     	=torch.Generator(device=snake_case__    ).manual_seed(snake_case__    )
       UpperCAmelCase     :			Optional[Any]     	=np.random.RandomState(snake_case__    ).standard_normal((1, 4, 64, 64)    )
       UpperCAmelCase     :			List[Any]     	=torch.from_numpy(snake_case__    ).to(device=snake_case__		, dtype=snake_case__    )
       UpperCAmelCase     :			Optional[Any]     	={
           '''prompt''': '''a photograph of an astronaut riding a horse''',
           '''latents''': latents,
           '''generator''': generator,
           '''num_inference_steps''': 3,
           '''guidance_scale''': 7.5,
           '''output_type''': '''numpy''',
       }
       return inputs






 def 						UpperCAmelCase__							(	self    )						->				str:
       '''simple docstring'''





       UpperCAmelCase     :			List[Any]     	=DiffusionPipeline.from_pretrained('''stabilityai/stable-diffusion-2-base'''    )
       pipe.to(snake_case__    )
       pipe.set_progress_bar_config(disable=snake_case__    )

       UpperCAmelCase     :			List[Any]     	=self.get_inputs(snake_case__    )
       UpperCAmelCase     :			str     	=pipe(**snake_case__    ).images
       UpperCAmelCase     :			int     	=image[0, -3:, -3:, -1].flatten()

       assert image.shape == (1, 512, 512, 3)
       UpperCAmelCase     :			Any     	=np.array([0.4_9493, 0.4_7896, 0.4_0798, 0.5_4214, 0.5_3212, 0.4_8202, 0.4_7656, 0.4_6329, 0.4_8506]    )
       assert np.abs(image_slice - expected_slice    ).max() < 7e-3
