






from dataclasses import dataclass
from typing import Dict, Optional, Union

import torch
import torch.nn.functional as F
from torch import nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput
from .attention import BasicTransformerBlock
from .attention_processor import AttentionProcessor, AttnProcessor
from .embeddings import TimestepEmbedding, Timesteps
from .modeling_utils import ModelMixin



@dataclass
class 							SCREAMING_SNAKE_CASE__		(       UpperCamelCase_		):





					"""simple docstring"""






					a_			=	42



class 							SCREAMING_SNAKE_CASE__		(       UpperCamelCase_				,     UpperCamelCase_		):





					"""simple docstring"""






					@register_to_config
					def __init__(      self  :						int							,	__A  :						int = 3_2							,	__A  :						int = 6_4							,	__A  :						int = 2_0							,	__A  :						int = 7_6_8							,	__A  :						Dict=7_7							,	__A  :						Any=4							,	__A  :						float = 0.0							,	__A  :						str = "silu"							,	__A  :						Optional[str] = None							,	__A  :						Optional[str] = None							,	__A  :						Optional[str] = "linear"							,	__A  :						Optional[str] = "prd"							,	__A  :						Optional[int] = None							,	__A  :						Optional[int] = None							,	__A  :						Optional[int] = None							,	):
						super().__init__()
						snake_case__				:   Dict     =						num_attention_heads
						snake_case__				:   Any     =						attention_head_dim
						snake_case__				:   Union[str, Any]     =						num_attention_heads * attention_head_dim
						snake_case__				:   Optional[Any]     =						additional_embeddings

						snake_case__				:   Tuple     =						time_embed_dim or inner_dim
						snake_case__				:   Optional[int]     =						embedding_proj_dim or embedding_dim
						snake_case__				:   Dict     =						clip_embed_dim or embedding_dim

						snake_case__				:   List[Any]     =						Timesteps(__A							,	__A							,	0 )
						snake_case__				:   int     =						TimestepEmbedding(__A							,	__A							,	out_dim=__A							,	act_fn=__A )

						snake_case__				:   List[str]     =						nn.Linear(__A							,	__A )

						if embedding_proj_norm_type is None:
							snake_case__				:   int     =						None
						elif embedding_proj_norm_type == "layer":
							snake_case__				:   int     =						nn.LayerNorm(__A )
						else:
							raise ValueError(f'''unsupported embedding_proj_norm_type: {embedding_proj_norm_type}''' )

						snake_case__				:   List[str]     =						nn.Linear(__A							,	__A )

						if encoder_hid_proj_type is None:
							snake_case__				:   str     =						None
						elif encoder_hid_proj_type == "linear":
							snake_case__				:   Optional[Any]     =						nn.Linear(__A							,	__A )
						else:
							raise ValueError(f'''unsupported encoder_hid_proj_type: {encoder_hid_proj_type}''' )

						snake_case__				:   Any     =						nn.Parameter(torch.zeros(1							,	num_embeddings + additional_embeddings							,	__A ) )

						if added_emb_type == "prd":
							snake_case__				:   int     =						nn.Parameter(torch.zeros(1							,	1							,	__A ) )
						elif added_emb_type is None:
							snake_case__				:   Tuple     =						None
						else:
							raise ValueError(
							    f'''`added_emb_type`: {added_emb_type} is not supported. Make sure to choose one of `\'prd\'` or `None`.''' )

						snake_case__				:   List[str]     =						nn.ModuleList(
						    [
						        BasicTransformerBlock(
						            __A							,	__A							,	__A							,	dropout=__A							,	activation_fn="gelu"							,	attention_bias=__A							,	)
						        for d in range(__A )
						    ] )

						if norm_in_type == "layer":
							snake_case__				:   Any     =						nn.LayerNorm(__A )
						elif norm_in_type is None:
							snake_case__				:   str     =						None
						else:
							raise ValueError(f'''Unsupported norm_in_type: {norm_in_type}.''' )

						snake_case__				:   Optional[int]     =						nn.LayerNorm(__A )

						snake_case__				:   Optional[Any]     =						nn.Linear(__A							,	__A )

						snake_case__				:   Optional[Any]     =						torch.full(
						    [num_embeddings + additional_embeddings, num_embeddings + additional_embeddings]							,	-1_0_0_0_0.0 )
						causal_attention_mask.triu_(1 )
						snake_case__				:   int     =						causal_attention_mask[None, ...]
						self.register_buffer("causal_attention_mask"							,	__A							,	persistent=__A )

						snake_case__				:   Tuple     =						nn.Parameter(torch.zeros(1							,	__A ) )
						snake_case__				:   Union[str, Any]     =						nn.Parameter(torch.zeros(1							,	__A ) )



					@property
					# Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
					def  _lowercase  (      self  :						int ):
						snake_case__				:   Tuple     =						{}

						def fn_recursive_add_processors(__A  :						str							,	__A  :						torch.nn.Module							,	__A  :						Dict[str, AttentionProcessor] ):
							if hasattr(__A							,	"set_processor" ):
								snake_case__				:   List[str]     =						module.processor

							for sub_name, child in module.named_children():
								fn_recursive_add_processors(f'''{name}.{sub_name}'''							,	__A							,	__A )

							return processors

						for name, module in self.named_children():
							fn_recursive_add_processors(__A							,	__A							,	__A )

						return processors



					def  _lowercase  (      self  :						Any							,	__A  :						Union[AttentionProcessor, Dict[str, AttentionProcessor]] ):
						snake_case__				:   int     =						len(self.attn_processors.keys() )

						if isinstance(__A							,	__A ) and len(__A ) != count:
							raise ValueError(
							    f'''A dict of processors was passed, but the number of processors {len(__A )} does not match the'''
							    f''' number of attention layers: {count}. Please make sure to pass {count} processor classes.''' )

						def fn_recursive_attn_processor(__A  :						str							,	__A  :						torch.nn.Module							,	__A  :						Dict ):
							if hasattr(__A							,	"set_processor" ):
								if not isinstance(__A							,	__A ):
									module.set_processor(__A )
								else:
									module.set_processor(processor.pop(f'''{name}.processor''' ) )

							for sub_name, child in module.named_children():
								fn_recursive_attn_processor(f'''{name}.{sub_name}'''							,	__A							,	__A )

						for name, module in self.named_children():
							fn_recursive_attn_processor(__A							,	__A							,	__A )



					def  _lowercase  (      self  :						str ):
						self.set_attn_processor(AttnProcessor() )



					def  _lowercase  (      self  :						str							,	__A  :						Any							,	__A  :						Union[torch.Tensor, float, int]							,	__A  :						torch.FloatTensor							,	__A  :						Optional[torch.FloatTensor] = None							,	__A  :						Optional[torch.BoolTensor] = None							,	__A  :						bool = True							,	):
						snake_case__				:   List[str]     =						hidden_states.shape[0]

						snake_case__				:   int     =						timestep
						if not torch.is_tensor(__A ):
							snake_case__				:   Dict     =						torch.tensor([timesteps]							,	dtype=torch.long							,	device=hidden_states.device )
						elif torch.is_tensor(__A ) and len(timesteps.shape ) == 0:
							snake_case__				:   Union[str, Any]     =						timesteps[None].to(hidden_states.device )

						# broadcast to batch dimension in a way that's compatible with ONNX/Core ML
						snake_case__				:   Optional[Any]     =						timesteps * torch.ones(__A							,	dtype=timesteps.dtype							,	device=timesteps.device )

						snake_case__				:   Any     =						self.time_proj(__A )

						# timesteps does not contain any weights and will always return f32 tensors
						# but time_embedding might be fp16, so we need to cast here.
						snake_case__				:   int     =						timesteps_projected.to(dtype=self.dtype )
						snake_case__				:   List[str]     =						self.time_embedding(__A )

						if self.embedding_proj_norm is not None:
							snake_case__				:   List[str]     =						self.embedding_proj_norm(__A )

						snake_case__				:   str     =						self.embedding_proj(__A )
						if self.encoder_hidden_states_proj is not None and encoder_hidden_states is not None:
							snake_case__				:   Any     =						self.encoder_hidden_states_proj(__A )
						elif self.encoder_hidden_states_proj is not None and encoder_hidden_states is None:
							raise ValueError("`encoder_hidden_states_proj` requires `encoder_hidden_states` to be set" )

						snake_case__				:   List[str]     =						self.proj_in(__A )

						snake_case__				:   Union[str, Any]     =						self.positional_embedding.to(hidden_states.dtype )

						snake_case__				:   Optional[int]     =						[]
						snake_case__				:   Dict     =						0

						if encoder_hidden_states is not None:
							additional_embeds.append(__A )
							additional_embeddings_len += encoder_hidden_states.shape[1]

						if len(proj_embeddings.shape ) == 2:
							snake_case__				:   Tuple     =						proj_embeddings[:, None, :]

						if len(hidden_states.shape ) == 2:
							snake_case__				:   Dict     =						hidden_states[:, None, :]

						snake_case__				:   Any     =						additional_embeds + [
						    proj_embeddings,
						    time_embeddings[:, None, :],
						    hidden_states,
						]

						if self.prd_embedding is not None:
							snake_case__				:   Any     =						self.prd_embedding.to(hidden_states.dtype ).expand(__A							,	-1							,	-1 )
							additional_embeds.append(__A )

						snake_case__				:   Dict     =						torch.cat(
						    __A							,	dim=1							,	)

						# Allow positional_embedding to not include the `addtional_embeddings` and instead pad it with zeros for these additional tokens
						snake_case__				:   List[Any]     =						additional_embeddings_len + proj_embeddings.shape[1] + 1
						if positional_embeddings.shape[1] < hidden_states.shape[1]:
							snake_case__				:   Union[str, Any]     =						F.pad(
							    __A							,	(
							        0,
							        0,
							        additional_embeddings_len,
							        self.prd_embedding.shape[1] if self.prd_embedding is not None else 0,
							    )							,	value=0.0							,	)

						snake_case__				:   List[Any]     =						hidden_states + positional_embeddings

						if attention_mask is not None:
							snake_case__				:   Any     =						(1 - attention_mask.to(hidden_states.dtype )) * -1_0_0_0_0.0
							snake_case__				:   Tuple     =						F.pad(__A							,	(0, self.additional_embeddings)							,	value=0.0 )
							snake_case__				:   Dict     =						(attention_mask[:, None, :] + self.causal_attention_mask).to(hidden_states.dtype )
							snake_case__				:   List[str]     =						attention_mask.repeat_interleave(self.config.num_attention_heads							,	dim=0 )

						if self.norm_in is not None:
							snake_case__				:   Optional[int]     =						self.norm_in(__A )

						for block in self.transformer_blocks:
							snake_case__				:   Union[str, Any]     =						block(__A							,	attention_mask=__A )

						snake_case__				:   Optional[int]     =						self.norm_out(__A )

						if self.prd_embedding is not None:
							snake_case__				:   Dict     =						hidden_states[:, -1]
						else:
							snake_case__				:   Optional[int]     =						hidden_states[:, additional_embeddings_len:]

						snake_case__				:   Union[str, Any]     =						self.proj_to_clip_embeddings(__A )

						if not return_dict:
							return (predicted_image_embedding,)

						return PriorTransformerOutput(predicted_image_embedding=__A )






					def  _lowercase  (      self  :						Any							,	__A  :						Optional[int] ):
						snake_case__				:   List[str]     =						(prior_latents * self.clip_std) + self.clip_mean
						return prior_latents











import unittest

from transformers import is_torch_available
from transformers.testing_utils import require_torch


if is_torch_available():
	import torch

	from transformers.activations import gelu_new, gelu_python, get_activation



@require_torch
class 							SCREAMING_SNAKE_CASE__		(       unittest.TestCase		):





					"""simple docstring"""






					def  _lowercase  (      self  :						Tuple ):
						snake_case__				:   List[str]     =						torch.tensor([-1_0_0, -1, -0.1, 0, 0.1, 1.0, 1_0_0] )
						snake_case__				:   Tuple     =						get_activation("gelu" )
						self.assertTrue(torch.allclose(gelu_python(__A )							,	torch_builtin(__A ) ) )
						self.assertFalse(torch.allclose(gelu_python(__A )							,	gelu_new(__A ) ) )



					def  _lowercase  (      self  :						Dict ):
						snake_case__				:   str     =						torch.tensor([-1_0_0, -1, -0.1, 0, 0.1, 1.0, 1_0_0] )
						snake_case__				:   Union[str, Any]     =						get_activation("gelu" )
						snake_case__				:   int     =						get_activation("gelu_10" )

						snake_case__				:   Optional[int]     =						torch_builtin(__A )
						snake_case__				:   Dict     =						geluaa(__A )

						snake_case__				:   Optional[Any]     =						torch.where(y_gelu_aa < 1_0.0							,	1							,	0 )

						self.assertTrue(torch.max(__A ).item() == 1_0.0 )
						self.assertTrue(torch.allclose(y_gelu * clipped_mask							,	y_gelu_aa * clipped_mask ) )



					def  _lowercase  (      self  :						str ):
						get_activation("gelu" )
						get_activation("gelu_10" )
						get_activation("gelu_fast" )
						get_activation("gelu_new" )
						get_activation("gelu_python" )
						get_activation("gelu_pytorch_tanh" )
						get_activation("linear" )
						get_activation("mish" )
						get_activation("quick_gelu" )
						get_activation("relu" )
						get_activation("sigmoid" )
						get_activation("silu" )
						get_activation("swish" )
						get_activation("tanh" )
						with self.assertRaises(__A ):
							get_activation("bogus" )
						with self.assertRaises(__A ):
							get_activation(__A )






					def  _lowercase  (      self  :						List[str] ):
						snake_case__				:   List[str]     =						get_activation("gelu" )
						snake_case__				:   Any     =						1
						snake_case__				:   Union[str, Any]     =						get_activation("gelu" )
						self.assertEqual(acta.a							,	1 )
						with self.assertRaises(__A ):
							snake_case__				:   int     =						acta.a



