



"""simple docstring"""






import gc
import random
import unittest

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

from diffusers import (
    AutoencoderKL,
    DiffusionPipeline,
    EulerDiscreteScheduler,
    StableDiffusionXLImgaImgPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import floats_tensor, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import (
    IMAGE_TO_IMAGE_IMAGE_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()






class 			__snake_case			(      _lowercase    ,	_lowercase    ,	unittest.TestCase):
  snake_case__							:  Optional[Any]         =		StableDiffusionXLImgaImgPipeline
  snake_case__							:  Union[str, Any]         =		TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {"height", "width"}
  snake_case__							:  int         =		PipelineTesterMixin.required_optional_params - {"latents"}
  snake_case__							:  Union[str, Any]         =		TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
  snake_case__							:  Dict         =		IMAGE_TO_IMAGE_IMAGE_PARAMS
  snake_case__							:  Any         =		IMAGE_TO_IMAGE_IMAGE_PARAMS




  def 		SCREAMING_SNAKE_CASE							(					self	:		Optional[Any]						):






         """simple docstring"""






         torch.manual_seed(0						)
         _lowerCamelCase		:							Union[str, Any]            =					UNetaDConditionModel(
             block_out_channels=(3_2, 6_4)		,   layers_per_block=2		,   sample_size=3_2		,   in_channels=4		,   out_channels=4		,   down_block_types=('''DownBlock2D''', '''CrossAttnDownBlock2D''')		,   up_block_types=('''CrossAttnUpBlock2D''', '''UpBlock2D''')		,   attention_head_dim=(2, 4)		,   use_linear_projection=__lowerCAmelCase		,   addition_embed_type='''text_time'''		,   addition_time_embed_dim=8		,   transformer_layers_per_block=(1, 2)		,   projection_class_embeddings_input_dim=8_0		,   cross_attention_dim=6_4		,   )
         _lowerCamelCase		:							Dict            =					EulerDiscreteScheduler(
             beta_start=0.0_00_85		,   beta_end=0.0_12		,   steps_offset=1		,   beta_schedule='''scaled_linear'''		,   timestep_spacing='''leading'''		,   )
         torch.manual_seed(0						)
         _lowerCamelCase		:							str            =					AutoencoderKL(
             block_out_channels=[3_2, 6_4]		,   in_channels=3		,   out_channels=3		,   down_block_types=['''DownEncoderBlock2D''', '''DownEncoderBlock2D''']		,   up_block_types=['''UpDecoderBlock2D''', '''UpDecoderBlock2D''']		,   latent_channels=4		,   sample_size=1_2_8		,   )
         torch.manual_seed(0						)
         _lowerCamelCase		:							Any            =					CLIPTextConfig(
             bos_token_id=0		,   eos_token_id=2		,   hidden_size=3_2		,   intermediate_size=3_7		,   layer_norm_eps=1E-05		,   num_attention_heads=4		,   num_hidden_layers=5		,   pad_token_id=1		,   vocab_size=1_0_0_0		,   hidden_act='''gelu'''		,   projection_dim=3_2		,   )
         _lowerCamelCase		:							Dict            =					CLIPTextModel(__lowerCAmelCase						)
         _lowerCamelCase		:							Dict            =					CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''		,   local_files_only=__lowerCAmelCase						)

         _lowerCamelCase		:							List[str]            =					CLIPTextModelWithProjection(__lowerCAmelCase						)
         _lowerCamelCase		:							str            =					CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''		,   local_files_only=__lowerCAmelCase						)

         _lowerCamelCase		:							List[Any]            =					{
             '''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 		SCREAMING_SNAKE_CASE							(					self	:		int		,   __lowerCAmelCase	:		Tuple		,   __lowerCAmelCase	:		Any=0						):






         """simple docstring"""






         _lowerCamelCase		:							Dict            =					floats_tensor((1, 3, 3_2, 3_2)		,   rng=random.Random(__lowerCAmelCase						)						).to(__lowerCAmelCase						)
         _lowerCamelCase		:							str            =					image / 2 + 0.5
         if str(__lowerCAmelCase						).startswith('''mps'''						):
                _lowerCamelCase		:							Any            =					torch.manual_seed(__lowerCAmelCase						)
         else:
                _lowerCamelCase		:							str            =					torch.Generator(device=__lowerCAmelCase						).manual_seed(__lowerCAmelCase						)
         _lowerCamelCase		:							Union[str, Any]            =					{
             '''prompt''': '''A painting of a squirrel eating a burger''',
             '''image''': image,
             '''generator''': generator,
             '''num_inference_steps''': 2,
             '''guidance_scale''': 5.0,
             '''output_type''': '''numpy''',
             '''strength''': 0.75,
         }
         return inputs




  def 		SCREAMING_SNAKE_CASE							(					self	:		Dict						):






         """simple docstring"""






         _lowerCamelCase		:							Dict            =					'''cpu'''  # ensure determinism for the device-dependent torch.Generator
         _lowerCamelCase		:							int            =					self.get_dummy_components()
         _lowerCamelCase		:							Any            =					StableDiffusionXLImgaImgPipeline(**__lowerCAmelCase						)
         _lowerCamelCase		:							List[Any]            =					sd_pipe.to(__lowerCAmelCase						)
         sd_pipe.set_progress_bar_config(disable=__lowerCAmelCase						)

         _lowerCamelCase		:							Optional[Any]            =					self.get_dummy_inputs(__lowerCAmelCase						)
         _lowerCamelCase		:							List[Any]            =					sd_pipe(**__lowerCAmelCase						).images
         _lowerCamelCase		:							Tuple            =					image[0, -3:, -3:, -1]

         assert image.shape == (1, 3_2, 3_2, 3)

         _lowerCamelCase		:							Any            =					np.array([0.46_56, 0.48_40, 0.44_39, 0.66_98, 0.55_74, 0.45_24, 0.57_99, 0.59_43, 0.51_65]						)

         assert np.abs(image_slice.flatten() - expected_slice						).max() < 1E-2




  def 		SCREAMING_SNAKE_CASE							(					self	:		str						):






         """simple docstring"""






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




  def 		SCREAMING_SNAKE_CASE							(					self	:		List[str]						):






         """simple docstring"""






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




  def 		SCREAMING_SNAKE_CASE							(					self	:		Any						):






         """simple docstring"""






         pass




  def 		SCREAMING_SNAKE_CASE							(					self	:		int						):






         """simple docstring"""






         _lowerCamelCase		:							Dict            =					self.get_dummy_components()
         _lowerCamelCase		:							Dict            =					StableDiffusionXLImgaImgPipeline(**__lowerCAmelCase						)
         _lowerCamelCase		:							Optional[Any]            =					sd_pipe.to(__lowerCAmelCase						)
         _lowerCamelCase		:							Optional[int]            =					sd_pipe.to(__lowerCAmelCase						)
         sd_pipe.set_progress_bar_config(disable=__lowerCAmelCase						)

         # forward without prompt embeds
         _lowerCamelCase		:							List[Any]            =					self.get_dummy_inputs(__lowerCAmelCase						)
         _lowerCamelCase		:							str            =					3 * ['''this is a negative prompt''']
         _lowerCamelCase		:							Dict            =					negative_prompt
         _lowerCamelCase		:							int            =					3 * [inputs['''prompt''']]

         _lowerCamelCase		:							Dict            =					sd_pipe(**__lowerCAmelCase						)
         _lowerCamelCase		:							Tuple            =					output.images[0, -3:, -3:, -1]

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

         (
             (
             _lowerCamelCase
         )     ,       (
             _lowerCamelCase
         )     ,       (
             _lowerCamelCase
         )     ,       (
             _lowerCamelCase
         )     ,       
         )		:							Any            =					sd_pipe.encode_prompt(__lowerCAmelCase		,   negative_prompt=__lowerCAmelCase						)

         _lowerCamelCase		:							Optional[Any]            =					sd_pipe(
             **__lowerCAmelCase		,   prompt_embeds=__lowerCAmelCase		,   negative_prompt_embeds=__lowerCAmelCase		,   pooled_prompt_embeds=__lowerCAmelCase		,   negative_pooled_prompt_embeds=__lowerCAmelCase		,   )
         _lowerCamelCase		:							str            =					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






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




  def 		SCREAMING_SNAKE_CASE							(					self	:		List[str]						):






         """simple docstring"""






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




  def 		SCREAMING_SNAKE_CASE							(					self	:		str		,   __lowerCAmelCase	:		Any		,   __lowerCAmelCase	:		Tuple="cpu"		,   __lowerCAmelCase	:		Tuple=torch.floataa		,   __lowerCAmelCase	:		Dict=0						):






         """simple docstring"""






         _lowerCamelCase		:							Tuple            =					torch.Generator(device=__lowerCAmelCase						).manual_seed(__lowerCAmelCase						)
         _lowerCamelCase		:							Optional[Any]            =					np.random.RandomState(__lowerCAmelCase						).standard_normal((1, 4, 6_4, 6_4)						)
         _lowerCamelCase		:							List[Any]            =					torch.from_numpy(__lowerCAmelCase						).to(device=__lowerCAmelCase		,   dtype=__lowerCAmelCase						)
         _lowerCamelCase		:							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 		SCREAMING_SNAKE_CASE							(					self	:		Optional[int]						):






         """simple docstring"""






         _lowerCamelCase		:							List[str]            =					DiffusionPipeline.from_pretrained('''stabilityai/stable-diffusion-2-base'''						)
         pipe.to(__lowerCAmelCase						)
         pipe.set_progress_bar_config(disable=__lowerCAmelCase						)

         _lowerCamelCase		:							List[Any]            =					self.get_inputs(__lowerCAmelCase						)
         _lowerCamelCase		:							int            =					pipe(**__lowerCAmelCase						).images
         _lowerCamelCase		:							Tuple            =					image[0, -3:, -3:, -1].flatten()

         assert image.shape == (1, 5_1_2, 5_1_2, 3)
         _lowerCamelCase		:							Optional[int]            =					np.array([0.4_94_93, 0.4_78_96, 0.4_07_98, 0.5_42_14, 0.5_32_12, 0.4_82_02, 0.4_76_56, 0.4_63_29, 0.4_85_06]						)
         assert np.abs(image_slice - expected_slice						).max() < 7E-3




'''simple docstring'''
from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_flax_available,
    is_sentencepiece_available,
    is_tf_available,
    is_tokenizers_available,
    is_torch_available,
)


if is_sentencepiece_available():
						from ..ta.tokenization_ta import TaTokenizer
else:
						from ...utils.dummy_sentencepiece_objects import TaTokenizer

lowerCAmelCase					:    Tuple           = TaTokenizer

if is_tokenizers_available():
						from ..ta.tokenization_ta_fast import TaTokenizerFast
else:
						from ...utils.dummy_tokenizers_objects import TaTokenizerFast

lowerCAmelCase					:    Optional[int]           = TaTokenizerFast

lowerCAmelCase					:    Any           = {'configuration_mt5': ['MT5Config', 'MT5OnnxConfig']}

try:
						if not is_torch_available():
												raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
						pass
else:
						lowerCAmelCase					:    Optional[int]           = [
						    'MT5EncoderModel',
						    'MT5ForConditionalGeneration',
						    'MT5ForQuestionAnswering',
						    'MT5Model',
						    'MT5PreTrainedModel',
						    'MT5Stack',
						]

try:
						if not is_tf_available():
												raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
						pass
else:
						lowerCAmelCase					:    Dict           = ['TFMT5EncoderModel', 'TFMT5ForConditionalGeneration', 'TFMT5Model']

try:
						if not is_flax_available():
												raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
						pass
else:
						lowerCAmelCase					:    Optional[Any]           = ['FlaxMT5EncoderModel', 'FlaxMT5ForConditionalGeneration', 'FlaxMT5Model']


if TYPE_CHECKING:
						from .configuration_mta import MTaConfig, MTaOnnxConfig

						try:
												if not is_torch_available():
																		raise OptionalDependencyNotAvailable()
						except OptionalDependencyNotAvailable:
												pass
						else:
												from .modeling_mta import (
												    MTaEncoderModel,
												    MTaForConditionalGeneration,
												    MTaForQuestionAnswering,
												    MTaModel,
												    MTaPreTrainedModel,
												    MTaStack,
												)

						try:
												if not is_tf_available():
																		raise OptionalDependencyNotAvailable()
						except OptionalDependencyNotAvailable:
												pass
						else:
												from .modeling_tf_mta import TFMTaEncoderModel, TFMTaForConditionalGeneration, TFMTaModel

						try:
												if not is_flax_available():
																		raise OptionalDependencyNotAvailable()
						except OptionalDependencyNotAvailable:
												pass
						else:
												from .modeling_flax_mta import FlaxMTaEncoderModel, FlaxMTaForConditionalGeneration, FlaxMTaModel

else:
						import sys

						lowerCAmelCase					:    Tuple           = _LazyModule(
						    __name__,
						    globals()['__file__'],
						    _import_structure,
						    extra_objects={'MT5Tokenizer': MTaTokenizer, 'MT5TokenizerFast': MTaTokenizerFast},
						    module_spec=__spec__,
						)




