





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 			lowercase				(							a_     ):



						"""simple docstring"""

						_UpperCamelCase  :	torch.FloatTensor



class 			lowercase				(							a_		,			a_     ):



						"""simple docstring"""

						@register_to_config
						def __init__(    self			:		int ,				lowerCamelCase_			:		int = 32 ,				lowerCamelCase_			:		int = 64 ,				lowerCamelCase_			:		int = 20 ,				lowerCamelCase_			:		int = 7_68 ,				lowerCamelCase_			:		str=77 ,				lowerCamelCase_			:		List[Any]=4 ,				lowerCamelCase_			:		float = 0.0 ,				lowerCamelCase_			:		str = "silu" ,				lowerCamelCase_			:		Optional[str] = None ,				lowerCamelCase_			:		Optional[str] = None ,				lowerCamelCase_			:		Optional[str] = "linear" ,				lowerCamelCase_			:		Optional[str] = "prd" ,				lowerCamelCase_			:		Optional[int] = None ,				lowerCamelCase_			:		Optional[int] = None ,				lowerCamelCase_			:		Optional[int] = None ,				):





										'''simple docstring'''







										super().__init__()
										_snake_case :			Dict  						=   num_attention_heads
										_snake_case :			Optional[Any]  						=   attention_head_dim
										_snake_case :			Tuple  						=   num_attention_heads * attention_head_dim
										_snake_case :			Any  						=   additional_embeddings

										_snake_case :			Union[str, Any]  						=   time_embed_dim or inner_dim
										_snake_case :			Any  						=   embedding_proj_dim or embedding_dim
										_snake_case :			int  						=   clip_embed_dim or embedding_dim

										_snake_case :			Tuple  						=   Timesteps(lowerCamelCase_ ,				lowerCamelCase_ ,				0					)
										_snake_case :			int  						=   TimestepEmbedding(lowerCamelCase_ ,				lowerCamelCase_ ,				out_dim=lowerCamelCase_ ,				act_fn=lowerCamelCase_					)

										_snake_case :			Any  						=   nn.Linear(lowerCamelCase_ ,				lowerCamelCase_					)

										if embedding_proj_norm_type is None:
														_snake_case :			str  						=   None
										elif embedding_proj_norm_type == "layer":
														_snake_case :			List[Any]  						=   nn.LayerNorm(lowerCamelCase_					)
										else:
														raise ValueError(f'''unsupported embedding_proj_norm_type: {embedding_proj_norm_type}'''					)

										_snake_case :			Any  						=   nn.Linear(lowerCamelCase_ ,				lowerCamelCase_					)

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

										_snake_case :			List[Any]  						=   nn.Parameter(torch.zeros(1 ,				num_embeddings + additional_embeddings ,				lowerCamelCase_					)					)

										if added_emb_type == "prd":
														_snake_case :			Any  						=   nn.Parameter(torch.zeros(1 ,				1 ,				lowerCamelCase_					)					)
										elif added_emb_type is None:
														_snake_case :			Optional[Any]  						=   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 :			Any  						=   nn.ModuleList(
										    [
										        BasicTransformerBlock(
										            lowerCamelCase_ ,				lowerCamelCase_ ,				lowerCamelCase_ ,				dropout=lowerCamelCase_ ,				activation_fn='gelu' ,				attention_bias=lowerCamelCase_ ,				)
										        for d in range(lowerCamelCase_					)
										    ]					)

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

										_snake_case :			Optional[int]  						=   nn.LayerNorm(lowerCamelCase_					)

										_snake_case :			int  						=   nn.Linear(lowerCamelCase_ ,				lowerCamelCase_					)

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

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







						@property
						# Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
						def 		__UpperCAmelCase       (    self			:		List[str]					):





										'''simple docstring'''







										_snake_case :			Any  						=   {}

										def fn_recursive_add_processors(lowerCamelCase_			:		str ,				lowerCamelCase_			:		torch.nn.Module ,				lowerCamelCase_			:		Dict[str, AttentionProcessor]					):
														if hasattr(lowerCamelCase_ ,				'set_processor'					):
																		_snake_case :			Dict  						=   module.processor

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

														return processors

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

										return processors







						def 		__UpperCAmelCase       (    self			:		List[Any] ,				lowerCamelCase_			:		Union[AttentionProcessor, Dict[str, AttentionProcessor]]					):





										'''simple docstring'''







										_snake_case :			Dict  						=   len(self.attn_processors.keys()					)

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

										def fn_recursive_attn_processor(lowerCamelCase_			:		str ,				lowerCamelCase_			:		torch.nn.Module ,				lowerCamelCase_			:		List[Any]					):
														if hasattr(lowerCamelCase_ ,				'set_processor'					):
																		if not isinstance(lowerCamelCase_ ,				lowerCamelCase_					):
																						module.set_processor(lowerCamelCase_					)
																		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}''' ,				lowerCamelCase_ ,				lowerCamelCase_					)

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







						def 		__UpperCAmelCase       (    self			:		Dict					):





										'''simple docstring'''







										self.set_attn_processor(AttnProcessor()					)







						def 		__UpperCAmelCase       (    self			:		Optional[Any] ,				lowerCamelCase_			:		Optional[int] ,				lowerCamelCase_			:		Union[torch.Tensor, float, int] ,				lowerCamelCase_			:		torch.FloatTensor ,				lowerCamelCase_			:		Optional[torch.FloatTensor] = None ,				lowerCamelCase_			:		Optional[torch.BoolTensor] = None ,				lowerCamelCase_			:		bool = True ,				):





										'''simple docstring'''







										_snake_case :			Any  						=   hidden_states.shape[0]

										_snake_case :			Tuple  						=   timestep
										if not torch.is_tensor(lowerCamelCase_					):
														_snake_case :			Union[str, Any]  						=   torch.tensor([timesteps] ,				dtype=torch.long ,				device=hidden_states.device					)
										elif torch.is_tensor(lowerCamelCase_					) and len(timesteps.shape					) == 0:
														_snake_case :			List[Any]  						=   timesteps[None].to(hidden_states.device					)

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

										_snake_case :			Any  						=   self.time_proj(lowerCamelCase_					)

										# 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 :			List[str]  						=   timesteps_projected.to(dtype=self.dtype					)
										_snake_case :			int  						=   self.time_embedding(lowerCamelCase_					)

										if self.embedding_proj_norm is not None:
														_snake_case :			Dict  						=   self.embedding_proj_norm(lowerCamelCase_					)

										_snake_case :			str  						=   self.embedding_proj(lowerCamelCase_					)
										if self.encoder_hidden_states_proj is not None and encoder_hidden_states is not None:
														_snake_case :			int  						=   self.encoder_hidden_states_proj(lowerCamelCase_					)
										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(lowerCamelCase_					)

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

										_snake_case :			Optional[int]  						=   []
										_snake_case :			Tuple  						=   0

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

										if len(proj_embeddings.shape					) == 2:
														_snake_case :			int  						=   proj_embeddings[:, None, :]

										if len(hidden_states.shape					) == 2:
														_snake_case :			Optional[int]  						=   hidden_states[:, None, :]

										_snake_case :			Tuple  						=   additional_embeds + [
										    proj_embeddings,
										    time_embeddings[:, None, :],
										    hidden_states,
										]

										if self.prd_embedding is not None:
														_snake_case :			List[str]  						=   self.prd_embedding.to(hidden_states.dtype					).expand(lowerCamelCase_ ,				-1 ,				-1					)
														additional_embeds.append(lowerCamelCase_					)

										_snake_case :			str  						=   torch.cat(
										    lowerCamelCase_ ,				dim=1 ,				)

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

										_snake_case :			Union[str, Any]  						=   hidden_states + positional_embeddings

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

										if self.norm_in is not None:
														_snake_case :			Optional[Any]  						=   self.norm_in(lowerCamelCase_					)

										for block in self.transformer_blocks:
														_snake_case :			int  						=   block(lowerCamelCase_ ,				attention_mask=lowerCamelCase_					)

										_snake_case :			Dict  						=   self.norm_out(lowerCamelCase_					)

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

										_snake_case :			List[str]  						=   self.proj_to_clip_embeddings(lowerCamelCase_					)

										if not return_dict:
														return (predicted_image_embedding,)

										return PriorTransformerOutput(predicted_image_embedding=lowerCamelCase_					)







						def 		__UpperCAmelCase       (    self			:		List[Any] ,				lowerCamelCase_			:		Optional[int]					):





										'''simple docstring'''







										_snake_case :			Any  						=   (prior_latents * self.clip_std) + self.clip_mean
										return prior_latents













from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_tokenizers_available, is_torch_available


lowercase_		:				Any				=						{
    '''configuration_canine''': ['''CANINE_PRETRAINED_CONFIG_ARCHIVE_MAP''', '''CanineConfig'''],
    '''tokenization_canine''': ['''CanineTokenizer'''],
}

try:
		if not is_torch_available():
				raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
		pass
else:
		lowercase_		:				Optional[Any]				=						[
		    '''CANINE_PRETRAINED_MODEL_ARCHIVE_LIST''',
		    '''CanineForMultipleChoice''',
		    '''CanineForQuestionAnswering''',
		    '''CanineForSequenceClassification''',
		    '''CanineForTokenClassification''',
		    '''CanineLayer''',
		    '''CanineModel''',
		    '''CaninePreTrainedModel''',
		    '''load_tf_weights_in_canine''',
		]


if TYPE_CHECKING:
		from .configuration_canine import CANINE_PRETRAINED_CONFIG_ARCHIVE_MAP, CanineConfig
		from .tokenization_canine import CanineTokenizer

		try:
				if not is_torch_available():
						raise OptionalDependencyNotAvailable()
		except OptionalDependencyNotAvailable:
				pass
		else:
				from .modeling_canine import (
				    CANINE_PRETRAINED_MODEL_ARCHIVE_LIST,
				    CanineForMultipleChoice,
				    CanineForQuestionAnswering,
				    CanineForSequenceClassification,
				    CanineForTokenClassification,
				    CanineLayer,
				    CanineModel,
				    CaninePreTrainedModel,
				    load_tf_weights_in_canine,
				)


else:
		import sys

		lowercase_		:				str				=						_LazyModule(__name__, globals()['''__file__'''], _import_structure, module_spec=__spec__)






