


'''simple docstring'''






import math

import torch
from torch import nn

from ..configuration_utils import ConfigMixin, register_to_config
from .attention_processor import Attention
from .embeddings import get_timestep_embedding
from .modeling_utils import ModelMixin
class 						__UpperCamelCase	(						__a						,  __a							):





							@register_to_config
							def __init__(  self		:Tuple  ,_UpperCamelCase		:Optional[Any] = 1_2_8  ,_UpperCamelCase		:Dict = 2_5_6  ,_UpperCamelCase		:Tuple = 2_0_0_0.0  ,_UpperCamelCase		:str = 7_6_8  ,_UpperCamelCase		:Optional[Any] = 1_2  ,_UpperCamelCase		:str = 1_2  ,_UpperCamelCase		:Any = 6_4  ,_UpperCamelCase		:Optional[Any] = 2_0_4_8  ,_UpperCamelCase		:str = 0.1  ,):
									super().__init__()

									snake_case_		:							List[str]			       =						nn.Sequential(
									    nn.Linear(lowerCAmelCase_  ,d_model * 4  ,bias=lowerCAmelCase_			)  ,nn.SiLU()  ,nn.Linear(d_model * 4  ,d_model * 4  ,bias=lowerCAmelCase_			)  ,nn.SiLU()  ,)

									snake_case_		:							Union[str, Any]			       =						nn.Embedding(lowerCAmelCase_  ,lowerCAmelCase_			)
									snake_case_		:							Optional[int]			       =						False

									snake_case_		:							List[Any]			       =						nn.Linear(lowerCAmelCase_  ,lowerCAmelCase_  ,bias=lowerCAmelCase_			)

									snake_case_		:							List[str]			       =						nn.Dropout(p=lowerCAmelCase_			)

									snake_case_		:							List[str]			       =						nn.ModuleList()
									for lyr_num in range(lowerCAmelCase_			):
											# FiLM conditional T5 decoder
											snake_case_		:							int			       =						DecoderLayer(d_model=lowerCAmelCase_  ,d_kv=lowerCAmelCase_  ,num_heads=lowerCAmelCase_  ,d_ff=lowerCAmelCase_  ,dropout_rate=lowerCAmelCase_			)
											self.decoders.append(lowerCAmelCase_			)

									snake_case_		:							List[Any]			       =						TaLayerNorm(lowerCAmelCase_			)

									snake_case_		:							List[Any]			       =						nn.Dropout(p=lowerCAmelCase_			)
									snake_case_		:							int			       =						nn.Linear(lowerCAmelCase_  ,lowerCAmelCase_  ,bias=lowerCAmelCase_			)





							def 		a__   (  self		:int  ,_UpperCamelCase		:Dict  ,_UpperCamelCase		:int			):
									snake_case_		:							List[str]			       =						torch.mul(query_input.unsqueeze(-1			)  ,key_input.unsqueeze(-2			)			)
									return mask.unsqueeze(-3			)





							def 		a__   (  self		:Any  ,_UpperCamelCase		:Any  ,_UpperCamelCase		:Optional[Any]  ,_UpperCamelCase		:Tuple			):
									snake_case_       ,      snake_case_       ,      snake_case_		:							int			       =						decoder_input_tokens.shape
									assert decoder_noise_time.shape == (batch,)

									# decoder_noise_time is in [0, 1), so rescale to expected timing range.
									snake_case_		:							Union[str, Any]			       =						get_timestep_embedding(
									    decoder_noise_time * self.config.max_decoder_noise_time  ,embedding_dim=self.config.d_model  ,max_period=self.config.max_decoder_noise_time  ,).to(dtype=self.dtype			)

									snake_case_		:							Tuple			       =						self.conditioning_emb(lowerCAmelCase_			).unsqueeze(1			)

									assert conditioning_emb.shape == (batch, 1, self.config.d_model * 4)

									snake_case_		:							Union[str, Any]			       =						decoder_input_tokens.shape[1]

									# If we want to use relative positions for audio context, we can just offset
									# this sequence by the length of encodings_and_masks.
									snake_case_		:							Tuple			       =						torch.broadcast_to(
									    torch.arange(lowerCAmelCase_  ,device=decoder_input_tokens.device			)  ,(batch, seq_length)  ,)

									snake_case_		:							Any			       =						self.position_encoding(lowerCAmelCase_			)

									snake_case_		:							Union[str, Any]			       =						self.continuous_inputs_projection(lowerCAmelCase_			)
									inputs += position_encodings
									snake_case_		:							str			       =						self.dropout(lowerCAmelCase_			)

									# decoder: No padding present.
									snake_case_		:							Optional[int]			       =						torch.ones(
									    decoder_input_tokens.shape[:2]  ,device=decoder_input_tokens.device  ,dtype=inputs.dtype			)

									# Translate encoding masks to encoder-decoder masks.
									snake_case_		:							Optional[Any]			       =						[(x, self.encoder_decoder_mask(lowerCAmelCase_  ,lowerCAmelCase_			)) for x, y in encodings_and_masks]

									# cross attend style: concat encodings
									snake_case_		:							int			       =						torch.cat([x[0] for x in encodings_and_encdec_masks]  ,dim=1			)
									snake_case_		:							Tuple			       =						torch.cat([x[1] for x in encodings_and_encdec_masks]  ,dim=-1			)

									for lyr in self.decoders:
											snake_case_		:							List[str]			       =						lyr(
											    lowerCAmelCase_  ,conditioning_emb=lowerCAmelCase_  ,encoder_hidden_states=lowerCAmelCase_  ,encoder_attention_mask=lowerCAmelCase_  ,)[0]

									snake_case_		:							Tuple			       =						self.decoder_norm(lowerCAmelCase_			)
									snake_case_		:							str			       =						self.post_dropout(lowerCAmelCase_			)

									snake_case_		:							Tuple			       =						self.spec_out(lowerCAmelCase_			)
									return spec_out





class 						__UpperCamelCase	(						nn.Module							):





							def __init__(  self		:str  ,_UpperCamelCase		:Dict  ,_UpperCamelCase		:Union[str, Any]  ,_UpperCamelCase		:str  ,_UpperCamelCase		:Any  ,_UpperCamelCase		:List[Any]  ,_UpperCamelCase		:List[str]=1E-6			):
									super().__init__()
									snake_case_		:							List[Any]			       =						nn.ModuleList()

									# cond self attention: layer 0
									self.layer.append(
									    TaLayerSelfAttentionCond(d_model=lowerCAmelCase_  ,d_kv=lowerCAmelCase_  ,num_heads=lowerCAmelCase_  ,dropout_rate=lowerCAmelCase_			)			)

									# cross attention: layer 1
									self.layer.append(
									    TaLayerCrossAttention(
									        d_model=lowerCAmelCase_  ,d_kv=lowerCAmelCase_  ,num_heads=lowerCAmelCase_  ,dropout_rate=lowerCAmelCase_  ,layer_norm_epsilon=lowerCAmelCase_  ,)			)

									# Film Cond MLP + dropout: last layer
									self.layer.append(
									    TaLayerFFCond(d_model=lowerCAmelCase_  ,d_ff=lowerCAmelCase_  ,dropout_rate=lowerCAmelCase_  ,layer_norm_epsilon=lowerCAmelCase_			)			)





							def 		a__   (  self		:Dict  ,_UpperCamelCase		:Union[str, Any]  ,_UpperCamelCase		:Tuple=None  ,_UpperCamelCase		:List[Any]=None  ,_UpperCamelCase		:List[str]=None  ,_UpperCamelCase		:Any=None  ,_UpperCamelCase		:Union[str, Any]=None  ,):
									snake_case_		:							List[str]			       =						self.layer[0](
									    lowerCAmelCase_  ,conditioning_emb=lowerCAmelCase_  ,attention_mask=lowerCAmelCase_  ,)

									if encoder_hidden_states is not None:
											snake_case_		:							Optional[int]			       =						torch.where(encoder_attention_mask > 0  ,0  ,-1E1_0			).to(
											    encoder_hidden_states.dtype			)

											snake_case_		:							Union[str, Any]			       =						self.layer[1](
											    lowerCAmelCase_  ,key_value_states=lowerCAmelCase_  ,attention_mask=lowerCAmelCase_  ,)

									# Apply Film Conditional Feed Forward layer
									snake_case_		:							List[str]			       =						self.layer[-1](lowerCAmelCase_  ,lowerCAmelCase_			)

									return (hidden_states,)





class 						__UpperCamelCase	(						nn.Module							):





							def __init__(  self		:Optional[Any]  ,_UpperCamelCase		:Optional[int]  ,_UpperCamelCase		:Optional[Any]  ,_UpperCamelCase		:List[str]  ,_UpperCamelCase		:Union[str, Any]			):
									super().__init__()
									snake_case_		:							Dict			       =						TaLayerNorm(lowerCAmelCase_			)
									snake_case_		:							List[Any]			       =						TaFiLMLayer(in_features=d_model * 4  ,out_features=lowerCAmelCase_			)
									snake_case_		:							Optional[Any]			       =						Attention(query_dim=lowerCAmelCase_  ,heads=lowerCAmelCase_  ,dim_head=lowerCAmelCase_  ,out_bias=lowerCAmelCase_  ,scale_qk=lowerCAmelCase_			)
									snake_case_		:							Optional[int]			       =						nn.Dropout(lowerCAmelCase_			)





							def 		a__   (  self		:int  ,_UpperCamelCase		:str  ,_UpperCamelCase		:Dict=None  ,_UpperCamelCase		:Dict=None  ,):
									snake_case_		:							Optional[int]			       =						self.layer_norm(lowerCAmelCase_			)

									if conditioning_emb is not None:
											snake_case_		:							List[str]			       =						self.FiLMLayer(lowerCAmelCase_  ,lowerCAmelCase_			)

									# Self-attention block
									snake_case_		:							List[Any]			       =						self.attention(lowerCAmelCase_			)

									snake_case_		:							List[Any]			       =						hidden_states + self.dropout(lowerCAmelCase_			)

									return hidden_states





class 						__UpperCamelCase	(						nn.Module							):





							def __init__(  self		:Union[str, Any]  ,_UpperCamelCase		:Optional[Any]  ,_UpperCamelCase		:Optional[Any]  ,_UpperCamelCase		:Dict  ,_UpperCamelCase		:Union[str, Any]  ,_UpperCamelCase		:int			):
									super().__init__()
									snake_case_		:							int			       =						Attention(query_dim=lowerCAmelCase_  ,heads=lowerCAmelCase_  ,dim_head=lowerCAmelCase_  ,out_bias=lowerCAmelCase_  ,scale_qk=lowerCAmelCase_			)
									snake_case_		:							Dict			       =						TaLayerNorm(lowerCAmelCase_  ,eps=lowerCAmelCase_			)
									snake_case_		:							Optional[int]			       =						nn.Dropout(lowerCAmelCase_			)





							def 		a__   (  self		:Dict  ,_UpperCamelCase		:int  ,_UpperCamelCase		:List[str]=None  ,_UpperCamelCase		:List[Any]=None  ,):
									snake_case_		:							Optional[Any]			       =						self.layer_norm(lowerCAmelCase_			)
									snake_case_		:							Optional[Any]			       =						self.attention(
									    lowerCAmelCase_  ,encoder_hidden_states=lowerCAmelCase_  ,attention_mask=attention_mask.squeeze(1			)  ,)
									snake_case_		:							Optional[Any]			       =						hidden_states + self.dropout(lowerCAmelCase_			)
									return layer_output





class 						__UpperCamelCase	(						nn.Module							):





							def __init__(  self		:List[str]  ,_UpperCamelCase		:Union[str, Any]  ,_UpperCamelCase		:Dict  ,_UpperCamelCase		:Dict  ,_UpperCamelCase		:Optional[int]			):
									super().__init__()
									snake_case_		:							List[Any]			       =						TaDenseGatedActDense(d_model=lowerCAmelCase_  ,d_ff=lowerCAmelCase_  ,dropout_rate=lowerCAmelCase_			)
									snake_case_		:							Dict			       =						TaFiLMLayer(in_features=d_model * 4  ,out_features=lowerCAmelCase_			)
									snake_case_		:							Union[str, Any]			       =						TaLayerNorm(lowerCAmelCase_  ,eps=lowerCAmelCase_			)
									snake_case_		:							List[Any]			       =						nn.Dropout(lowerCAmelCase_			)





							def 		a__   (  self		:Optional[int]  ,_UpperCamelCase		:str  ,_UpperCamelCase		:Any=None			):
									snake_case_		:							Tuple			       =						self.layer_norm(lowerCAmelCase_			)
									if conditioning_emb is not None:
											snake_case_		:							int			       =						self.film(lowerCAmelCase_  ,lowerCAmelCase_			)

									snake_case_		:							Any			       =						self.DenseReluDense(lowerCAmelCase_			)
									snake_case_		:							str			       =						hidden_states + self.dropout(lowerCAmelCase_			)
									return hidden_states





class 						__UpperCamelCase	(						nn.Module							):





							def __init__(  self		:Optional[int]  ,_UpperCamelCase		:List[Any]  ,_UpperCamelCase		:Any  ,_UpperCamelCase		:Optional[Any]			):
									super().__init__()
									snake_case_		:							Any			       =						nn.Linear(lowerCAmelCase_  ,lowerCAmelCase_  ,bias=lowerCAmelCase_			)
									snake_case_		:							List[Any]			       =						nn.Linear(lowerCAmelCase_  ,lowerCAmelCase_  ,bias=lowerCAmelCase_			)
									snake_case_		:							Dict			       =						nn.Linear(lowerCAmelCase_  ,lowerCAmelCase_  ,bias=lowerCAmelCase_			)
									snake_case_		:							List[Any]			       =						nn.Dropout(lowerCAmelCase_			)
									snake_case_		:							Union[str, Any]			       =						NewGELUActivation()





							def 		a__   (  self		:str  ,_UpperCamelCase		:Dict			):
									snake_case_		:							Tuple			       =						self.act(self.wi_a(lowerCAmelCase_			)			)
									snake_case_		:							Optional[Any]			       =						self.wi_a(lowerCAmelCase_			)
									snake_case_		:							Optional[Any]			       =						hidden_gelu * hidden_linear
									snake_case_		:							Optional[int]			       =						self.dropout(lowerCAmelCase_			)

									snake_case_		:							Tuple			       =						self.wo(lowerCAmelCase_			)
									return hidden_states





class 						__UpperCamelCase	(						nn.Module							):





							def __init__(  self		:Optional[Any]  ,_UpperCamelCase		:Optional[int]  ,_UpperCamelCase		:Union[str, Any]=1E-6			):
									super().__init__()
									snake_case_		:							Any			       =						nn.Parameter(torch.ones(lowerCAmelCase_			)			)
									snake_case_		:							Optional[Any]			       =						eps





							def 		a__   (  self		:List[Any]  ,_UpperCamelCase		:Union[str, Any]			):
									snake_case_		:							Optional[Any]			       =						hidden_states.to(torch.floataa			).pow(2			).mean(-1  ,keepdim=lowerCAmelCase_			)
									snake_case_		:							List[Any]			       =						hidden_states * torch.rsqrt(variance + self.variance_epsilon			)

									# convert into half-precision if necessary
									if self.weight.dtype in [torch.floataa, torch.bfloataa]:
											snake_case_		:							Optional[int]			       =						hidden_states.to(self.weight.dtype			)

									return self.weight * hidden_states





class 						__UpperCamelCase	(						nn.Module							):





							def 		a__   (  self		:Any  ,_UpperCamelCase		:Optional[Any]			):
									return 0.5 * input * (1.0 + torch.tanh(math.sqrt(2.0 / math.pi			) * (input + 0.04_47_15 * torch.pow(lowerCAmelCase_  ,3.0			))			))





class 						__UpperCamelCase	(						nn.Module							):





							def __init__(  self		:str  ,_UpperCamelCase		:str  ,_UpperCamelCase		:Optional[int]			):
									super().__init__()
									snake_case_		:							str			       =						nn.Linear(lowerCAmelCase_  ,out_features * 2  ,bias=lowerCAmelCase_			)





							def 		a__   (  self		:int  ,_UpperCamelCase		:List[Any]  ,_UpperCamelCase		:Optional[int]			):
									snake_case_		:							Optional[int]			       =						self.scale_bias(lowerCAmelCase_			)
									snake_case_       ,      snake_case_		:							Union[str, Any]			       =						torch.chunk(lowerCAmelCase_  ,2  ,-1			)
									snake_case_		:							Tuple			       =						x * (1 + scale) + shift
									return x



'''simple docstring'''






import unittest

import numpy as np
import torch
from torch import nn
from transformers import (
    CLIPImageProcessor,
    CLIPTextConfig,
    CLIPTextModelWithProjection,
    CLIPTokenizer,
    CLIPVisionConfig,
    CLIPVisionModelWithProjection,
)

from diffusers import KandinskyVaaPriorPipeline, PriorTransformer, UnCLIPScheduler
from diffusers.utils import torch_device
from diffusers.utils.testing_utils import enable_full_determinism, skip_mps

from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()
class 						__UpperCamelCase	(						lowercase__						,  unittest.TestCase							):
							lowercase			:		List[str]   		=  KandinskyVaaPriorPipeline
							lowercase			:		Optional[Any]   		=  ['prompt']
							lowercase			:		Dict   		=  ['prompt', 'negative_prompt']
							lowercase			:		Dict   		=  [
							    'num_images_per_prompt',
							    'generator',
							    'num_inference_steps',
							    'latents',
							    'negative_prompt',
							    'guidance_scale',
							    'output_type',
							    'return_dict',
							]
							lowercase			:		int   		=  False





							@property
							def 		a__   (  self		:int			):
									return 3_2





							@property
							def 		a__   (  self		:Any			):
									return 3_2





							@property
							def 		a__   (  self		:Union[str, Any]			):
									return self.time_input_dim





							@property
							def 		a__   (  self		:int			):
									return self.time_input_dim * 4





							@property
							def 		a__   (  self		:Dict			):
									return 1_0_0





							@property
							def 		a__   (  self		:List[Any]			):
									snake_case_		:							List[Any]			       =						CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""			)
									return tokenizer





							@property
							def 		a__   (  self		:int			):
									torch.manual_seed(0			)
									snake_case_		:							Optional[Any]			       =						CLIPTextConfig(
									    bos_token_id=0  ,eos_token_id=2  ,hidden_size=self.text_embedder_hidden_size  ,projection_dim=self.text_embedder_hidden_size  ,intermediate_size=3_7  ,layer_norm_eps=1E-0_5  ,num_attention_heads=4  ,num_hidden_layers=5  ,pad_token_id=1  ,vocab_size=1_0_0_0  ,)
									return CLIPTextModelWithProjection(_UpperCamelCase			)





							@property
							def 		a__   (  self		:Dict			):
									torch.manual_seed(0			)

									snake_case_		:							Tuple			       =						{
									    """num_attention_heads""": 2,
									    """attention_head_dim""": 1_2,
									    """embedding_dim""": self.text_embedder_hidden_size,
									    """num_layers""": 1,
									}

									snake_case_		:							int			       =						PriorTransformer(**_UpperCamelCase			)
									# clip_std and clip_mean is initialized to be 0 so PriorTransformer.post_process_latents will always return 0 - set clip_std to be 1 so it won't return 0
									snake_case_		:							Any			       =						nn.Parameter(torch.ones(model.clip_std.shape			)			)
									return model





							@property
							def 		a__   (  self		:Optional[int]			):
									torch.manual_seed(0			)
									snake_case_		:							Any			       =						CLIPVisionConfig(
									    hidden_size=self.text_embedder_hidden_size  ,image_size=2_2_4  ,projection_dim=self.text_embedder_hidden_size  ,intermediate_size=3_7  ,num_attention_heads=4  ,num_channels=3  ,num_hidden_layers=5  ,patch_size=1_4  ,)

									snake_case_		:							Optional[Any]			       =						CLIPVisionModelWithProjection(_UpperCamelCase			)
									return model





							@property
							def 		a__   (  self		:List[Any]			):
									snake_case_		:							Any			       =						CLIPImageProcessor(
									    crop_size=2_2_4  ,do_center_crop=_UpperCamelCase  ,do_normalize=_UpperCamelCase  ,do_resize=_UpperCamelCase  ,image_mean=[0.48_14_54_66, 0.4_57_82_75, 0.40_82_10_73]  ,image_std=[0.26_86_29_54, 0.26_13_02_58, 0.27_57_77_11]  ,resample=3  ,size=2_2_4  ,)

									return image_processor





							def 		a__   (  self		:List[Any]			):
									snake_case_		:							Tuple			       =						self.dummy_prior
									snake_case_		:							Any			       =						self.dummy_image_encoder
									snake_case_		:							Optional[int]			       =						self.dummy_text_encoder
									snake_case_		:							Any			       =						self.dummy_tokenizer
									snake_case_		:							Union[str, Any]			       =						self.dummy_image_processor

									snake_case_		:							Tuple			       =						UnCLIPScheduler(
									    variance_type="""fixed_small_log"""  ,prediction_type="""sample"""  ,num_train_timesteps=1_0_0_0  ,clip_sample=_UpperCamelCase  ,clip_sample_range=10.0  ,)

									snake_case_		:							Tuple			       =						{
									    """prior""": prior,
									    """image_encoder""": image_encoder,
									    """text_encoder""": text_encoder,
									    """tokenizer""": tokenizer,
									    """scheduler""": scheduler,
									    """image_processor""": image_processor,
									}

									return components





							def 		a__   (  self		:Dict  ,_UpperCamelCase		:str  ,_UpperCamelCase		:Union[str, Any]=0			):
									if str(_UpperCamelCase			).startswith("""mps"""			):
											snake_case_		:							Optional[int]			       =						torch.manual_seed(_UpperCamelCase			)
									else:
											snake_case_		:							int			       =						torch.Generator(device=_UpperCamelCase			).manual_seed(_UpperCamelCase			)
									snake_case_		:							Dict			       =						{
									    """prompt""": """horse""",
									    """generator""": generator,
									    """guidance_scale""": 4.0,
									    """num_inference_steps""": 2,
									    """output_type""": """np""",
									}
									return inputs





							def 		a__   (  self		:int			):
									snake_case_		:							Union[str, Any]			       =						"""cpu"""

									snake_case_		:							int			       =						self.get_dummy_components()

									snake_case_		:							int			       =						self.pipeline_class(**_UpperCamelCase			)
									snake_case_		:							Union[str, Any]			       =						pipe.to(_UpperCamelCase			)

									pipe.set_progress_bar_config(disable=_UpperCamelCase			)

									snake_case_		:							Any			       =						pipe(**self.get_dummy_inputs(_UpperCamelCase			)			)
									snake_case_		:							List[Any]			       =						output.image_embeds

									snake_case_		:							List[Any]			       =						pipe(
									    **self.get_dummy_inputs(_UpperCamelCase			)  ,return_dict=_UpperCamelCase  ,)[0]

									snake_case_		:							Dict			       =						image[0, -1_0:]
									snake_case_		:							Optional[Any]			       =						image_from_tuple[0, -1_0:]

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

									snake_case_		:							Optional[int]			       =						np.array(
									    [-0.05_32, 1.71_20, 0.36_56, -1.08_52, -0.89_46, -1.17_56, 0.43_48, 0.24_82, 0.51_46, -0.11_56]			)

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





							@skip_mps
							def 		a__   (  self		:List[Any]			):
									snake_case_		:							Any			       =						torch_device == """cpu"""
									snake_case_		:							Any			       =						True
									snake_case_		:							Dict			       =						False

									self._test_inference_batch_single_identical(
									    test_max_difference=_UpperCamelCase  ,relax_max_difference=_UpperCamelCase  ,test_mean_pixel_difference=_UpperCamelCase  ,)





							@skip_mps
							def 		a__   (  self		:str			):
									snake_case_		:							str			       =						torch_device == """cpu"""
									snake_case_		:							List[str]			       =						False

									self._test_attention_slicing_forward_pass(
									    test_max_difference=_UpperCamelCase  ,test_mean_pixel_difference=_UpperCamelCase  ,)