



"""simple docstring"""







import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    StableDiffusionSAGPipeline,
    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        __A       (	A_     ,A_     ,unittest.TestCase    ):







		'''simple docstring'''





		lowerCAmelCase			:					str   	=				StableDiffusionSAGPipeline
		lowerCAmelCase			:					Union[str, Any]   	=				TEXT_TO_IMAGE_PARAMS
		lowerCAmelCase			:					Tuple   	=				TEXT_TO_IMAGE_BATCH_PARAMS
		lowerCAmelCase			:					List[Any]   	=				TEXT_TO_IMAGE_IMAGE_PARAMS
		lowerCAmelCase			:					str   	=				TEXT_TO_IMAGE_IMAGE_PARAMS
		lowerCAmelCase			:					Optional[int]   	=				False







		def 					UpperCAmelCase    (	self						:  Tuple						)					->		List[str]:






				"""simple docstring"""



				torch.manual_seed(0						)
				lowercase__   :    Dict				=			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''')			,cross_attention_dim=32			,)
				lowercase__   :    Dict				=			DDIMScheduler(
				    beta_start=0.0_0085			,beta_end=0.012			,beta_schedule='''scaled_linear'''			,clip_sample=_snake_case			,set_alpha_to_one=_snake_case			,)
				torch.manual_seed(0						)
				lowercase__   :    Dict				=			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			,)
				torch.manual_seed(0						)
				lowercase__   :    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=1_000			,)
				lowercase__   :    str				=			CLIPTextModel(_snake_case						)
				lowercase__   :    Optional[int]				=			CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''						)

				lowercase__   :    int				=			{
				    '''unet''': unet,
				    '''scheduler''': scheduler,
				    '''vae''': vae,
				    '''text_encoder''': text_encoder,
				    '''tokenizer''': tokenizer,
				    '''safety_checker''': None,
				    '''feature_extractor''': None,
				}
				return components







		def 					UpperCAmelCase    (	self						:  Dict			,_snake_case						:  List[str]			,_snake_case						:  str=0						)					->		List[Any]:






				"""simple docstring"""



				if str(_snake_case						).startswith('''mps'''						):
						lowercase__   :    Optional[Any]				=			torch.manual_seed(_snake_case						)
				else:
						lowercase__   :    int				=			torch.Generator(device=_snake_case						).manual_seed(_snake_case						)
				lowercase__   :    Any				=			{
				    '''prompt''': '''.''',
				    '''generator''': generator,
				    '''num_inference_steps''': 2,
				    '''guidance_scale''': 1.0,
				    '''sag_scale''': 1.0,
				    '''output_type''': '''numpy''',
				}
				return inputs







		def 					UpperCAmelCase    (	self						:  str						)					->		Optional[Any]:






				"""simple docstring"""



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





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







		'''simple docstring'''





		def 					UpperCAmelCase    (	self						:  Optional[Any]						)					->		List[Any]:






				"""simple docstring"""



				super().tearDown()
				gc.collect()
				torch.cuda.empty_cache()







		def 					UpperCAmelCase    (	self						:  Optional[Any]						)					->		Dict:






				"""simple docstring"""



				lowercase__   :    str				=			StableDiffusionSAGPipeline.from_pretrained('''CompVis/stable-diffusion-v1-4'''						)
				lowercase__   :    Any				=			sag_pipe.to(_snake_case						)
				sag_pipe.set_progress_bar_config(disable=_snake_case						)

				lowercase__   :    str				=			'''.'''
				lowercase__   :    Optional[Any]				=			torch.manual_seed(0						)
				lowercase__   :    Any				=			sag_pipe(
				    [prompt]			,generator=_snake_case			,guidance_scale=7.5			,sag_scale=1.0			,num_inference_steps=20			,output_type='''np'''						)

				lowercase__   :    Any				=			output.images

				lowercase__   :    Any				=			image[0, -3:, -3:, -1]

				assert image.shape == (1, 512, 512, 3)
				lowercase__   :    int				=			np.array([0.1568, 0.1738, 0.1695, 0.1693, 0.1507, 0.1705, 0.1547, 0.1751, 0.1949]						)

				assert np.abs(image_slice.flatten() - expected_slice						).max() < 5e-2







		def 					UpperCAmelCase    (	self						:  int						)					->		Any:






				"""simple docstring"""



				lowercase__   :    Dict				=			StableDiffusionSAGPipeline.from_pretrained('''stabilityai/stable-diffusion-2-1-base'''						)
				lowercase__   :    Optional[Any]				=			sag_pipe.to(_snake_case						)
				sag_pipe.set_progress_bar_config(disable=_snake_case						)

				lowercase__   :    Any				=			'''.'''
				lowercase__   :    Optional[Any]				=			torch.manual_seed(0						)
				lowercase__   :    int				=			sag_pipe(
				    [prompt]			,generator=_snake_case			,guidance_scale=7.5			,sag_scale=1.0			,num_inference_steps=20			,output_type='''np'''						)

				lowercase__   :    Dict				=			output.images

				lowercase__   :    Dict				=			image[0, -3:, -3:, -1]

				assert image.shape == (1, 512, 512, 3)
				lowercase__   :    int				=			np.array([0.3459, 0.2876, 0.2537, 0.3002, 0.2671, 0.2160, 0.3026, 0.2262, 0.2371]						)

				assert np.abs(image_slice.flatten() - expected_slice						).max() < 5e-2







		def 					UpperCAmelCase    (	self						:  Any						)					->		Optional[Any]:






				"""simple docstring"""



				lowercase__   :    int				=			StableDiffusionSAGPipeline.from_pretrained('''stabilityai/stable-diffusion-2-1-base'''						)
				lowercase__   :    Tuple				=			sag_pipe.to(_snake_case						)
				sag_pipe.set_progress_bar_config(disable=_snake_case						)

				lowercase__   :    Tuple				=			'''.'''
				lowercase__   :    List[Any]				=			torch.manual_seed(0						)
				lowercase__   :    Dict				=			sag_pipe(
				    [prompt]			,width=768			,height=512			,generator=_snake_case			,guidance_scale=7.5			,sag_scale=1.0			,num_inference_steps=20			,output_type='''np'''			,)

				lowercase__   :    List[str]				=			output.images

				assert image.shape == (1, 512, 768, 3)





"""simple docstring"""







from collections import UserDict
from typing import List, Union

from ..utils import (
    add_end_docstrings,
    is_tf_available,
    is_torch_available,
    is_vision_available,
    logging,
    requires_backends,
)
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_vision_available():
							from PIL import Image

							from ..image_utils import load_image

if is_torch_available():
							from ..models.auto.modeling_auto import MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING

if is_tf_available():
							from ..models.auto.modeling_tf_auto import TF_MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING
							from ..tf_utils import stable_softmax

lowerCAmelCase_   						=   logging.get_logger(__name__)
@add_end_docstrings(A_    )
class        __A       (	A_    ):







		'''simple docstring'''





		def __init__(	self						:  List[str]			,**_snake_case						:  Dict						)					->		List[Any]:






				"""simple docstring"""



				super().__init__(**_snake_case						)

				requires_backends(self			,'''vision'''						)
				self.check_model_type(
				    TF_MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING
				    if self.framework == '''tf'''
				    else MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING						)







		def __call__(	self						:  Optional[int]			,_snake_case						:  Union[str, List[str], "Image", List["Image"]]			,**_snake_case						:  int						)					->		Optional[Any]:






				"""simple docstring"""



				return super().__call__(_snake_case			,**_snake_case						)







		def 					UpperCAmelCase    (	self						:  Dict			,**_snake_case						:  Optional[int]						)					->		List[Any]:






				"""simple docstring"""



				lowercase__   :    List[str]				=			{}
				if "candidate_labels" in kwargs:
						lowercase__   :    Any				=			kwargs['''candidate_labels''']
				if "hypothesis_template" in kwargs:
						lowercase__   :    Optional[Any]				=			kwargs['''hypothesis_template''']

				return preprocess_params, {}, {}







		def 					UpperCAmelCase    (	self						:  Optional[Any]			,_snake_case						:  Optional[int]			,_snake_case						:  Dict=None			,_snake_case						:  Union[str, Any]="This is a photo of {}."						)					->		List[str]:






				"""simple docstring"""



				lowercase__   :    List[Any]				=			load_image(_snake_case						)
				lowercase__   :    int				=			self.image_processor(images=[image]			,return_tensors=self.framework						)
				lowercase__   :    str				=			candidate_labels
				lowercase__   :    Dict				=			[hypothesis_template.format(_snake_case						) for x in candidate_labels]
				lowercase__   :    Any				=			self.tokenizer(_snake_case			,return_tensors=self.framework			,padding=_snake_case						)
				lowercase__   :    Optional[int]				=			[text_inputs]
				return inputs







		def 					UpperCAmelCase    (	self						:  Optional[Any]			,_snake_case						:  Optional[int]						)					->		List[Any]:






				"""simple docstring"""



				lowercase__   :    Optional[int]				=			model_inputs.pop('''candidate_labels'''						)
				lowercase__   :    Union[str, Any]				=			model_inputs.pop('''text_inputs'''						)
				if isinstance(text_inputs[0]			,_snake_case						):
						lowercase__   :    List[str]				=			text_inputs[0]
				else:
						# Batching case.
						lowercase__   :    int				=			text_inputs[0][0]

				lowercase__   :    Tuple				=			self.model(**_snake_case			,**_snake_case						)

				lowercase__   :    Union[str, Any]				=			{
				    '''candidate_labels''': candidate_labels,
				    '''logits''': outputs.logits_per_image,
				}
				return model_outputs







		def 					UpperCAmelCase    (	self						:  Any			,_snake_case						:  Tuple						)					->		Any:






				"""simple docstring"""



				lowercase__   :    Dict				=			model_outputs.pop('''candidate_labels'''						)
				lowercase__   :    Optional[Any]				=			model_outputs['''logits'''][0]
				if self.framework == "pt":
						lowercase__   :    Optional[int]				=			logits.softmax(dim=-1						).squeeze(-1						)
						lowercase__   :    Tuple				=			probs.tolist()
						if not isinstance(_snake_case			,_snake_case						):
								lowercase__   :    Any				=			[scores]
				elif self.framework == "tf":
						lowercase__   :    List[str]				=			stable_softmax(_snake_case			,axis=-1						)
						lowercase__   :    Optional[Any]				=			probs.numpy().tolist()
				else:
						raise ValueError(f"""Unsupported framework: {self.framework}"""						)

				lowercase__   :    Union[str, Any]				=			[
				    {'''score''': score, '''label''': candidate_label}
				    for score, candidate_label in sorted(zip(_snake_case			,_snake_case						)			,key=lambda _snake_case						: -x[0]						)
				]
				return result
