


'''simple docstring'''




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  lowerCAmelCase				(				a							):
	_lowerCamelCase   :					torch.FloatTensor



class  lowerCAmelCase				(				a      ,				a							):



	@register_to_config
	def __init__(					self ,					snake_case__ = 32 ,					snake_case__ = 64 ,					snake_case__ = 20 ,					snake_case__ = 768 ,					snake_case__=77 ,					snake_case__=4 ,					snake_case__ = 0.0 ,					snake_case__ = "silu" ,					snake_case__ = None ,					snake_case__ = None ,					snake_case__ = "linear" ,					snake_case__ = "prd" ,					snake_case__ = None ,					snake_case__ = None ,					snake_case__ = None ,					):
		super().__init__()
		lowerCAmelCase					:    List[str]      							= num_attention_heads
		lowerCAmelCase					:    Union[str, Any]      							= attention_head_dim
		lowerCAmelCase					:    Union[str, Any]      							= num_attention_heads * attention_head_dim
		lowerCAmelCase					:    int      							= additional_embeddings

		lowerCAmelCase					:    str      							= time_embed_dim or inner_dim
		lowerCAmelCase					:    Optional[int]      							= embedding_proj_dim or embedding_dim
		lowerCAmelCase					:    Tuple      							= clip_embed_dim or embedding_dim

		lowerCAmelCase					:    Optional[Any]      							= Timesteps(snake_case__ ,					snake_case__ ,					0    )
		lowerCAmelCase					:    Any      							= TimestepEmbedding(snake_case__ ,					snake_case__ ,					out_dim=snake_case__ ,					act_fn=snake_case__    )

		lowerCAmelCase					:    int      							= nn.Linear(snake_case__ ,					snake_case__    )

		if embedding_proj_norm_type is None:
			lowerCAmelCase					:    List[str]      							= None
		elif embedding_proj_norm_type == "layer":
			lowerCAmelCase					:    int      							= nn.LayerNorm(snake_case__    )
		else:
			raise ValueError(f"unsupported embedding_proj_norm_type: {embedding_proj_norm_type}"    )

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

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

		lowerCAmelCase					:    Union[str, Any]      							= nn.Parameter(torch.zeros(1 ,					num_embeddings + additional_embeddings ,					snake_case__    )    )

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

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

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

		lowerCAmelCase					:    Tuple      							= nn.LayerNorm(snake_case__    )

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

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

		lowerCAmelCase					:    Union[str, Any]      							= nn.Parameter(torch.zeros(1 ,					snake_case__    )    )
		lowerCAmelCase					:    Optional[int]      							= nn.Parameter(torch.zeros(1 ,					snake_case__    )    )



	@property
	# Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
	def 				lowercase     (					self    ):
		lowerCAmelCase					:    Optional[int]      							= {}

		def fn_recursive_add_processors(snake_case__ ,					snake_case__ ,					snake_case__    ):
			if hasattr(snake_case__ ,					'set_processor'    ):
				lowerCAmelCase					:    str      							= module.processor

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

			return processors

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

		return processors



	def 				lowercase     (					self ,					snake_case__    ):
		lowerCAmelCase					:    List[str]      							= len(self.attn_processors.keys()    )

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

		def fn_recursive_attn_processor(snake_case__ ,					snake_case__ ,					snake_case__    ):
			if hasattr(snake_case__ ,					'set_processor'    ):
				if not isinstance(snake_case__ ,					snake_case__    ):
					module.set_processor(snake_case__    )
				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}" ,					snake_case__ ,					snake_case__    )

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



	def 				lowercase     (					self    ):
		self.set_attn_processor(AttnProcessor()    )



	def 				lowercase     (					self ,					snake_case__ ,					snake_case__ ,					snake_case__ ,					snake_case__ = None ,					snake_case__ = None ,					snake_case__ = True ,					):
		lowerCAmelCase					:    Any      							= hidden_states.shape[0]

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

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

		lowerCAmelCase					:    Optional[Any]      							= self.time_proj(snake_case__    )

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

		if self.embedding_proj_norm is not None:
			lowerCAmelCase					:    Optional[Any]      							= self.embedding_proj_norm(snake_case__    )

		lowerCAmelCase					:    Any      							= self.embedding_proj(snake_case__    )
		if self.encoder_hidden_states_proj is not None and encoder_hidden_states is not None:
			lowerCAmelCase					:    List[str]      							= self.encoder_hidden_states_proj(snake_case__    )
		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'    )

		lowerCAmelCase					:    Any      							= self.proj_in(snake_case__    )

		lowerCAmelCase					:    List[str]      							= self.positional_embedding.to(hidden_states.dtype    )

		lowerCAmelCase					:    str      							= []
		lowerCAmelCase					:    int      							= 0

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

		if len(proj_embeddings.shape    ) == 2:
			lowerCAmelCase					:    str      							= proj_embeddings[:, None, :]

		if len(hidden_states.shape    ) == 2:
			lowerCAmelCase					:    List[Any]      							= hidden_states[:, None, :]

		lowerCAmelCase					:    int      							= additional_embeds + [
		    proj_embeddings,
		    time_embeddings[:, None, :],
		    hidden_states,
		]

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

		lowerCAmelCase					:    Optional[Any]      							= torch.cat(
		    snake_case__ ,					dim=1 ,					)

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

		lowerCAmelCase					:    List[str]      							= hidden_states + positional_embeddings

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

		if self.norm_in is not None:
			lowerCAmelCase					:    str      							= self.norm_in(snake_case__    )

		for block in self.transformer_blocks:
			lowerCAmelCase					:    Tuple      							= block(snake_case__ ,					attention_mask=snake_case__    )

		lowerCAmelCase					:    str      							= self.norm_out(snake_case__    )

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

		lowerCAmelCase					:    int      							= self.proj_to_clip_embeddings(snake_case__    )

		if not return_dict:
			return (predicted_image_embedding,)

		return PriorTransformerOutput(predicted_image_embedding=snake_case__    )



	def 				lowercase     (					self ,					snake_case__    ):
		lowerCAmelCase					:    Tuple      							= (prior_latents * self.clip_std) + self.clip_mean
		return prior_latents




'''simple docstring'''




def 			__UpperCamelCase    (    _A    :				int				)    ->      bool:




	"""simple docstring"""





	return number & 1 == 0


if __name__ == "__main__":
				import doctest

				doctest.testmod()
