from typing import Any, Callable, Dict, List, Optional, Union

import torch
from transformers import CLIPImageProcessor, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DiffusionPipeline,
    LMSDiscreteScheduler,
    PNDMScheduler,
    StableDiffusionPipeline,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion import StableDiffusionPipelineOutput
from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker


_lowerCamelCase					:   Optional[int]							=		'''CompVis/stable-diffusion-v1-1'''
_lowerCamelCase					:   List[Any]							=		'''CompVis/stable-diffusion-v1-2'''
_lowerCamelCase					:   str							=		'''CompVis/stable-diffusion-v1-3'''
_lowerCamelCase					:   Union[str, Any]							=		'''CompVis/stable-diffusion-v1-4'''


class 						lowercase    (     __lowercase     ):




       def __init__(						self   :					Dict	,							_UpperCamelCase   :					AutoencoderKL	,							_UpperCamelCase   :					CLIPTextModel	,							_UpperCamelCase   :					CLIPTokenizer	,							_UpperCamelCase   :					UNetaDConditionModel	,							_UpperCamelCase   :					Union[DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler]	,							_UpperCamelCase   :					StableDiffusionSafetyChecker	,							_UpperCamelCase   :					CLIPImageProcessor	,							_UpperCamelCase   :					bool = True	,							)					->				Any:



            '''simple docstring'''

            super()._init_()

            SCREAMING_SNAKE_CASE    			=       StableDiffusionPipeline.from_pretrained(_A      )
            SCREAMING_SNAKE_CASE    			=       StableDiffusionPipeline.from_pretrained(_A      )
            SCREAMING_SNAKE_CASE    			=       StableDiffusionPipeline.from_pretrained(_A      )
            SCREAMING_SNAKE_CASE    			=       StableDiffusionPipeline(
                vae=_A	,							text_encoder=_A	,							tokenizer=_A	,							unet=_A	,							scheduler=_A	,							safety_checker=_A	,							feature_extractor=_A	,							requires_safety_checker=_A	,							)

            self.register_modules(pipelinea=self.pipea	,							pipelinea=self.pipea	,							pipelinea=self.pipea	,							pipelinea=self.pipea      )




       @property
       def 	__snake_case(						self   :					Optional[Any]      )					->				Optional[Any]:



            '''simple docstring'''

            return {k: getattr(self	,							_A      ) for k in self.config.keys() if not k.startswith("_"      )}




       def 	__snake_case(						self   :					List[str]	,							_UpperCamelCase   :					Optional[Union[str, int]] = "auto"      )					->				Tuple:



            '''simple docstring'''

            if slice_size == "auto":
                 # half the attention head size is usually a good trade-off between
                 # speed and memory
                 SCREAMING_SNAKE_CASE    			=       self.unet.config.attention_head_dim // 2
            self.unet.set_attention_slice(_A      )




       def 	__snake_case(						self   :					Any      )					->				int:



            '''simple docstring'''

            self.enable_attention_slicing(_A      )




       @torch.no_grad()
       def 	__snake_case(						self   :					Optional[Any]	,							_UpperCamelCase   :					Union[str, List[str]]	,							_UpperCamelCase   :					int = 512	,							_UpperCamelCase   :					int = 512	,							_UpperCamelCase   :					int = 50	,							_UpperCamelCase   :					float = 7.5	,							_UpperCamelCase   :					Optional[Union[str, List[str]]] = None	,							_UpperCamelCase   :					Optional[int] = 1	,							_UpperCamelCase   :					float = 0.0	,							_UpperCamelCase   :					Optional[torch.Generator] = None	,							_UpperCamelCase   :					Optional[torch.FloatTensor] = None	,							_UpperCamelCase   :					Optional[str] = "pil"	,							_UpperCamelCase   :					bool = True	,							_UpperCamelCase   :					Optional[Callable[[int, int, torch.FloatTensor], None]] = None	,							_UpperCamelCase   :					int = 1	,							**_UpperCamelCase   :					Dict	,							)					->				List[str]:



            '''simple docstring'''

            return self.pipea(
                prompt=_A	,							height=_A	,							width=_A	,							num_inference_steps=_A	,							guidance_scale=_A	,							negative_prompt=_A	,							num_images_per_prompt=_A	,							eta=_A	,							generator=_A	,							latents=_A	,							output_type=_A	,							return_dict=_A	,							callback=_A	,							callback_steps=_A	,							**_A	,							)




       @torch.no_grad()
       def 	__snake_case(						self   :					Any	,							_UpperCamelCase   :					Union[str, List[str]]	,							_UpperCamelCase   :					int = 512	,							_UpperCamelCase   :					int = 512	,							_UpperCamelCase   :					int = 50	,							_UpperCamelCase   :					float = 7.5	,							_UpperCamelCase   :					Optional[Union[str, List[str]]] = None	,							_UpperCamelCase   :					Optional[int] = 1	,							_UpperCamelCase   :					float = 0.0	,							_UpperCamelCase   :					Optional[torch.Generator] = None	,							_UpperCamelCase   :					Optional[torch.FloatTensor] = None	,							_UpperCamelCase   :					Optional[str] = "pil"	,							_UpperCamelCase   :					bool = True	,							_UpperCamelCase   :					Optional[Callable[[int, int, torch.FloatTensor], None]] = None	,							_UpperCamelCase   :					int = 1	,							**_UpperCamelCase   :					Optional[Any]	,							)					->				Any:



            '''simple docstring'''

            return self.pipea(
                prompt=_A	,							height=_A	,							width=_A	,							num_inference_steps=_A	,							guidance_scale=_A	,							negative_prompt=_A	,							num_images_per_prompt=_A	,							eta=_A	,							generator=_A	,							latents=_A	,							output_type=_A	,							return_dict=_A	,							callback=_A	,							callback_steps=_A	,							**_A	,							)




       @torch.no_grad()
       def 	__snake_case(						self   :					List[str]	,							_UpperCamelCase   :					Union[str, List[str]]	,							_UpperCamelCase   :					int = 512	,							_UpperCamelCase   :					int = 512	,							_UpperCamelCase   :					int = 50	,							_UpperCamelCase   :					float = 7.5	,							_UpperCamelCase   :					Optional[Union[str, List[str]]] = None	,							_UpperCamelCase   :					Optional[int] = 1	,							_UpperCamelCase   :					float = 0.0	,							_UpperCamelCase   :					Optional[torch.Generator] = None	,							_UpperCamelCase   :					Optional[torch.FloatTensor] = None	,							_UpperCamelCase   :					Optional[str] = "pil"	,							_UpperCamelCase   :					bool = True	,							_UpperCamelCase   :					Optional[Callable[[int, int, torch.FloatTensor], None]] = None	,							_UpperCamelCase   :					int = 1	,							**_UpperCamelCase   :					List[Any]	,							)					->				int:



            '''simple docstring'''

            return self.pipea(
                prompt=_A	,							height=_A	,							width=_A	,							num_inference_steps=_A	,							guidance_scale=_A	,							negative_prompt=_A	,							num_images_per_prompt=_A	,							eta=_A	,							generator=_A	,							latents=_A	,							output_type=_A	,							return_dict=_A	,							callback=_A	,							callback_steps=_A	,							**_A	,							)




       @torch.no_grad()
       def 	__snake_case(						self   :					Dict	,							_UpperCamelCase   :					Union[str, List[str]]	,							_UpperCamelCase   :					int = 512	,							_UpperCamelCase   :					int = 512	,							_UpperCamelCase   :					int = 50	,							_UpperCamelCase   :					float = 7.5	,							_UpperCamelCase   :					Optional[Union[str, List[str]]] = None	,							_UpperCamelCase   :					Optional[int] = 1	,							_UpperCamelCase   :					float = 0.0	,							_UpperCamelCase   :					Optional[torch.Generator] = None	,							_UpperCamelCase   :					Optional[torch.FloatTensor] = None	,							_UpperCamelCase   :					Optional[str] = "pil"	,							_UpperCamelCase   :					bool = True	,							_UpperCamelCase   :					Optional[Callable[[int, int, torch.FloatTensor], None]] = None	,							_UpperCamelCase   :					int = 1	,							**_UpperCamelCase   :					Optional[int]	,							)					->				Optional[int]:



            '''simple docstring'''

            return self.pipea(
                prompt=_A	,							height=_A	,							width=_A	,							num_inference_steps=_A	,							guidance_scale=_A	,							negative_prompt=_A	,							num_images_per_prompt=_A	,							eta=_A	,							generator=_A	,							latents=_A	,							output_type=_A	,							return_dict=_A	,							callback=_A	,							callback_steps=_A	,							**_A	,							)





       @torch.no_grad()
       def 	__snake_case(						self   :					Dict	,							_UpperCamelCase   :					Union[str, List[str]]	,							_UpperCamelCase   :					int = 512	,							_UpperCamelCase   :					int = 512	,							_UpperCamelCase   :					int = 50	,							_UpperCamelCase   :					float = 7.5	,							_UpperCamelCase   :					Optional[Union[str, List[str]]] = None	,							_UpperCamelCase   :					Optional[int] = 1	,							_UpperCamelCase   :					float = 0.0	,							_UpperCamelCase   :					Optional[torch.Generator] = None	,							_UpperCamelCase   :					Optional[torch.FloatTensor] = None	,							_UpperCamelCase   :					Optional[str] = "pil"	,							_UpperCamelCase   :					bool = True	,							_UpperCamelCase   :					Optional[Callable[[int, int, torch.FloatTensor], None]] = None	,							_UpperCamelCase   :					int = 1	,							**_UpperCamelCase   :					List[Any]	,							)					->				int:



            '''simple docstring'''

            SCREAMING_SNAKE_CASE    			=       "cuda" if torch.cuda.is_available() else "cpu"
            self.to(_A      )

            # Checks if the height and width are divisible by 8 or not
            if height % 8 != 0 or width % 8 != 0:
                 raise ValueError(F"`height` and `width` must be divisible by 8 but are {height} and {width}."      )

            # Get first result from Stable Diffusion Checkpoint v1.1
            SCREAMING_SNAKE_CASE    			=       self.textaimg_sda_a(
                prompt=_A	,							height=_A	,							width=_A	,							num_inference_steps=_A	,							guidance_scale=_A	,							negative_prompt=_A	,							num_images_per_prompt=_A	,							eta=_A	,							generator=_A	,							latents=_A	,							output_type=_A	,							return_dict=_A	,							callback=_A	,							callback_steps=_A	,							**_A	,							)

            # Get first result from Stable Diffusion Checkpoint v1.2
            SCREAMING_SNAKE_CASE    			=       self.textaimg_sda_a(
                prompt=_A	,							height=_A	,							width=_A	,							num_inference_steps=_A	,							guidance_scale=_A	,							negative_prompt=_A	,							num_images_per_prompt=_A	,							eta=_A	,							generator=_A	,							latents=_A	,							output_type=_A	,							return_dict=_A	,							callback=_A	,							callback_steps=_A	,							**_A	,							)

            # Get first result from Stable Diffusion Checkpoint v1.3
            SCREAMING_SNAKE_CASE    			=       self.textaimg_sda_a(
                prompt=_A	,							height=_A	,							width=_A	,							num_inference_steps=_A	,							guidance_scale=_A	,							negative_prompt=_A	,							num_images_per_prompt=_A	,							eta=_A	,							generator=_A	,							latents=_A	,							output_type=_A	,							return_dict=_A	,							callback=_A	,							callback_steps=_A	,							**_A	,							)

            # Get first result from Stable Diffusion Checkpoint v1.4
            SCREAMING_SNAKE_CASE    			=       self.textaimg_sda_a(
                prompt=_A	,							height=_A	,							width=_A	,							num_inference_steps=_A	,							guidance_scale=_A	,							negative_prompt=_A	,							num_images_per_prompt=_A	,							eta=_A	,							generator=_A	,							latents=_A	,							output_type=_A	,							return_dict=_A	,							callback=_A	,							callback_steps=_A	,							**_A	,							)

            # Get all result images into a single list and pass it via StableDiffusionPipelineOutput for final result
            return StableDiffusionPipelineOutput([resa[0], resa[0], resa[0], resa[0]]      )





import torch

from diffusers import KDPMaDiscreteScheduler
from diffusers.utils import torch_device

from .test_schedulers import SchedulerCommonTest


class 						lowercase    (     a     ):
							lowercase__						:       Tuple								=							(KDPMaDiscreteScheduler,)
							lowercase__						:       Optional[int]								=							10




							def 	__snake_case(						self   :					Optional[Any]	,							**_UpperCamelCase   :					List[Any]      )					->				Tuple:



												'''simple docstring'''

												SCREAMING_SNAKE_CASE    			=       {
												    "num_train_timesteps": 1_100,
												    "beta_start": 0.0_0_0_1,
												    "beta_end": 0.0_2,
												    "beta_schedule": "linear",
												}

												config.update(**_UpperCamelCase      )
												return config




							def 	__snake_case(						self   :					int      )					->				List[Any]:



												'''simple docstring'''

												for timesteps in [10, 50, 100, 1_000]:
																	self.check_over_configs(num_train_timesteps=_UpperCamelCase      )




							def 	__snake_case(						self   :					Union[str, Any]      )					->				Union[str, Any]:



												'''simple docstring'''

												for beta_start, beta_end in zip([0.0_0_0_0_1, 0.0_0_0_1, 0.0_0_1]	,							[0.0_0_0_2, 0.0_0_2, 0.0_2]      ):
																	self.check_over_configs(beta_start=_UpperCamelCase	,							beta_end=_UpperCamelCase      )




							def 	__snake_case(						self   :					Union[str, Any]      )					->				str:



												'''simple docstring'''

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




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



												'''simple docstring'''

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




							def 	__snake_case(						self   :					Optional[int]      )					->				Any:



												'''simple docstring'''

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

												scheduler.set_timesteps(self.num_inference_steps      )

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

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

																	SCREAMING_SNAKE_CASE    			=       model(_UpperCamelCase	,							_UpperCamelCase      )

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

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

												if torch_device in ["cpu", "mps"]:
																	assert abs(result_sum.item() - 4.6_934e-07      ) < 1e-2
																	assert abs(result_mean.item() - 6.1_112e-10      ) < 1e-3
												else:
																	# CUDA
																	assert abs(result_sum.item() - 4.693_428_650_170_972e-07      ) < 1e-2
																	assert abs(result_mean.item() - 0.0_0_0_2      ) < 1e-3




							def 	__snake_case(						self   :					Dict      )					->				int:



												'''simple docstring'''

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

												scheduler.set_timesteps(self.num_inference_steps      )

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

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

																	SCREAMING_SNAKE_CASE    			=       model(_UpperCamelCase	,							_UpperCamelCase      )

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

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

												if torch_device in ["cpu", "mps"]:
																	assert abs(result_sum.item() - 2_0.4_1_2_5      ) < 1e-2
																	assert abs(result_mean.item() - 0.0_2_6_6      ) < 1e-3
												else:
																	# CUDA
																	assert abs(result_sum.item() - 2_0.4_1_2_5      ) < 1e-2
																	assert abs(result_mean.item() - 0.0_2_6_6      ) < 1e-3





							def 	__snake_case(						self   :					Tuple      )					->				str:



												'''simple docstring'''

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

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

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

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

																	SCREAMING_SNAKE_CASE    			=       model(_UpperCamelCase	,							_UpperCamelCase      )

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

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

												if str(_UpperCamelCase      ).startswith("cpu"      ):
																	# The following sum varies between 148 and 156 on mps. Why?
																	assert abs(result_sum.item() - 2_0.4_1_2_5      ) < 1e-2
																	assert abs(result_mean.item() - 0.0_2_6_6      ) < 1e-3
												else:
																	# CUDA
																	assert abs(result_sum.item() - 2_0.4_1_2_5      ) < 1e-2
																	assert abs(result_mean.item() - 0.0_2_6_6      ) < 1e-3




