



import gc
import unittest

import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTextModelWithProjection, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DDPMScheduler,
    PriorTransformer,
    StableUnCLIPPipeline,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.stable_unclip_image_normalizer import StableUnCLIPImageNormalizer
from diffusers.utils.testing_utils import enable_full_determinism, load_numpy, require_torch_gpu, slow, torch_device

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,
    assert_mean_pixel_difference,
)


enable_full_determinism()






class     __a			(  SCREAMING_SNAKE_CASE						,     SCREAMING_SNAKE_CASE						,     SCREAMING_SNAKE_CASE						,     unittest.TestCase					):
	SCREAMING_SNAKE_CASE  	=							StableUnCLIPPipeline
	SCREAMING_SNAKE_CASE  	=							TEXT_TO_IMAGE_PARAMS
	SCREAMING_SNAKE_CASE  	=							TEXT_TO_IMAGE_BATCH_PARAMS
	SCREAMING_SNAKE_CASE  	=							TEXT_TO_IMAGE_IMAGE_PARAMS
	SCREAMING_SNAKE_CASE  	=							TEXT_TO_IMAGE_IMAGE_PARAMS

	# TODO(will) Expected attn_bias.stride(1) == 0 to be true, but got false
	SCREAMING_SNAKE_CASE  	=							False



	def 		UpperCamelCase				(    self				:     List[str])->				Optional[Any]:
						__lowerCAmelCase						 =32
						__lowerCAmelCase						 =embedder_hidden_size

						# prior components

						torch.manual_seed(0)
						__lowerCAmelCase						 =CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip""")

						torch.manual_seed(0)
						__lowerCAmelCase						 =CLIPTextModelWithProjection(
						    CLIPTextConfig(
						        bos_token_id=0					,						eos_token_id=2					,						hidden_size=snake_case_					,						projection_dim=snake_case_					,						intermediate_size=37					,						layer_norm_eps=1e-05					,						num_attention_heads=4					,						num_hidden_layers=5					,						pad_token_id=1					,						vocab_size=10_00					,						))

						torch.manual_seed(0)
						__lowerCAmelCase						 =PriorTransformer(
						    num_attention_heads=2					,						attention_head_dim=12					,						embedding_dim=snake_case_					,						num_layers=1					,						)

						torch.manual_seed(0)
						__lowerCAmelCase						 =DDPMScheduler(
						    variance_type="""fixed_small_log"""					,						prediction_type="""sample"""					,						num_train_timesteps=10_00					,						clip_sample=snake_case_					,						clip_sample_range=5.0					,						beta_schedule="""squaredcos_cap_v2"""					,						)

						# regular denoising components

						torch.manual_seed(0)
						__lowerCAmelCase						 =StableUnCLIPImageNormalizer(embedding_dim=snake_case_)
						__lowerCAmelCase						 =DDPMScheduler(beta_schedule="""squaredcos_cap_v2""")

						torch.manual_seed(0)
						__lowerCAmelCase						 =CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip""")

						torch.manual_seed(0)
						__lowerCAmelCase						 =CLIPTextModel(
						    CLIPTextConfig(
						        bos_token_id=0					,						eos_token_id=2					,						hidden_size=snake_case_					,						projection_dim=32					,						intermediate_size=37					,						layer_norm_eps=1e-05					,						num_attention_heads=4					,						num_hidden_layers=5					,						pad_token_id=1					,						vocab_size=10_00					,						))

						torch.manual_seed(0)
						__lowerCAmelCase						 =UNetaDConditionModel(
						    sample_size=32					,						in_channels=4					,						out_channels=4					,						down_block_types=("""CrossAttnDownBlock2D""", """DownBlock2D""")					,						up_block_types=("""UpBlock2D""", """CrossAttnUpBlock2D""")					,						block_out_channels=(32, 64)					,						attention_head_dim=(2, 4)					,						class_embed_type="""projection"""					,						projection_class_embeddings_input_dim=embedder_projection_dim * 2					,						cross_attention_dim=snake_case_					,						layers_per_block=1					,						upcast_attention=snake_case_					,						use_linear_projection=snake_case_					,						)

						torch.manual_seed(0)
						__lowerCAmelCase						 =DDIMScheduler(
						    beta_schedule="""scaled_linear"""					,						beta_start=0.0_0_0_8_5					,						beta_end=0.0_1_2					,						prediction_type="""v_prediction"""					,						set_alpha_to_one=snake_case_					,						steps_offset=1					,						)

						torch.manual_seed(0)
						__lowerCAmelCase						 =AutoencoderKL()

						__lowerCAmelCase						 ={
						    # prior components
						    """prior_tokenizer""": prior_tokenizer,
						    """prior_text_encoder""": prior_text_encoder,
						    """prior""": prior,
						    """prior_scheduler""": prior_scheduler,
						    # image noising components
						    """image_normalizer""": image_normalizer,
						    """image_noising_scheduler""": image_noising_scheduler,
						    # regular denoising components
						    """tokenizer""": tokenizer,
						    """text_encoder""": text_encoder,
						    """unet""": unet,
						    """scheduler""": scheduler,
						    """vae""": vae,
						}

						return components



	def 		UpperCamelCase				(    self				:     Optional[Any]					,						snake_case_				:     List[Any]					,						snake_case_				:     Optional[int]=0)->				Optional[int]:
						if str(snake_case_).startswith("""mps"""):
											__lowerCAmelCase						 =torch.manual_seed(snake_case_)
						else:
											__lowerCAmelCase						 =torch.Generator(device=snake_case_).manual_seed(snake_case_)
						__lowerCAmelCase						 ={
						    """prompt""": """A painting of a squirrel eating a burger""",
						    """generator""": generator,
						    """num_inference_steps""": 2,
						    """prior_num_inference_steps""": 2,
						    """output_type""": """numpy""",
						}
						return inputs



	def 		UpperCamelCase				(    self				:     List[str])->				Optional[Any]:
						__lowerCAmelCase						 =torch_device == """cpu"""

						self._test_attention_slicing_forward_pass(test_max_difference=snake_case_)



	def 		UpperCamelCase				(    self				:     Tuple)->				List[Any]:
						__lowerCAmelCase						 =torch_device in ["""cpu""", """mps"""]

						self._test_inference_batch_single_identical(test_max_difference=snake_case_)






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



	def 		UpperCamelCase				(    self				:     Any)->				Union[str, Any]:
						# clean up the VRAM after each test
						super().tearDown()
						gc.collect()
						torch.cuda.empty_cache()



	def 		UpperCamelCase				(    self				:     str)->				Union[str, Any]:
						__lowerCAmelCase						 =load_numpy(
						    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_l_anime_turtle_fp16.npy""")

						__lowerCAmelCase						 =StableUnCLIPPipeline.from_pretrained("""fusing/stable-unclip-2-1-l"""					,						torch_dtype=torch.floataa)
						pipe.to(snake_case_)
						pipe.set_progress_bar_config(disable=snake_case_)
						# stable unclip will oom when integration tests are run on a V100,
						# so turn on memory savings
						pipe.enable_attention_slicing()
						pipe.enable_sequential_cpu_offload()

						__lowerCAmelCase						 =torch.Generator(device="""cpu""").manual_seed(0)
						__lowerCAmelCase						 =pipe("""anime turle"""					,						generator=snake_case_					,						output_type="""np""")

						__lowerCAmelCase						 =output.images[0]

						assert image.shape == (7_68, 7_68, 3)

						assert_mean_pixel_difference(snake_case_					,						snake_case_)



	def 		UpperCamelCase				(    self				:     Union[str, Any])->				Dict:
						torch.cuda.empty_cache()
						torch.cuda.reset_max_memory_allocated()
						torch.cuda.reset_peak_memory_stats()

						__lowerCAmelCase						 =StableUnCLIPPipeline.from_pretrained("""fusing/stable-unclip-2-1-l"""					,						torch_dtype=torch.floataa)
						__lowerCAmelCase						 =pipe.to(snake_case_)
						pipe.set_progress_bar_config(disable=snake_case_)
						pipe.enable_attention_slicing()
						pipe.enable_sequential_cpu_offload()

						__lowerCAmelCase						 =pipe(
						    """anime turtle"""					,						prior_num_inference_steps=2					,						num_inference_steps=2					,						output_type="""np"""					,						)

						__lowerCAmelCase						 =torch.cuda.max_memory_allocated()
						# make sure that less than 7 GB is allocated
						assert mem_bytes < 7 * 10**9









from collections import OrderedDict
from typing import Any, Mapping, Optional

from ... import PreTrainedTokenizer, TensorType, is_torch_available
from ...configuration_utils import PretrainedConfig
from ...onnx import OnnxConfigWithPast
from ...utils import logging


lowercase_													=      logging.get_logger(__name__)

lowercase_													=      {
    '''EleutherAI/gpt-neo-1.3B''': '''https://huggingface.co/EleutherAI/gpt-neo-1.3B/resolve/main/config.json''',
    # See all GPTNeo models at https://huggingface.co/models?filter=gpt_neo
}






class     __a			(  SCREAMING_SNAKE_CASE					):
	SCREAMING_SNAKE_CASE  	=							"gpt_neo"
	SCREAMING_SNAKE_CASE  	=							["past_key_values"]
	SCREAMING_SNAKE_CASE  	=							{"num_attention_heads": "num_heads", "num_hidden_layers": "num_layers"}



	def __init__(    self				:     Optional[int]					,						snake_case_				:     int=5_02_57					,						snake_case_				:     Tuple=20_48					,						snake_case_				:     Optional[int]=20_48					,						snake_case_				:     int=24					,						snake_case_				:     List[Any]=[[["global", "local"], 12]]					,						snake_case_				:     Tuple=16					,						snake_case_				:     Optional[int]=None					,						snake_case_				:     Dict=2_56					,						snake_case_				:     List[str]="gelu_new"					,						snake_case_				:     List[Any]=0.0					,						snake_case_				:     Tuple=0.0					,						snake_case_				:     Tuple=0.0					,						snake_case_				:     str=0.1					,						snake_case_				:     Union[str, Any]=1e-5					,						snake_case_				:     str=0.0_2					,						snake_case_				:     Union[str, Any]=True					,						snake_case_				:     Optional[int]=5_02_56					,						snake_case_				:     Tuple=5_02_56					,						**snake_case_				:     Optional[int]					,						)->				Tuple:
						__lowerCAmelCase						 =vocab_size
						__lowerCAmelCase						 =max_position_embeddings
						__lowerCAmelCase						 =hidden_size
						__lowerCAmelCase						 =num_layers
						__lowerCAmelCase						 =num_heads
						__lowerCAmelCase						 =intermediate_size
						__lowerCAmelCase						 =window_size
						__lowerCAmelCase						 =activation_function
						__lowerCAmelCase						 =resid_dropout
						__lowerCAmelCase						 =embed_dropout
						__lowerCAmelCase						 =attention_dropout
						__lowerCAmelCase						 =classifier_dropout
						__lowerCAmelCase						 =layer_norm_epsilon
						__lowerCAmelCase						 =initializer_range
						__lowerCAmelCase						 =use_cache

						__lowerCAmelCase						 =bos_token_id
						__lowerCAmelCase						 =eos_token_id

						__lowerCAmelCase						 =attention_types
						__lowerCAmelCase						 =self.expand_attention_types_params(snake_case_)

						if len(self.attention_layers) != self.num_layers:
											raise ValueError(
											    """Configuration for convolutional module is incorrect. """
											    """It is required that `len(config.attention_layers)` == `config.num_layers` """
											    F"""but is `len(config.attention_layers) = {len(self.attention_layers)}`, """
											    F"""`config.num_layers = {self.num_layers}`. """
											    """`config.attention_layers` is prepared using `config.attention_types`. """
											    """Please verify the value of `config.attention_types` argument.""")

						super().__init__(bos_token_id=snake_case_					,						eos_token_id=snake_case_					,						**snake_case_)



	@staticmethod
	def 		UpperCamelCase				(    snake_case_				:     Optional[int])->				Any:
						__lowerCAmelCase						 =[]
						for item in attention_types:
											for _ in range(item[1]):
																attentions.extend(item[0])
						return attentions






def        __lowerCAmelCase		(	__lowerCamelCase						:     Any					, __lowerCamelCase						:     List[str]					, __lowerCamelCase						:     Optional[Any]					, __lowerCamelCase						:     int						)	->    int:
					import torch

					__lowerCAmelCase						 =input.size()
					__lowerCAmelCase						 =len(__lowerCamelCase						)
					__lowerCAmelCase						 =shape[dimension]

					__lowerCAmelCase						 =torch.arange(0					, __lowerCamelCase					, __lowerCamelCase						)
					__lowerCAmelCase						 =torch.div(sizedim - size					, __lowerCamelCase					, rounding_mode="""floor"""						) + 1
					__lowerCAmelCase						 =torch.arange(__lowerCamelCase						) + low_indices[:min_length][:, None]

					__lowerCAmelCase						 =[slice(__lowerCamelCase						)] * rank
					__lowerCAmelCase						 =indices
					__lowerCAmelCase						 =input[s]

					__lowerCAmelCase						 =list(range(0					, rank + 1						)						)
					perm.append(perm.pop(dimension + 1						)						)

					return sliced.permute(__lowerCamelCase						)






def        __lowerCAmelCase		(	__lowerCamelCase						:     Optional[Any]					, __lowerCamelCase						:     Any						)	->    List[str]:
					import torch

					__lowerCAmelCase						 =torch.arange(1					, __lowerCamelCase						)
					__lowerCAmelCase						 =torch.remainder(__lowerCamelCase					, __lowerCamelCase						)
					__lowerCAmelCase						 =remainders == 0
					__lowerCAmelCase						 =candidates[divisor_indices]
					__lowerCAmelCase						 =torch.max(__lowerCamelCase						)
					return largest_divisor, torch.div(__lowerCamelCase					, __lowerCamelCase					, rounding_mode="""floor"""						)






class     __a			(  SCREAMING_SNAKE_CASE					):



	@property
	def 		UpperCamelCase				(    self				:     Union[str, Any])->				Mapping[str, Mapping[int, str]]:
						__lowerCAmelCase						 =OrderedDict({"""input_ids""": {0: """batch""", 1: """sequence"""}})
						if self.use_past:
											self.fill_with_past_key_values_(snake_case_					,						direction="""inputs""")
											__lowerCAmelCase						 ={0: """batch""", 1: """past_sequence + sequence"""}
						else:
											__lowerCAmelCase						 ={0: """batch""", 1: """sequence"""}

						return common_inputs



	@property
	def 		UpperCamelCase				(    self				:     Optional[int])->				int:
						return self._config.num_heads



	def 		UpperCamelCase				(    self				:     int					,						snake_case_				:     PreTrainedTokenizer					,						snake_case_				:     int = -1					,						snake_case_				:     int = -1					,						snake_case_				:     bool = False					,						snake_case_				:     Optional[TensorType] = None					,						)->				Mapping[str, Any]:
						__lowerCAmelCase						 =super(snake_case_					,						self).generate_dummy_inputs(
						    snake_case_					,						batch_size=snake_case_					,						seq_length=snake_case_					,						is_pair=snake_case_					,						framework=snake_case_)

						# We need to order the input in the way they appears in the forward()
						__lowerCAmelCase						 =OrderedDict({"""input_ids""": common_inputs["""input_ids"""]})

						# Need to add the past_keys
						if self.use_past:
											if not is_torch_available():
																raise ValueError("""Cannot generate dummy past_keys inputs without PyTorch installed.""")
											else:
																import torch

																__lowerCAmelCase ,						__lowerCAmelCase						 =common_inputs["""input_ids"""].shape
																# Not using the same length for past_key_values
																__lowerCAmelCase						 =seqlen + 2
																__lowerCAmelCase						 =(
																    batch,
																    self.num_attention_heads,
																    past_key_values_length,
																    self._config.hidden_size // self.num_attention_heads,
																)
																__lowerCAmelCase						 =[
																    (torch.zeros(snake_case_), torch.zeros(snake_case_)) for _ in range(self.num_layers)
																]

						__lowerCAmelCase						 =common_inputs["""attention_mask"""]
						if self.use_past:
											__lowerCAmelCase						 =ordered_inputs["""attention_mask"""].dtype
											__lowerCAmelCase						 =torch.cat(
											    [ordered_inputs["""attention_mask"""], torch.ones(snake_case_					,						snake_case_					,						dtype=snake_case_)]					,						dim=1)

						return ordered_inputs



	@property
	def 		UpperCamelCase				(    self				:     Union[str, Any])->				int:
						return 13




